Merge pull request #635 from heljoyLiu/pulgin-gdb-update-to-set-property

gdbwrtier:  support set-property

gdbwriter/doc/gdbwriter.md

@@ -41,6 +41,14 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
                             {
                                 "random": "60,64",
                                 "type": "string"
+                            },
+                            {
+                                "random": "100,1000",
+                                "type": "long"
+                            },
+                            {
+                                "random": "32,48",
+                                "type": "string"
                             }
                         ],
                         "sliceRecordCount": 1000
@@ -70,6 +78,18 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
                                 "name": "vertex_propKey",
                                 "value": "${2}",
                                 "type": "string",
+                                "columnType": "vertexSetProperty"
+                            },
+                            {
+                                "name": "vertex_propKey",
+                                "value": "${3}",
+                                "type": "long",
+                                "columnType": "vertexSetProperty"
+                            },
+                            {
+                                "name": "vertex_propKey2",
+                                "value": "${4}",
+                                "type": "string",
                                 "columnType": "vertexProperty"
                             }
                         ]
@@ -290,6 +310,7 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
       * primaryKey：表示该字段是主键id
     * 点枚举值：
       * vertexProperty：labelType为点时，表示该字段是点的普通属性
+      * vertexSetProperty：labelType为点时，表示该字段是点的SET属性，value是SET属性中的一个属性值
       * vertexJsonProperty：labelType为点时，表示是点json属性，value结构请见备注**json properties示例**，点配置最多只允许出现一个json属性；
     * 边枚举值：
       * srcPrimaryKey：labelType为边时，表示该字段是起点主键id
@@ -305,6 +326,14 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
   >    {"k":"age","t":"int","v":"20"},
   >    {"k":"sex","t":"string","v":"male"}
   > ]}
+  >
+  > # json格式同样支持给点添加SET属性，格式如下
+  > {"properties":[
+  >    {"k":"name","t":"string","v":"tom","c":"set"},
+  >    {"k":"name","t":"string","v":"jack","c":"set"},
+  >    {"k":"age","t":"int","v":"20"},
+  >    {"k":"sex","t":"string","v":"male"}
+  > ]}
   > ```
 
 ## 4 性能报告
@@ -367,4 +396,5 @@ DataX压测机器
 - GDBWriter插件与用户查询DSL使用相同的GDB实例端口，导入时可能会影响查询性能
 
 ## FAQ
-无
+1. 使用SET属性需要升级GDB实例到`1.0.20`版本及以上。
+2. 边只支持普通单值属性，不能给边写SET属性数据。

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/GdbWriterErrorCode.java

@@ -27,7 +27,6 @@ public enum GdbWriterErrorCode implements ErrorCode {
 
     @Override
     public String toString() {
-        return String.format("Code:[%s], Description:[%s]. ", this.code,
-                this.description);
+        return String.format("Code:[%s], Description:[%s]. ", this.code, this.description);
     }
 }
\ No newline at end of file

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/Key.java

@@ -6,136 +6,164 @@ public final class Key {
      * 此处声明插件用到的需要插件使用者提供的配置项
      */
 
-	public final static String HOST = "host";
-	public final static String PORT = "port";
+    public final static String HOST = "host";
+    public final static String PORT = "port";
     public final static String USERNAME = "username";
-	public static final String PASSWORD = "password";
-
-	/**
-	 * import type and mode
-	 */
-	public static final String IMPORT_TYPE = "labelType";
-	public static final String UPDATE_MODE = "writeMode";
-
-	/**
-	 * label prefix issue
-	 */
-	public static final String ID_TRANS_RULE = "idTransRule";
-	public static final String SRC_ID_TRANS_RULE = "srcIdTransRule";
-	public static final String DST_ID_TRANS_RULE = "dstIdTransRule";
-
-	public static final String LABEL = "label";
-	public static final String SRC_LABEL = "srcLabel";
-	public static final String DST_LABEL = "dstLabel";
-
-	public static final String MAPPING = "mapping";
-
-	/**
-	 * column define in Gdb
-	 */
-	public static final String COLUMN = "column";
-	public static final String COLUMN_NAME = "name";
-	public static final String COLUMN_VALUE = "value";
-	public static final String COLUMN_TYPE = "type";
-	public static final String COLUMN_NODE_TYPE = "columnType";
-
-	/**
-	 * Gdb Vertex/Edge elements
-	 */
-	public static final String ID = "id";
-	public static final String FROM = "from";
-	public static final String TO = "to";
-	public static final String PROPERTIES = "properties";
-	public static final String PROP_KEY = "name";
-	public static final String PROP_VALUE = "value";
-	public static final String PROP_TYPE = "type";
-
-	public static final String PROPERTIES_JSON_STR = "propertiesJsonStr";
-	public static final String MAX_PROPERTIES_BATCH_NUM = "maxPropertiesBatchNumber";
-
-	/**
-	 * session less client configure for connect pool
-	 */
-	public static final String MAX_IN_PROCESS_PER_CONNECTION = "maxInProcessPerConnection";
-	public static final String MAX_CONNECTION_POOL_SIZE = "maxConnectionPoolSize";
-	public static final String MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = "maxSimultaneousUsagePerConnection";
-
-	public static final String MAX_RECORDS_IN_BATCH = "maxRecordsInBatch";
-	public static final String SESSION_STATE = "session";
-
-	public static enum ImportType {
-		/**
-		 * Import vertices
-		 */
-		VERTEX,
-		/**
-		 * Import edges
-		 */
-		EDGE;
-	}
-	
-	public static enum UpdateMode {
-		/**
-		 * Insert new records, fail if exists
-		 */
-		INSERT,
-		/**
-		 * Skip this record if exists
-		 */
-		SKIP,
-		/**
-		 * Update property of this record if exists
-		 */
-		MERGE;
-	}
-
-	public static enum ColumnType {
-		/**
-		 * vertex or edge id
-		 */
-		primaryKey,
-
-		/**
-		 * vertex property
-		 */
-		vertexProperty,
-
-		/**
-		 * start vertex id of edge
-		 */
-		srcPrimaryKey,
-
-		/**
-		 * end vertex id of edge
-		 */
-		dstPrimaryKey,
-
-		/**
-		 * edge property
-		 */
-		edgeProperty,
-
-		/**
-		 * vertex json style property
-		 */
-		vertexJsonProperty,
-
-		/**
-		 * edge json style property
-		 */
-		edgeJsonProperty
-	}
-
-	public static enum IdTransRule {
-		/**
-		 * vertex or edge id with 'label' prefix
-		 */
-		labelPrefix,
-
-		/**
-		 * vertex or edge id raw
-		 */
-		none
-	}
+    public static final String PASSWORD = "password";
+
+    /**
+     * import type and mode
+     */
+    public static final String IMPORT_TYPE = "labelType";
+    public static final String UPDATE_MODE = "writeMode";
+
+    /**
+     * label prefix issue
+     */
+    public static final String ID_TRANS_RULE = "idTransRule";
+    public static final String SRC_ID_TRANS_RULE = "srcIdTransRule";
+    public static final String DST_ID_TRANS_RULE = "dstIdTransRule";
+
+    public static final String LABEL = "label";
+    public static final String SRC_LABEL = "srcLabel";
+    public static final String DST_LABEL = "dstLabel";
+
+    public static final String MAPPING = "mapping";
+
+    /**
+     * column define in Gdb
+     */
+    public static final String COLUMN = "column";
+    public static final String COLUMN_NAME = "name";
+    public static final String COLUMN_VALUE = "value";
+    public static final String COLUMN_TYPE = "type";
+    public static final String COLUMN_NODE_TYPE = "columnType";
+
+    /**
+     * Gdb Vertex/Edge elements
+     */
+    public static final String ID = "id";
+    public static final String FROM = "from";
+    public static final String TO = "to";
+    public static final String PROPERTIES = "properties";
+    public static final String PROP_KEY = "name";
+    public static final String PROP_VALUE = "value";
+    public static final String PROP_TYPE = "type";
+
+    public static final String PROPERTIES_JSON_STR = "propertiesJsonStr";
+    public static final String MAX_PROPERTIES_BATCH_NUM = "maxPropertiesBatchNumber";
+
+    /**
+     * session less client configure for connect pool
+     */
+    public static final String MAX_IN_PROCESS_PER_CONNECTION = "maxInProcessPerConnection";
+    public static final String MAX_CONNECTION_POOL_SIZE = "maxConnectionPoolSize";
+    public static final String MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = "maxSimultaneousUsagePerConnection";
+
+    public static final String MAX_RECORDS_IN_BATCH = "maxRecordsInBatch";
+    public static final String SESSION_STATE = "session";
+
+    /**
+     * request length limit, include gdb element string length GDB字段长度限制配置，可分别配置各字段的限制，超过限制的记录会当脏数据处理
+     */
+    public static final String MAX_GDB_STRING_LENGTH = "maxStringLengthLimit";
+    public static final String MAX_GDB_ID_LENGTH = "maxIdStringLengthLimit";
+    public static final String MAX_GDB_LABEL_LENGTH = "maxLabelStringLengthLimit";
+    public static final String MAX_GDB_PROP_KEY_LENGTH = "maxPropKeyStringLengthLimit";
+    public static final String MAX_GDB_PROP_VALUE_LENGTH = "maxPropValueStringLengthLimit";
+
+    public static final String MAX_GDB_REQUEST_LENGTH = "maxRequestLengthLimit";
+
+    public static enum ImportType {
+        /**
+         * Import vertices
+         */
+        VERTEX,
+        /**
+         * Import edges
+         */
+        EDGE;
+    }
+
+    public static enum UpdateMode {
+        /**
+         * Insert new records, fail if exists
+         */
+        INSERT,
+        /**
+         * Skip this record if exists
+         */
+        SKIP,
+        /**
+         * Update property of this record if exists
+         */
+        MERGE;
+    }
+
+    public static enum ColumnType {
+        /**
+         * vertex or edge id
+         */
+        primaryKey,
+
+        /**
+         * vertex property
+         */
+        vertexProperty,
+
+        /**
+         * vertex setProperty
+         */
+        vertexSetProperty,
+
+        /**
+         * start vertex id of edge
+         */
+        srcPrimaryKey,
+
+        /**
+         * end vertex id of edge
+         */
+        dstPrimaryKey,
+
+        /**
+         * edge property
+         */
+        edgeProperty,
+
+        /**
+         * vertex json style property
+         */
+        vertexJsonProperty,
+
+        /**
+         * edge json style property
+         */
+        edgeJsonProperty
+    }
+
+    public static enum IdTransRule {
+        /**
+         * vertex or edge id with 'label' prefix
+         */
+        labelPrefix,
+
+        /**
+         * vertex or edge id raw
+         */
+        none
+    }
+
+    public static enum PropertyType {
+        /**
+         * single Vertex Property
+         */
+        single,
+
+        /**
+         * set Vertex Property
+         */
+        set
+    }
 
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/client/GdbGraphManager.java

@@ -3,37 +3,37 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.client;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import com.alibaba.datax.common.util.Configuration;
 import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbGraph;
 import com.alibaba.datax.plugin.writer.gdbwriter.model.ScriptGdbGraph;
 
-import java.util.ArrayList;
-import java.util.List;
-
 /**
  * @author jerrywang
  *
  */
 public class GdbGraphManager implements AutoCloseable {
-	private static final GdbGraphManager instance = new GdbGraphManager();
-	
-	private List<GdbGraph> graphs = new ArrayList<>();
-	
-	public static GdbGraphManager instance() {
-		return instance;
-	}
+    private static final GdbGraphManager INSTANCE = new GdbGraphManager();
+
+    private List<GdbGraph> graphs = new ArrayList<>();
+
+    public static GdbGraphManager instance() {
+        return INSTANCE;
+    }
 
-	public GdbGraph getGraph(Configuration config, boolean session) {
-		GdbGraph graph = new ScriptGdbGraph(config, session);
-		graphs.add(graph);
-		return graph;
-	}
+    public GdbGraph getGraph(final Configuration config, final boolean session) {
+        final GdbGraph graph = new ScriptGdbGraph(config, session);
+        this.graphs.add(graph);
+        return graph;
+    }
 
-	@Override
-	public void close() {
-		for(GdbGraph graph : graphs) {
-			graph.close();
-		}
-		graphs.clear();
-	}
+    @Override
+    public void close() {
+        for (final GdbGraph graph : this.graphs) {
+            graph.close();
+        }
+        this.graphs.clear();
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/client/GdbWriterConfig.java

@@ -3,39 +3,43 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.client;
 
+import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.assertConfig;
+import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.assertHasContent;
+
 import com.alibaba.datax.common.util.Configuration;
 import com.alibaba.datax.plugin.writer.gdbwriter.Key;
 
-import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.*;
-
 /**
  * @author jerrywang
  *
  */
 public class GdbWriterConfig {
-	public static final int DEFAULT_MAX_IN_PROCESS_PER_CONNECTION = 4;
-	public static final int DEFAULT_MAX_CONNECTION_POOL_SIZE = 8;
-	public static final int DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = 8;
-	public static final int DEFAULT_BATCH_PROPERTY_NUM = 30;
-	public static final int DEFAULT_RECORD_NUM_IN_BATCH = 16;
-
-	private Configuration config;
-
-	private GdbWriterConfig(Configuration config) {
-		this.config = config;
-
-		validate();
-	}
-
-	private void validate() {
-		assertHasContent(config, Key.HOST);
-		assertConfig(Key.PORT, () -> config.getInt(Key.PORT) > 0);
-
-		assertHasContent(config, Key.USERNAME);
-		assertHasContent(config, Key.PASSWORD);
-	}
-	
-	public static GdbWriterConfig of(Configuration config) {
-		return new GdbWriterConfig(config);
-	}
+    public static final int DEFAULT_MAX_IN_PROCESS_PER_CONNECTION = 4;
+    public static final int DEFAULT_MAX_CONNECTION_POOL_SIZE = 8;
+    public static final int DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = 8;
+    public static final int DEFAULT_BATCH_PROPERTY_NUM = 30;
+    public static final int DEFAULT_RECORD_NUM_IN_BATCH = 16;
+
+    public static final int MAX_STRING_LENGTH = 10240;
+    public static final int MAX_REQUEST_LENGTH = 65535 - 1000;
+
+    private Configuration config;
+
+    private GdbWriterConfig(final Configuration config) {
+        this.config = config;
+
+        validate();
+    }
+
+    public static GdbWriterConfig of(final Configuration config) {
+        return new GdbWriterConfig(config);
+    }
+
+    private void validate() {
+        assertHasContent(this.config, Key.HOST);
+        assertConfig(Key.PORT, () -> this.config.getInt(Key.PORT) > 0);
+
+        assertHasContent(this.config, Key.USERNAME);
+        assertHasContent(this.config, Key.PASSWORD);
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/mapping/GdbMapper.java

@@ -13,5 +13,5 @@ import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbElement;
  *
  */
 public interface GdbMapper {
-	Function<Record, GdbElement> getMapper(MappingRule rule);
+    Function<Record, GdbElement> getMapper(MappingRule rule);
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/mapping/MapperConfig.java

+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public
+ * License version 2 as published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.writer.gdbwriter.mapping;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.gdbwriter.Key;
+import com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/10/15
+ */
+
+public class MapperConfig {
+    private static MapperConfig instance = new MapperConfig();
+    private int maxIdLength;
+    private int maxLabelLength;
+    private int maxPropKeyLength;
+    private int maxPropValueLength;
+
+    private MapperConfig() {
+        this.maxIdLength = GdbWriterConfig.MAX_STRING_LENGTH;
+        this.maxLabelLength = GdbWriterConfig.MAX_STRING_LENGTH;
+        this.maxPropKeyLength = GdbWriterConfig.MAX_STRING_LENGTH;
+        this.maxPropValueLength = GdbWriterConfig.MAX_STRING_LENGTH;
+    }
+
+    public static MapperConfig getInstance() {
+        return instance;
+    }
+
+    public void updateConfig(final Configuration config) {
+        final int length = config.getInt(Key.MAX_GDB_STRING_LENGTH, GdbWriterConfig.MAX_STRING_LENGTH);
+
+        Integer sLength = config.getInt(Key.MAX_GDB_ID_LENGTH);
+        this.maxIdLength = sLength == null ? length : sLength;
+
+        sLength = config.getInt(Key.MAX_GDB_LABEL_LENGTH);
+        this.maxLabelLength = sLength == null ? length : sLength;
+
+        sLength = config.getInt(Key.MAX_GDB_PROP_KEY_LENGTH);
+        this.maxPropKeyLength = sLength == null ? length : sLength;
+
+        sLength = config.getInt(Key.MAX_GDB_PROP_VALUE_LENGTH);
+        this.maxPropValueLength = sLength == null ? length : sLength;
+    }
+
+    public int getMaxIdLength() {
+        return this.maxIdLength;
+    }
+
+    public int getMaxLabelLength() {
+        return this.maxLabelLength;
+    }
+
+    public int getMaxPropKeyLength() {
+        return this.maxPropKeyLength;
+    }
+
+    public int getMaxPropValueLength() {
+        return this.maxPropValueLength;
+    }
+
+}

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/mapping/MappingRule.java

@@ -7,6 +7,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import com.alibaba.datax.plugin.writer.gdbwriter.Key.ImportType;
+import com.alibaba.datax.plugin.writer.gdbwriter.Key.PropertyType;
 
 import lombok.Data;
 
@@ -16,26 +17,30 @@ import lombok.Data;
  */
 @Data
 public class MappingRule {
-	private String id = null;
+    private String id = null;
 
-	private String label = null;
-	
-	private ImportType importType = null;
-	
-	private String from = null;
+    private String label = null;
 
-	private String to = null;
+    private ImportType importType = null;
 
-	private List<PropertyMappingRule> properties = new ArrayList<>();
+    private String from = null;
 
-	private String propertiesJsonStr = null;
+    private String to = null;
 
-	@Data
-	public static class PropertyMappingRule {
-		private String key = null;
-		
-		private String value = null;
-		
-		private ValueType valueType = null;
-	}
+    private List<PropertyMappingRule> properties = new ArrayList<>();
+
+    private String propertiesJsonStr = null;
+
+    private boolean numPattern = false;
+
+    @Data
+    public static class PropertyMappingRule {
+        private String key = null;
+
+        private String value = null;
+
+        private ValueType valueType = null;
+
+        private PropertyType pType = PropertyType.single;
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/mapping/ValueType.java

@@ -8,6 +8,7 @@ import java.util.Map;
 import java.util.function.Function;
 
 import com.alibaba.datax.common.element.Column;
+
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -16,56 +17,61 @@ import lombok.extern.slf4j.Slf4j;
  */
 @Slf4j
 public enum ValueType {
-	INT(Integer.class, "int", Column::asLong, Integer::valueOf),
-	LONG(Long.class, "long", Column::asLong, Long::valueOf),
-	DOUBLE(Double.class, "double", Column::asDouble, Double::valueOf),
-	FLOAT(Float.class, "float", Column::asDouble, Float::valueOf),
-	BOOLEAN(Boolean.class, "boolean", Column::asBoolean, Boolean::valueOf),
-	STRING(String.class, "string", Column::asString, String::valueOf);
+    /**
+     * property value type
+     */
+    INT(Integer.class, "int", Column::asLong, Integer::valueOf),
+    INTEGER(Integer.class, "integer", Column::asLong, Integer::valueOf),
+    LONG(Long.class, "long", Column::asLong, Long::valueOf),
+    DOUBLE(Double.class, "double", Column::asDouble, Double::valueOf),
+    FLOAT(Float.class, "float", Column::asDouble, Float::valueOf),
+    BOOLEAN(Boolean.class, "boolean", Column::asBoolean, Boolean::valueOf),
+    STRING(String.class, "string", Column::asString, String::valueOf);
+
+    private Class<?> type = null;
+    private String shortName = null;
+    private Function<Column, Object> columnFunc = null;
+    private Function<String, Object> fromStrFunc = null;
+
+    private ValueType(final Class<?> type, final String name, final Function<Column, Object> columnFunc,
+                      final Function<String, Object> fromStrFunc) {
+        this.type = type;
+        this.shortName = name;
+        this.columnFunc = columnFunc;
+        this.fromStrFunc = fromStrFunc;
+
+        ValueTypeHolder.shortName2type.put(name, this);
+    }
+
+    public static ValueType fromShortName(final String name) {
+        return ValueTypeHolder.shortName2type.get(name);
+    }
+
+    public Class<?> type() {
+        return this.type;
+    }
 
-	private Class<?> type = null;
-	private String shortName = null;
-	private Function<Column, Object> columnFunc = null;
-	private Function<String, Object> fromStrFunc = null;
+    public String shortName() {
+        return this.shortName;
+    }
 
-	private ValueType(Class<?> type, String name, Function<Column, Object> columnFunc, Function<String, Object> fromStrFunc) {
-		this.type = type;
-		this.shortName = name;
-		this.columnFunc = columnFunc;
-		this.fromStrFunc = fromStrFunc;
-		
-		ValueTypeHolder.shortName2type.put(name, this);
-	}
-	
-	public static ValueType fromShortName(String name) {
-		return ValueTypeHolder.shortName2type.get(name);
-	}
+    public Object applyColumn(final Column column) {
+        try {
+            if (column == null) {
+                return null;
+            }
+            return this.columnFunc.apply(column);
+        } catch (final Exception e) {
+            log.error("applyColumn error {}, column {}", e.toString(), column);
+            throw e;
+        }
+    }
 
-	public Class<?> type() {
-		return this.type;
-	}
-	
-	public String shortName() {
-		return this.shortName;
-	}
-	
-	public Object applyColumn(Column column) {
-		try {
-			if (column == null) {
-				return null;
-			}
-			return columnFunc.apply(column);
-		} catch (Exception e) {
-			log.error("applyColumn error {}, column {}", e.toString(), column);
-			throw e;
-		}
-	}
-	
-	public Object fromStrFunc(String str) {
-		return fromStrFunc.apply(str);
-	}
+    public Object fromStrFunc(final String str) {
+        return this.fromStrFunc.apply(str);
+    }
 
-	private static class ValueTypeHolder {
-		private static Map<String, ValueType> shortName2type = new HashMap<>();
-	}
+    private static class ValueTypeHolder {
+        private static Map<String, ValueType> shortName2type = new HashMap<>();
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/model/AbstractGdbGraph.java

@@ -3,20 +3,24 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.model;
 
-import com.alibaba.datax.common.util.Configuration;
-import com.alibaba.datax.plugin.writer.gdbwriter.Key;
-import com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig;
+import static com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM;
+import static com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig.MAX_REQUEST_LENGTH;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
-import lombok.extern.slf4j.Slf4j;
 import org.apache.tinkerpop.gremlin.driver.Client;
 import org.apache.tinkerpop.gremlin.driver.Cluster;
 import org.apache.tinkerpop.gremlin.driver.RequestOptions;
 import org.apache.tinkerpop.gremlin.driver.ResultSet;
 import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
 
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.gdbwriter.Key;
+import com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig;
+
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * @author jerrywang
@@ -24,128 +28,124 @@ import java.util.concurrent.TimeUnit;
  */
 @Slf4j
 public abstract class AbstractGdbGraph implements GdbGraph {
-	private final static int DEFAULT_TIMEOUT = 30000;
-
-	protected Client client = null;
-	protected Key.UpdateMode updateMode = Key.UpdateMode.INSERT;
-	protected int propertiesBatchNum = GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM;
-	protected boolean session = false;
-
-
-	protected AbstractGdbGraph() {}
-
-	protected AbstractGdbGraph(Configuration config, boolean session) {
-		initClient(config, session);
-	}
-
-	protected void initClient(Configuration config, boolean session) {
-		updateMode = Key.UpdateMode.valueOf(config.getString(Key.UPDATE_MODE, "INSERT"));
-		log.info("init graphdb client");
-		String host = config.getString(Key.HOST);
-		int port = config.getInt(Key.PORT);
-		String username = config.getString(Key.USERNAME);
-		String password = config.getString(Key.PASSWORD);
-		int maxDepthPerConnection = config.getInt(Key.MAX_IN_PROCESS_PER_CONNECTION,
-				GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION);
-
-		int maxConnectionPoolSize = config.getInt(Key.MAX_CONNECTION_POOL_SIZE,
-			GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE);
-
-		int maxSimultaneousUsagePerConnection = config.getInt(Key.MAX_SIMULTANEOUS_USAGE_PER_CONNECTION,
-			GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION);
-
-		this.session = session;
-		if (this.session) {
-			maxConnectionPoolSize = GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE;
-			maxDepthPerConnection = GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION;
-			maxSimultaneousUsagePerConnection = GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION;
-		}
-
-		try {
-			Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
-					.serializer(Serializers.GRAPHBINARY_V1D0)
-					.maxContentLength(1048576)
-				    .maxInProcessPerConnection(maxDepthPerConnection)
-				    .minInProcessPerConnection(0)
-				    .maxConnectionPoolSize(maxConnectionPoolSize)
-				    .minConnectionPoolSize(maxConnectionPoolSize)
-				    .maxSimultaneousUsagePerConnection(maxSimultaneousUsagePerConnection)
-				    .resultIterationBatchSize(64)
-				    .create();
-			client = session ? cluster.connect(UUID.randomUUID().toString()).init() : cluster.connect().init();
-			warmClient(maxConnectionPoolSize*maxDepthPerConnection);
-		} catch (RuntimeException e) {
-			log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
-			throw e;
-		}
-
-		propertiesBatchNum = config.getInt(Key.MAX_PROPERTIES_BATCH_NUM, GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM);
-	}
-
-
-	/**
-	 * @param dsl
-	 * @param parameters
-	 */
-	protected void runInternal(String dsl, final Map<String, Object> parameters) throws Exception {
-		RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
-		if (parameters != null && !parameters.isEmpty()) {
-			parameters.forEach(options::addParameter);
-		}
-
-		ResultSet results = client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
-		results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
-	}
-
-	void beginTx() {
-		if (!session) {
-			return;
-		}
-
-		String dsl = "g.tx().open()";
-		client.submit(dsl).all().join();
-	}
-
-	void doCommit() {
-		if (!session) {
-			return;
-		}
-
-		try {
-			String dsl = "g.tx().commit()";
-			client.submit(dsl).all().join();
-		} catch (Exception e) {
-			throw new RuntimeException(e);
-		}
-	}
-
-	void doRollback() {
-		if (!session) {
-			return;
-		}
-
-		String dsl = "g.tx().rollback()";
-		client.submit(dsl).all().join();
-	}
-
-	private void warmClient(int num) {
-		try {
-			beginTx();
-			runInternal("g.V('test')", null);
-			doCommit();
-			log.info("warm graphdb client over");
-		} catch (Exception e) {
-			log.error("warmClient error");
-			doRollback();
-			throw new RuntimeException(e);
-		}
-	}
-
-	@Override
-	public void close() {
-		if (client != null) {
-			log.info("close graphdb client");
-			client.close();
-		}
-	}
+    private final static int DEFAULT_TIMEOUT = 30000;
+
+    protected Client client = null;
+    protected Key.UpdateMode updateMode = Key.UpdateMode.INSERT;
+    protected int propertiesBatchNum = DEFAULT_BATCH_PROPERTY_NUM;
+    protected boolean session = false;
+    protected int maxRequestLength = GdbWriterConfig.MAX_REQUEST_LENGTH;
+
+    protected AbstractGdbGraph() {}
+
+    protected AbstractGdbGraph(final Configuration config, final boolean session) {
+        initClient(config, session);
+    }
+
+    protected void initClient(final Configuration config, final boolean session) {
+        this.updateMode = Key.UpdateMode.valueOf(config.getString(Key.UPDATE_MODE, "INSERT"));
+        log.info("init graphdb client");
+        final String host = config.getString(Key.HOST);
+        final int port = config.getInt(Key.PORT);
+        final String username = config.getString(Key.USERNAME);
+        final String password = config.getString(Key.PASSWORD);
+        int maxDepthPerConnection =
+            config.getInt(Key.MAX_IN_PROCESS_PER_CONNECTION, GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION);
+
+        int maxConnectionPoolSize =
+            config.getInt(Key.MAX_CONNECTION_POOL_SIZE, GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE);
+
+        int maxSimultaneousUsagePerConnection = config.getInt(Key.MAX_SIMULTANEOUS_USAGE_PER_CONNECTION,
+            GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION);
+
+        this.session = session;
+        if (this.session) {
+            maxConnectionPoolSize = GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE;
+            maxDepthPerConnection = GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION;
+            maxSimultaneousUsagePerConnection = GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION;
+        }
+
+        try {
+            final Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
+                .serializer(Serializers.GRAPHBINARY_V1D0).maxContentLength(1048576)
+                .maxInProcessPerConnection(maxDepthPerConnection).minInProcessPerConnection(0)
+                .maxConnectionPoolSize(maxConnectionPoolSize).minConnectionPoolSize(maxConnectionPoolSize)
+                .maxSimultaneousUsagePerConnection(maxSimultaneousUsagePerConnection).resultIterationBatchSize(64)
+                .create();
+            this.client = session ? cluster.connect(UUID.randomUUID().toString()).init() : cluster.connect().init();
+            warmClient(maxConnectionPoolSize * maxDepthPerConnection);
+        } catch (final RuntimeException e) {
+            log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
+            throw e;
+        }
+
+        this.propertiesBatchNum = config.getInt(Key.MAX_PROPERTIES_BATCH_NUM, DEFAULT_BATCH_PROPERTY_NUM);
+        this.maxRequestLength = config.getInt(Key.MAX_GDB_REQUEST_LENGTH, MAX_REQUEST_LENGTH);
+    }
+
+    /**
+     * @param dsl
+     * @param parameters
+     */
+    protected void runInternal(final String dsl, final Map<String, Object> parameters) throws Exception {
+        final RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
+        if (parameters != null && !parameters.isEmpty()) {
+            parameters.forEach(options::addParameter);
+        }
+
+        final ResultSet results = this.client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
+        results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
+    }
+
+    void beginTx() {
+        if (!this.session) {
+            return;
+        }
+
+        final String dsl = "g.tx().open()";
+        this.client.submit(dsl).all().join();
+    }
+
+    void doCommit() {
+        if (!this.session) {
+            return;
+        }
+
+        try {
+            final String dsl = "g.tx().commit()";
+            this.client.submit(dsl).all().join();
+        } catch (final Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    void doRollback() {
+        if (!this.session) {
+            return;
+        }
+
+        final String dsl = "g.tx().rollback()";
+        this.client.submit(dsl).all().join();
+    }
+
+    private void warmClient(final int num) {
+        try {
+            beginTx();
+            runInternal("g.V('test')", null);
+            doCommit();
+            log.info("warm graphdb client over");
+        } catch (final Exception e) {
+            log.error("warmClient error");
+            doRollback();
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (this.client != null) {
+            log.info("close graphdb client");
+            this.client.close();
+        }
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/model/GdbEdge.java

@@ -3,7 +3,8 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.model;
 
-import lombok.Data;
+import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MapperConfig;
+
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
 
@@ -11,10 +12,33 @@ import lombok.ToString;
  * @author jerrywang
  *
  */
-@Data
 @EqualsAndHashCode(callSuper = true)
 @ToString(callSuper = true)
 public class GdbEdge extends GdbElement {
-	private String from = null;
-	private String to = null;
+    private String from = null;
+    private String to = null;
+
+    public String getFrom() {
+        return this.from;
+    }
+
+    public void setFrom(final String from) {
+        final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
+        if (from.length() > maxIdLength) {
+            throw new IllegalArgumentException("from length over limit(" + maxIdLength + ")");
+        }
+        this.from = from;
+    }
+
+    public String getTo() {
+        return this.to;
+    }
+
+    public void setTo(final String to) {
+        final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
+        if (to.length() > maxIdLength) {
+            throw new IllegalArgumentException("to length over limit(" + maxIdLength + ")");
+        }
+        this.to = to;
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/model/GdbElement.java

@@ -3,18 +3,107 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.model;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.LinkedList;
+import java.util.List;
 
-import lombok.Data;
+import com.alibaba.datax.plugin.writer.gdbwriter.Key.PropertyType;
+import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MapperConfig;
 
 /**
  * @author jerrywang
  *
  */
-@Data
 public class GdbElement {
-	String id = null;
-	String label = null;
-	Map<String, Object> properties = new HashMap<>();
+    private String id = null;
+    private String label = null;
+    private List<GdbProperty> properties = new LinkedList<>();
+
+    public String getId() {
+        return this.id;
+    }
+
+    public void setId(final String id) {
+        final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
+        if (id.length() > maxIdLength) {
+            throw new IllegalArgumentException("id length over limit(" + maxIdLength + ")");
+        }
+        this.id = id;
+    }
+
+    public String getLabel() {
+        return this.label;
+    }
+
+    public void setLabel(final String label) {
+        final int maxLabelLength = MapperConfig.getInstance().getMaxLabelLength();
+        if (label.length() > maxLabelLength) {
+            throw new IllegalArgumentException("label length over limit(" + maxLabelLength + ")");
+        }
+        this.label = label;
+    }
+
+    public List<GdbProperty> getProperties() {
+        return this.properties;
+    }
+
+    public void addProperty(final String propKey, final Object propValue, final PropertyType card) {
+        if (propKey == null || propValue == null) {
+            return;
+        }
+
+        final int maxPropKeyLength = MapperConfig.getInstance().getMaxPropKeyLength();
+        if (propKey.length() > maxPropKeyLength) {
+            throw new IllegalArgumentException("property key length over limit(" + maxPropKeyLength + ")");
+        }
+        if (propValue instanceof String) {
+            final int maxPropValueLength = MapperConfig.getInstance().getMaxPropValueLength();
+            if (((String)propValue).length() > maxPropKeyLength) {
+                throw new IllegalArgumentException("property value length over limit(" + maxPropValueLength + ")");
+            }
+        }
+
+        this.properties.add(new GdbProperty(propKey, propValue, card));
+    }
+
+    public void addProperty(final String propKey, final Object propValue) {
+        addProperty(propKey, propValue, PropertyType.single);
+    }
+
+    @Override
+    public String toString() {
+        final StringBuffer sb = new StringBuffer(this.id + "[" + this.label + "]{");
+        this.properties.forEach(n -> {
+            sb.append(n.cardinality.name());
+            sb.append("[");
+            sb.append(n.key);
+            sb.append(" - ");
+            sb.append(String.valueOf(n.value));
+            sb.append("]");
+        });
+        return sb.toString();
+    }
+
+    public static class GdbProperty {
+        private String key;
+        private Object value;
+        private PropertyType cardinality;
+
+        private GdbProperty(final String key, final Object value, final PropertyType card) {
+            this.key = key;
+            this.value = value;
+            this.cardinality = card;
+        }
+
+        public PropertyType getCardinality() {
+            return this.cardinality;
+        }
+
+        public String getKey() {
+            return this.key;
+        }
+
+        public Object getValue() {
+            return this.value;
+        }
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/model/GdbGraph.java

@@ -3,18 +3,19 @@
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.model;
 
+import java.util.List;
+
 import com.alibaba.datax.common.element.Record;
-import groovy.lang.Tuple2;
 
-import java.util.List;
+import groovy.lang.Tuple2;
 
 /**
  * @author jerrywang
  *
  */
 public interface GdbGraph extends AutoCloseable {
-	List<Tuple2<Record, Exception>> add(List<Tuple2<Record, GdbElement>> records);
+    List<Tuple2<Record, Exception>> add(List<Tuple2<Record, GdbElement>> records);
 
-	@Override
-	void close();
+    @Override
+    void close();
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/util/ConfigHelper.java

@@ -7,53 +7,57 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.util.function.Supplier;
 
+import org.apache.commons.lang3.StringUtils;
+
 import com.alibaba.datax.common.exception.DataXException;
 import com.alibaba.datax.common.util.Configuration;
 import com.alibaba.datax.plugin.writer.gdbwriter.GdbWriterErrorCode;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 
-import org.apache.commons.lang3.StringUtils;
-
 /**
  * @author jerrywang
  *
  */
 public interface ConfigHelper {
-	static void assertConfig(String key, Supplier<Boolean> f) {
-		if (!f.get()) {
-			throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, key);
-		}
-	}
+    static void assertConfig(final String key, final Supplier<Boolean> f) {
+        if (!f.get()) {
+            throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, key);
+        }
+    }
+
+    static void assertHasContent(final Configuration config, final String key) {
+        assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
+    }
 
-	static void assertHasContent(Configuration config, String key) {
-		assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
-	}
+    /**
+     * NOTE: {@code Configuration::get(String, Class<T>)} doesn't work.
+     * 
+     * @param conf
+     *            Configuration
+     * @param key
+     *            key path to configuration
+     * @param cls
+     *            Class of result type
+     * @return the target configuration object of type T
+     */
+    static <T> T getConfig(final Configuration conf, final String key, final Class<T> cls) {
+        final JSONObject j = (JSONObject)conf.get(key);
+        return JSON.toJavaObject(j, cls);
+    }
 
-	/**
-	 * NOTE: {@code Configuration::get(String, Class<T>)} doesn't work.
-	 * 
-	 * @param conf Configuration
-	 * @param key key path to configuration
-	 * @param cls Class of result type
-	 * @return the target configuration object of type T
-	 */
-	static <T> T getConfig(Configuration conf, String key, Class<T> cls) {
-		JSONObject j = (JSONObject) conf.get(key);
-		return JSON.toJavaObject(j, cls);
-	}
-	
-	/**
-	 * Create a configuration from the specified file on the classpath.
-	 * 
-	 * @param name file name
-	 * @return Configuration instance.
-	 */
-	static Configuration fromClasspath(String name) {
-		try (InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
-			return Configuration.from(is);
-		} catch (IOException e) {
-			throw new IllegalArgumentException("File not found: " + name);
-		}
-	}
+    /**
+     * Create a configuration from the specified file on the classpath.
+     * 
+     * @param name
+     *            file name
+     * @return Configuration instance.
+     */
+    static Configuration fromClasspath(final String name) {
+        try (final InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
+            return Configuration.from(is);
+        } catch (final IOException e) {
+            throw new IllegalArgumentException("File not found: " + name);
+        }
+    }
 }

gdbwriter/src/main/java/com/alibaba/datax/plugin/writer/gdbwriter/util/GdbDuplicateIdException.java

 /*
- * (C)  2019-present Alibaba Group Holding Limited.
+ * (C) 2019-present Alibaba Group Holding Limited.
  *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 2 as
- * published by the Free Software Foundation.
+ * This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public
+ * License version 2 as published by the Free Software Foundation.
  */
 package com.alibaba.datax.plugin.writer.gdbwriter.util;
 
@@ -13,11 +12,11 @@ package com.alibaba.datax.plugin.writer.gdbwriter.util;
  */
 
 public class GdbDuplicateIdException extends Exception {
-	public GdbDuplicateIdException(Exception e) {
-		super(e);
-	}
+    public GdbDuplicateIdException(Exception e) {
+        super(e);
+    }
 
-	public GdbDuplicateIdException() {
-		super();
-	}
+    public GdbDuplicateIdException() {
+        super();
+    }
 }