add type check for map slice and mejson.M/S

postgres adaptor tests updated with new column for complex table to test the case where field is an array of map[string]interface{}

fixes #314

.gitignore

@@ -2,3 +2,4 @@
 .vagrant
 /transporter.yaml
 /pipeline.js
+.vscode

CHANGELOG.md

@@ -7,6 +7,13 @@
 
 ### Bugfixes
 
+## v0.2.2 [2017-03-20]
+
+### Features
+
+### Bugfixes
+- attempted fix for mejson.S conversion to json column in postgres adaptor [#314](https://github.com/compose/transporter/issues/314)
+
 ## v0.2.1 [2017-03-07]
 
 ### Features

adaptor/postgres/adaptor_test.go

@@ -35,6 +35,7 @@ const (
   colinteger integer,
   colinterval interval,
   coljson json,
+  colarrayjson json,
   coljsonb jsonb,
   colline line,
   collseg lseg,
@@ -70,6 +71,7 @@ var (
 		readerComplexTestData,
 		tailerTestData,
 		writerTestData,
+		writerComplexTestData,
 		writerUpdateTestData,
 		writerDeleteTestData,
 		writerComplexUpdateTestData,
@@ -148,6 +150,7 @@ func setupData(data *TestData) {
 							3,                   -- colinteger integer,
 							DEFAULT,             -- autoset colinterval interval,
 							'{"name": "batman"}',  -- coljson json,
+							'[{"name": "batman"},{"name":"robin"}]',  -- colarrayjson json,
 							'{"name": "alfred"}',  -- coljsonb jsonb,
 							'{1, 1, 3}',         -- colline line,
 							'[(10,10),(25,25)]', -- collseg lseg,
@@ -173,6 +176,7 @@ func setupData(data *TestData) {
 			`, data.Table, i, randomHeros[i%len(randomHeros)])); err != nil {
 				log.Errorf("unexpected Insert error, %s\n", err)
 			}
+			// '[{"name": "batman"}, {"name": "robin"}]',  -- arraycoljson json,
 		} else if data.Schema == basicSchema {
 			if _, err := defaultSession.pqSession.Exec(fmt.Sprintf(`INSERT INTO %s VALUES (
 			  %d,            -- id

adaptor/postgres/writer.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/compose/mejson"
 	"github.com/compose/transporter/client"
 	"github.com/compose/transporter/log"
 	"github.com/compose/transporter/message"
@@ -55,10 +56,12 @@ func insertMsg(m message.Msg, s *sql.DB) error {
 		placeholders = append(placeholders, fmt.Sprintf("$%v", i))
 
 		switch value.(type) {
-		case map[string]interface{}:
+		case map[string]interface{}, mejson.M, []map[string]interface{}, mejson.S:
 			value, _ = json.Marshal(value)
 		case []interface{}:
 			value, _ = json.Marshal(value)
+			value = string(value.([]byte))
+			value = fmt.Sprintf("{%v}", value.(string)[1:len(value.(string))-1])
 		}
 		data = append(data, value)
 
@@ -86,8 +89,12 @@ func deleteMsg(m message.Msg, s *sql.DB) error {
 			ckeys = append(ckeys, fmt.Sprintf("%v = $%v", key, i))
 		}
 		switch value.(type) {
-		case map[string]interface{}:
+		case map[string]interface{}, mejson.M, []map[string]interface{}, mejson.S:
+			value, _ = json.Marshal(value)
+		case []interface{}:
 			value, _ = json.Marshal(value)
+			value = string(value.([]byte))
+			value = fmt.Sprintf("{%v}", value.(string)[1:len(value.(string))-1])
 		}
 		vals = append(vals, value)
 		i = i + 1
@@ -124,7 +131,7 @@ func updateMsg(m message.Msg, s *sql.DB) error {
 		}
 
 		switch value.(type) {
-		case map[string]interface{}:
+		case map[string]interface{}, mejson.M, []map[string]interface{}, mejson.S:
 			value, _ = json.Marshal(value)
 		case []interface{}:
 			value, _ = json.Marshal(value)

adaptor/postgres/writer_test.go

@@ -73,6 +73,83 @@ func TestInsert(t *testing.T) {
 	}
 }
 
+var (
+	writerComplexTestData = &TestData{"writer_complex_insert_test", "complex_test_table", complexSchema, 0}
+)
+
+func TestComplexInsert(t *testing.T) {
+	w := newWriter(writerComplexTestData.DB)
+	for i := 0; i < 10; i++ {
+		msg := message.From(ops.Insert, fmt.Sprintf("public.%s", writerComplexTestData.Table), data.Data{
+			"id":                 i,
+			"colvar":             randomHeros[i],
+			"coltimestamp":       time.Now().UTC(),
+			"colarrayint":        []interface{}{1, 2, 3, 4},
+			"colarraystring":     "{\"one\", \"two\", \"three\", \"four\"}",
+			"colbigint":          int64(4000001240125),
+			"colbit":             "1",
+			"colboolean":         false,
+			"colbox":             "(10,10),(20,20)",
+			"colbytea":           "\\xDEADBEEF",
+			"colcharacter":       "a",
+			"colcidr":            "10.0.1.0/28",
+			"colcircle":          "<(5,10),3>",
+			"coldate":            time.Now().UTC(),
+			"coldoubleprecision": 0.314259892323,
+			"colenum":            "sad",
+			"colinet":            "10.0.1.0",
+			"colinteger":         int64(3),
+			"coljson":            map[string]interface{}{"name": "batman"},
+			"colarrayjson":       []map[string]interface{}{map[string]interface{}{"name": "batman"}, map[string]interface{}{"name": "robin"}},
+			"coljsonb":           map[string]interface{}{"name": "batman"},
+			"colline":            "{1, 1, 3}",
+			"collseg":            "((10,10),(25,25))",
+			"colmacaddr":         "08:00:2b:01:02:03",
+			"colmoney":           "35.68",
+			"colnumeric":         0.23509838,
+			"colpath":            "[(10,10),(20,20),(20,10),(15,15)]",
+			"colpg_lsn":          "0/3000000",
+			"colpoint":           "(15,15)",
+			"colpolygon":         "((10,10),(11, 11),(11,0),(5,5))",
+			"colreal":            7,
+			"colsmallint":        3,
+			"coltext":            "this is \n extremely important",
+			"coltime":            "13:45",
+			"coltsquery":         "'fat':AB & 'cat'",
+			"coltsvector":        "a fat cat sat on a mat and ate a fat rat",
+			"coluuid":            "f0a0da24-4068-4be4-961d-7c295117ccca",
+			"colxml":             "<person><name>Batman</name></person>",
+		})
+		if err := w.Write(msg)(defaultSession); err != nil {
+			t.Errorf("unexpected Insert error, %s\n", err)
+		}
+	}
+	var (
+		id          int
+		stringValue string
+		timeValue   time.Time
+	)
+	if err := defaultSession.pqSession.
+		QueryRow(fmt.Sprintf("SELECT id, colvar, coltimestamp FROM %s WHERE id = 4", writerComplexTestData.Table)).
+		Scan(&id, &stringValue, &timeValue); err != nil {
+		t.Fatalf("Error on test query: %v", err)
+	}
+	if id != 4 || stringValue != randomHeros[4] || timeValue.Before(time.Now().Add(-30*time.Second).UTC()) {
+		t.Fatalf("Values were not what they were expected to be: %v, %v, %v", id, stringValue, timeValue)
+	}
+
+	var count int
+	err := defaultSession.pqSession.
+		QueryRow(fmt.Sprintf("SELECT COUNT(id) FROM %s;", writerComplexTestData.Table)).
+		Scan(&count)
+	if err != nil {
+		t.Errorf("unable to count table, %s", err)
+	}
+	if count != 10 {
+		t.Errorf("wrong document count, expected 10, got %d", count)
+	}
+}
+
 var (
 	writerUpdateTestData = &TestData{"writer_update_test", "update_test_table", basicSchema, 0}
 )
@@ -148,6 +225,7 @@ func TestComplexUpdate(t *testing.T) {
 		"colinet":            "10.0.1.0",
 		"colinteger":         int64(3),
 		"coljson":            map[string]interface{}{"name": "batman"},
+		"colarrayjson":       []map[string]interface{}{map[string]interface{}{"name": "batman"}, map[string]interface{}{"name": "robin"}},
 		"coljsonb":           map[string]interface{}{"name": "batman"},
 		"colline":            "{1, 1, 3}",
 		"collseg":            "((10,10),(25,25))",