can config es id's constituent part

etc/river.toml

@@ -58,6 +58,8 @@ schema = "test"
 table = "t"
 index = "test"
 type = "t"
+# Default will use the mysql's primary key as es's id, if set id will use the id's column value as id
+id = ["id", "tags"]
 
 # Wildcard table rule, the wildcard table must be in source tables 
 # All tables which match the wildcard format will be synced to ES index `test` and type `t`.
@@ -113,3 +115,4 @@ type = "tfilter"
 
 # Only sync following columns
 filter = ["id", "name"]
+

river/river.go

@@ -212,6 +212,7 @@ func (r *River) prepareRule() error {
 					rr.Index = rule.Index
 					rr.Type = rule.Type
 					rr.Parent = rule.Parent
+					rr.ID = rule.ID
 					rr.FieldMapping = rule.FieldMapping
 				}
 			} else {

river/river_test.go

@@ -81,6 +81,14 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 			Type:         "river",
 			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
 		},
+
+		&Rule{Schema: "test",
+			Table:        "test_river",
+			Index:        "river",
+			Type:         "river_id",
+			ID:           []string{"id", "title"},
+			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
+		},
 
 		&Rule{Schema: "test",
 			Table:        "test_river_[0-9]{4}",
@@ -133,6 +141,18 @@ parent = "pid"
     title = "es_title"
     mylist = "es_mylist,list"
 
+[[rule]]
+schema = "test"
+table = "test_river"
+index = "river"
+type = "river_id"
+parent = "pid"
+id = ["id", "title"]
+
+    [rule.field]
+    title = "es_title"
+    mylist = "es_mylist,list"
+
 [[rule]]
 schema = "test"
 table = "test_river_[0-9]{4}"
@@ -169,9 +189,8 @@ func (s *riverTestSuite) testPrepareData(c *C) {
 	}
 }
 
-func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
+func (s *riverTestSuite) testElasticGet(c *C, docType string, id string) *elastic.Response {
 	index := "river"
-	docType := "river"
 
 	r, err := s.r.es.Get(index, docType, id)
 	c.Assert(err, IsNil)
@@ -204,16 +223,24 @@ func (s *riverTestSuite) TestRiver(c *C) {
 	testWaitSyncDone(c, s.r)
 
 	var r *elastic.Response
-	r = s.testElasticGet(c, "1")
+	r = s.testElasticGet(c, "river", "1")
 	c.Assert(r.Found, Equals, true)
 	c.Assert(r.Source["tenum"], Equals, "e1")
 	c.Assert(r.Source["tset"], Equals, "a,b")
-
-	r = s.testElasticGet(c, "100")
+
+	r = s.testElasticGet(c, "river_id", "1:first")
+	c.Assert(r.Found, Equals, true)
+	c.Assert(r.Source["tenum"], Equals, "e1")
+	c.Assert(r.Source["tset"], Equals, "a,b")
+
+	r = s.testElasticGet(c, "river_id", "1:second")
+	c.Assert(r.Found, Equals, false)
+
+	r = s.testElasticGet(c, "river", "100")
 	c.Assert(r.Found, Equals, false)
 
 	for i := 0; i < 10; i++ {
-		r = s.testElasticGet(c, fmt.Sprintf("%d", 5+i))
+		r = s.testElasticGet(c, "river", fmt.Sprintf("%d", 5+i))
 		c.Assert(r.Found, Equals, true)
 		c.Assert(r.Source["es_title"], Equals, "abc")
 	}
@@ -235,32 +262,32 @@ func (s *riverTestSuite) TestRiver(c *C) {
 
 	testWaitSyncDone(c, s.r)
 
-	r = s.testElasticGet(c, "1")
+	r = s.testElasticGet(c, "river", "1")
 	c.Assert(r.Found, Equals, false)
 
-	r = s.testElasticGet(c, "2")
+	r = s.testElasticGet(c, "river", "2")
 	c.Assert(r.Found, Equals, true)
 	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"})
 	c.Assert(r.Source["tbit"], Equals, float64(1))
 
-	r = s.testElasticGet(c, "4")
+	r = s.testElasticGet(c, "river", "4")
 	c.Assert(r.Found, Equals, true)
 	c.Assert(r.Source["tenum"], Equals, "")
 	c.Assert(r.Source["tset"], Equals, "a,b,c")
 	c.Assert(r.Source["tbit"], Equals, float64(0))
 
-	r = s.testElasticGet(c, "3")
+	r = s.testElasticGet(c, "river", "3")
 	c.Assert(r.Found, Equals, false)
 
-	r = s.testElasticGet(c, "30")
+	r = s.testElasticGet(c, "river", "30")
 	c.Assert(r.Found, Equals, true)
 	c.Assert(r.Source["es_title"], Equals, "second 30")
 
 	for i := 0; i < 10; i++ {
-		r = s.testElasticGet(c, fmt.Sprintf("%d", 5+i))
+		r = s.testElasticGet(c, "river", fmt.Sprintf("%d", 5+i))
 		c.Assert(r.Found, Equals, true)
 		c.Assert(r.Source["es_title"], Equals, "hello")
 	}

river/rule.go

@@ -13,6 +13,7 @@ type Rule struct {
 	Index  string `toml:"index"`
 	Type   string `toml:"type"`
 	Parent string `toml:"parent"`
+	ID []string `toml:"id"`
 
 	// Default, a MySQL table field name is mapped to Elasticsearch field name.
 	// Sometimes, you want to use different name, e.g, the MySQL file name is title,

river/sync.go

@@ -384,20 +384,35 @@ func (r *River) makeUpdateReqData(req *elastic.BulkRequest, rule *Rule,
 	}
 }
 
-// Get primary keys in one row and format them into a string
-// PK must not be nil
+// If id in toml file is none, get primary keys in one row and format them into a string, and PK must not be nil
+// Else get the ID's column in one row and format them into a string
 func (r *River) getDocID(rule *Rule, row []interface{}) (string, error) {
-	pks, err := canal.GetPKValues(rule.TableInfo, row)
-	if err != nil {
-		return "", err
+	var (
+  		ids []interface{}
+  		err error 
+	)
+	if rule.ID == nil {
+		ids, err = canal.GetPKValues(rule.TableInfo, row)
+		if err != nil {
+			return "", err
+		}
+	} else {
+		ids = make([]interface{}, 0, len(rule.ID))
+		for _, column := range rule.ID {
+			value, err := canal.GetColumnValue(rule.TableInfo, column, row)
+			if err != nil {
+				return "", err
+			}
+			ids = append(ids, value)
+		}
 	}
 
 	var buf bytes.Buffer
 
 	sep := ""
-	for i, value := range pks {
+	for i, value := range ids {
 		if value == nil {
-			return "", errors.Errorf("The %ds PK value is nil", i)
+			return "", errors.Errorf("The %ds id or PK value is nil", i)
 		}
 
 		buf.WriteString(fmt.Sprintf("%s%v", sep, value))