From c14ed724ff9f3b4f097fe006bf3da9b025ae1837 Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Wed, 13 Feb 2019 15:59:46 +0800
Subject: [PATCH] =?UTF-8?q?update=E4=B8=AD=E5=A2=9E=E5=8A=A0upsert?=
 =?UTF-8?q?=E5=88=A4=E6=96=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../client/adapter/es/support/ESTemplate.java | 54 ++++++++++---------
 1 file changed, 29 insertions(+), 25 deletions(-)

diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
index 55ba4237..5c389fae 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
@@ -100,34 +100,12 @@ public class ESTemplate {
      * @return
      */
     public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        append4Update(getBulk(), mapping, pkVal, esFieldData);
+        append4Update(mapping, pkVal, esFieldData);
         commitBatch();
     }
 
-    private void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal,
-                               Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
-            bulkRequestBuilder
-                .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
-                    .setDoc(esFieldData));
-        } else {
-            // TODO SearchResponse response =
-            // transportClient.prepareSearch(mapping.get_index())
-            // .setTypes(mapping.get_type())
-            // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
-            // .setSize(MAX_BATCH_SIZE)
-            // .get();
-            // for (SearchHit hit : response.getHits()) {
-            // bulkRequestBuilder
-            // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(),
-            // hit.getId())
-            // .setDoc(esFieldData));
-            // }
-        }
-    }
-
     /**
-     * TODO XXX update by query
+     * update by query
      *
      * @param config
      * @param paramsTmp
@@ -154,7 +132,7 @@ public class ESTemplate {
             try {
                 while (rs.next()) {
                     Object idVal = getIdValFromRS(mapping, rs);
-                    append4Update(getBulk(), mapping, idVal, esFieldData);
+                    append4Update(mapping, idVal, esFieldData);
                     commitBatch();
                     count++;
                 }
@@ -227,6 +205,32 @@ public class ESTemplate {
         }
     }
 
+    private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
+        if (mapping.get_id() != null) {
+            if (mapping.isUpsert()) {
+                getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
+                        .setDoc(esFieldData)
+                        .setDocAsUpsert(true));
+            } else {
+                getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
+                        .setDoc(esFieldData));
+            }
+        } else {
+            // TODO SearchResponse response =
+            // transportClient.prepareSearch(mapping.get_index())
+            // .setTypes(mapping.get_type())
+            // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
+            // .setSize(MAX_BATCH_SIZE)
+            // .get();
+            // for (SearchHit hit : response.getHits()) {
+            // bulkRequestBuilder
+            // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(),
+            // hit.getId())
+            // .setDoc(esFieldData));
+            // }
+        }
+    }
+
     public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
                                String columnName) throws SQLException {
         String esType = getEsType(mapping, fieldName);