Merge pull request go-mysql-org#32 from adriacidre/master

Map mysql fields to elasticsearch arrays

Author: siddontang

README.md

@@ -65,12 +65,38 @@ index = "t"
 type = "t"
 parent = "parent_id"
 
-    [rule.field]
-    title = "my_title"
+    [[rule.fields]]
+    mysql = "title"
+    elastic = "my_title"
 ```
 
 In the example above, we will use a new index and type both named "t" instead of default "t1", and use "my_title" instead of field name "title".
 
+## Rule field types
+
+In order to map a mysql column on different elasticsearch types you can define the field type as follows:
+
+```
+[[rule]]
+schema = "test"
+table = "t1"
+index = "t"
+type = "t"
+parent = "parent_id"
+
+    [rule.field]
+    // This will map column title to elastic search my_title
+    title="my_title"
+
+    // This will map column title to elastic search my_title and use array type
+    title="my_title,list"
+
+    // This will map column title to elastic search title and use array type
+    title=",list"
+```
+
+Modifier "list" will translates a mysql string field like "a,b,c" on an elastic array type '{"a", "b", "c"}' this is specially useful if you need to use those fields on filtering on elasticsearch.
+
 ## Wildcard table
 
 go-mysql-elasticsearch only allows you determind which table to be synced, but sometimes, if you split a big table into multi sub tables, like 1024, table_0000, table_0001, ... table_1023, it is very hard to write rules for every table.

etc/river.toml

@@ -40,7 +40,12 @@ type = "river"
 
     # title is MySQL test_river field name, es_title is the customized name in Elasticsearch
     [rule.field]
-    title = "es_title"
+    # This will map column title to elastic search my_title
+    title="es_title"
+    # This will map column tags to elastic search my_tags and use array type
+    tags="my_tags,list"
+    # This will map column keywords to elastic search keywords and use array type
+    keywords=",list"
 
 # wildcard table rule, the wildcard table must be in source tables 
 [[rule]]
@@ -50,7 +55,8 @@ index = "river"
 type = "river"
 
     # title is MySQL test_river field name, es_title is the customized name in Elasticsearch
-    [rule.field]
-    title = "es_title"
+    [[rule.fields]]
+    mysql = "title"
+    elastic = "es_title"
 
 

river/river_test.go

@@ -37,6 +37,7 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
             id INT,
             title VARCHAR(256),
             content VARCHAR(256),
+            mylist VARCHAR(256),
             tenum ENUM("e1", "e2", "e3"),
             tset SET("a", "b", "c"),
             PRIMARY KEY(id)) ENGINE=INNODB;
@@ -74,13 +75,16 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 			Table:        "test_river",
 			Index:        "river",
 			Type:         "river",
-			FieldMapping: map[string]string{"title": "es_title"}},
+			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
+		},
 
 		&Rule{Schema: "test",
 			Table:        "test_river_[0-9]{4}",
 			Index:        "river",
 			Type:         "river",
-			FieldMapping: map[string]string{"title": "es_title"}}}
+			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
+		},
+	}
 
 	s.r, err = NewRiver(cfg)
 	c.Assert(err, IsNil)
@@ -123,6 +127,7 @@ parent = "pid"
 
     [rule.field]
     title = "es_title"
+    mylist = "es_mylist,list"
 
 [[rule]]
 schema = "test"
@@ -132,6 +137,7 @@ type = "river"
 
     [rule.field]
     title = "es_title"
+    mylist = "es_mylist,list"
 
 `
 
@@ -196,7 +202,7 @@ func (s *riverTestSuite) TestRiver(c *C) {
 		c.Assert(r.Source["es_title"], Equals, "abc")
 	}
 
-	s.testExecute(c, "UPDATE test_river SET title = ?, tenum = ?, tset = ? WHERE id = ?", "second 2", "e3", "a,b,c", 2)
+	s.testExecute(c, "UPDATE test_river SET title = ?, tenum = ?, tset = ?, mylist = ? WHERE id = ?", "second 2", "e3", "a,b,c", "a,b,c", 2)
 	s.testExecute(c, "DELETE FROM test_river WHERE id = ?", 1)
 	s.testExecute(c, "UPDATE test_river SET title = ?, id = ? WHERE id = ?", "second 30", 30, 3)
 
@@ -221,6 +227,7 @@ func (s *riverTestSuite) TestRiver(c *C) {
 	c.Assert(r.Source["es_title"], Equals, "second 2")
 	c.Assert(r.Source["tenum"], Equals, "e3")
 	c.Assert(r.Source["tset"], Equals, "a,b,c")
+	c.Assert(r.Source["es_mylist"], DeepEquals, []interface{}{"a", "b", "c"})
 
 	r = s.testElasticGet(c, "4")
 	c.Assert(r.Found, Equals, true)

river/sync.go

@@ -18,6 +18,10 @@ const (
 	syncUpdateDoc
 )
 
+const (
+	fieldTypeList = "list"
+)
+
 type rowsEventHandler struct {
 	r *River
 }
@@ -187,15 +191,46 @@ func (r *River) makeReqColumnData(col *schema.TableColumn, value interface{}) in
 	return value
 }
 
+func (r *River) getFieldParts(k string, v string) (string, string, string) {
+	composedField := strings.Split(v, ",")
+
+	mysql := k
+	elastic := composedField[0]
+	fieldType := ""
+
+	if 0 == len(elastic) {
+		elastic = mysql
+	}
+	if 2 == len(composedField) {
+		fieldType = composedField[1]
+	}
+
+	return mysql, elastic, fieldType
+}
+
 func (r *River) makeInsertReqData(req *elastic.BulkRequest, rule *Rule, values []interface{}) {
 	req.Data = make(map[string]interface{}, len(values))
 	req.Action = elastic.ActionIndex
 
 	for i, c := range rule.TableInfo.Columns {
-		if name, ok := rule.FieldMapping[c.Name]; ok {
-			// has custom field mapping
-			req.Data[name] = r.makeReqColumnData(&c, values[i])
-		} else {
+		mapped := false
+		for k, v := range rule.FieldMapping {
+			mysql, elastic, fieldType := r.getFieldParts(k, v)
+			if mysql == c.Name {
+				mapped = true
+				v := r.makeReqColumnData(&c, values[i])
+				if fieldType == fieldTypeList {
+					if str, ok := v.(string); ok {
+						req.Data[elastic] = strings.Split(str, ",")
+					} else {
+						req.Data[elastic] = v
+					}
+				} else {
+					req.Data[elastic] = v
+				}
+			}
+		}
+		if mapped == false {
 			req.Data[c.Name] = r.makeReqColumnData(&c, values[i])
 		}
 	}
@@ -209,16 +244,33 @@ func (r *River) makeUpdateReqData(req *elastic.BulkRequest, rule *Rule,
 	req.Action = elastic.ActionUpdate
 
 	for i, c := range rule.TableInfo.Columns {
+		mapped := false
 		if reflect.DeepEqual(beforeValues[i], afterValues[i]) {
 			//nothing changed
 			continue
 		}
-		if name, ok := rule.FieldMapping[c.Name]; ok {
-			// has custom field mapping
-			req.Data[name] = r.makeReqColumnData(&c, afterValues[i])
-		} else {
+		for k, v := range rule.FieldMapping {
+			mysql, elastic, fieldType := r.getFieldParts(k, v)
+			if mysql == c.Name {
+				mapped = true
+				// has custom field mapping
+				v := r.makeReqColumnData(&c, afterValues[i])
+				str, ok := v.(string)
+				if ok == false {
+					req.Data[c.Name] = v
+				} else {
+					if fieldType == fieldTypeList {
+						req.Data[elastic] = strings.Split(str, ",")
+					} else {
+						req.Data[elastic] = str
+					}
+				}
+			}
+		}
+		if mapped == false {
 			req.Data[c.Name] = r.makeReqColumnData(&c, afterValues[i])
 		}
+
 	}
 }