Support docAsUpsert as option

Author: dqduc

src/main/java/org/xbib/elasticsearch/common/util/PlainKeyValueStreamListener.java

@@ -37,7 +37,7 @@
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 
 public class PlainKeyValueStreamListener<K, V> implements KeyValueStreamListener<K, V> {
-    private final static String ValueSeperators = ",|;~!@#$%^&";
+    private final static String ValueSeperators = ",;|/~!@#$%^&";
     private final static Pattern p = Pattern.compile("^(.*)\\[(.*?)\\]$");
 
     /**
@@ -291,6 +291,7 @@ protected Map<String, Object> merge(Map<String, Object> map, Object k, Object va
             if (index == null || index.isEmpty()) {
                 map.put(head, new Values(map.get(head), value, isSequence));
             } else if (index.length() == 1 && ValueSeperators.contains(index)) {
+				// Use difference seperator, pass it as Pattern.quote(index) since Values use String.split
                 map.put(head, new Values(map.get(head), value, isSequence, Pattern.quote(index)));
             } else {
                 if (!map.containsKey(head)) {

src/main/java/org/xbib/elasticsearch/jdbc/strategy/standard/StandardSink.java

@@ -67,6 +67,8 @@ public class StandardSink<C extends StandardContext> implements Sink<C> {
     protected String type;
 
     protected String id;
+	
+	protected boolean docAsUpsert;
 
     private final static SinkMetric sinkMetric = new SinkMetric().start();
 
@@ -96,6 +98,7 @@ public synchronized void beforeFetch() throws IOException {
         Settings settings = context.getSettings();
         String index = settings.get("index", "jdbc");
         String type = settings.get("type", "jdbc");
+		this.docAsUpsert = settings.getAsBoolean("docAsUpsert", false);
         if (clientAPI == null) {
             clientAPI = createClient(settings);
             if (clientAPI.client() != null) {
@@ -276,10 +279,11 @@ public void update(IndexableObject object) throws IOException {
             setId(object.id());
         }
         if (getId() == null) {
-            return; // skip if no doc is specified to delete
+            return; // skip if no doc is specified to update
         }
         UpdateRequest request = new UpdateRequest().index(this.index).type(this.type).id(getId()).doc(object.source());
-        request.docAsUpsert(true);
+		if (this.docAsUpsert)
+			request.docAsUpsert(true);
 
         if (object.meta(ControlKeys._version.name()) != null) {
             request.versionType(VersionType.EXTERNAL)
@@ -292,7 +296,7 @@ public void update(IndexableObject object) throws IOException {
             request.parent(object.meta(ControlKeys._parent.name()));
         }
         if (logger.isTraceEnabled()) {
-            logger.trace("adding bulk update action {}/{}/{}", request.index(), request.type(), request.id());
+            logger.trace("adding bulk update action {}/{}/{} docAsUpsert {}", request.index(), request.type(), request.id(), this.docAsUpsert);
         }
         clientAPI.bulkUpdate(request);
     }