Merge pull request #1276 from alibaba/dependabot/maven/org.apache.logging.log4j-log4j-core-2.17.1

Bump log4j-core from 2.17.0 to 2.17.1

README.md

@@ -3,12 +3,12 @@
 
 # DataX
 
-DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。
+DataX 是阿里云 [DataWorks数据集成](https://www.aliyun.com/product/bigdata/ide) 的开源版本，在阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、OceanBase、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS 等各种异构数据源之间高效的数据同步功能。
 
 # DataX 商业版本
 阿里云DataWorks数据集成是DataX团队在阿里云上的商业化产品，致力于提供复杂网络环境下、丰富的异构数据源之间高速稳定的数据移动能力，以及繁杂业务背景下的数据同步解决方案。目前已经支持云上近3000家客户，单日同步数据超过3万亿条。DataWorks数据集成目前支持离线50+种数据源，可以进行整库迁移、批量上云、增量同步、分库分表等各类同步解决方案。2020年更新实时同步能力，2020年更新实时同步能力，支持10+种数据源的读写任意组合。提供MySQL，Oracle等多种数据源到阿里云MaxCompute，Hologres等大数据引擎的一键全增量同步解决方案。
 
-https://www.aliyun.com/product/bigdata/ide
+商业版本参见：  https://www.aliyun.com/product/bigdata/ide
 
 
 # Features
@@ -39,6 +39,7 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
 | ------------ | ---------- | :-------: | :-------: |:-------: |
 | RDBMS 关系型数据库 | MySQL      |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mysqlreader/doc/mysqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mysqlwriter/doc/mysqlwriter.md)|
 |              | Oracle     |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/oraclereader/doc/oraclereader.md) 、[写](https://github.com/alibaba/DataX/blob/master/oraclewriter/doc/oraclewriter.md)|
+|              | OceanBase  |     √     |     √     |[读](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase) 、[写](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase)|
 |              | SQLServer  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/sqlserverreader/doc/sqlserverreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/sqlserverwriter/doc/sqlserverwriter.md)|
 |              | PostgreSQL |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/postgresqlreader/doc/postgresqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/postgresqlwriter/doc/postgresqlwriter.md)|
 |              | DRDS |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/drdsreader/doc/drdsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/drdswriter/doc/drdswriter.md)|
@@ -46,7 +47,7 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
 | 阿里云数仓数据存储    | ODPS       |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/odpsreader/doc/odpsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/odpswriter/doc/odpswriter.md)|
 |              | ADS        |           |     √     |[写](https://github.com/alibaba/DataX/blob/master/adswriter/doc/adswriter.md)|
 |              | OSS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/ossreader/doc/ossreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/osswriter/doc/osswriter.md)|
-|              | OCS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/ocsreader/doc/ocsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)|
+|              | OCS        |           |     √     |[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)|
 | NoSQL数据存储    | OTS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/otsreader/doc/otsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/otswriter/doc/otswriter.md)|
 |              | Hbase0.94  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase094xreader/doc/hbase094xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase094xwriter/doc/hbase094xwriter.md)|
 |              | Hbase1.1   |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase11xreader/doc/hbase11xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase11xwriter/doc/hbase11xwriter.md)|
@@ -152,3 +153,6 @@ This software is free to use under the Apache License [Apache license](https://g
 - DataX开源用户交流群5
     - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A45.jpg" width="20%" height="20%">
 
+- DataX开源用户交流群6 
+    - <img src="https://user-images.githubusercontent.com/1905000/124073771-139cbd00-da75-11eb-9a3f-598cba145a76.png" width="20%" height="20%">
+

core/src/main/bin/datax.py

@@ -1,23 +1,26 @@
 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
 
-import sys
-import os
-import signal
-import subprocess
-import time
-import re
-import socket
-import json
-from optparse import OptionParser
-from optparse import OptionGroup
-from string import Template
 import codecs
+import json
+import os
 import platform
+import re
+import signal
+import socket
+import subprocess
+import sys
+import time
+from optparse import OptionGroup
+from optparse import OptionParser
+from string import Template
+
+ispy2 = sys.version_info.major == 2
 
 def isWindows():
     return platform.system() == 'Windows'
 
+
 DATAX_HOME = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 DATAX_VERSION = 'DATAX-OPENSOURCE-3.0'
@@ -52,13 +55,19 @@ def getLocalIp():
 
 def suicide(signum, e):
     global child_process
-    print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
+    if ispy2:
+        print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
+    else:
+        print("[Error] DataX receive unexpected signal %d, starts to suicide." % (signum), sys.stderr)
 
     if child_process:
         child_process.send_signal(signal.SIGQUIT)
         time.sleep(1)
         child_process.kill()
-    print >> sys.stderr, "DataX Process was killed ! you did ?"
+    if ispy2:
+        print >> sys.stderr, "DataX Process was killed ! you did ?"
+    else:
+        print("DataX Process was killed ! you did ?", sys.stderr)
     sys.exit(RET_STATE["KILL"])
 
 
@@ -92,10 +101,10 @@ def getOptionParser():
                                        'if you have mutiple parameters: -p"-DtableName=your-table-name -DcolumnName=your-column-name".'
                                        'Note: you should config in you job tableName with ${tableName}.')
     prodEnvOptionGroup.add_option("-r", "--reader", metavar="<parameter used in view job config[reader] template>",
-                                  action="store", dest="reader",type="string",
+                                  action="store", dest="reader", type="string",
                                   help='View job config[reader] template, eg: mysqlreader,streamreader')
     prodEnvOptionGroup.add_option("-w", "--writer", metavar="<parameter used in view job config[writer] template>",
-                                  action="store", dest="writer",type="string",
+                                  action="store", dest="writer", type="string",
                                   help='View job config[writer] template, eg: mysqlwriter,streamwriter')
     parser.add_option_group(prodEnvOptionGroup)
 
@@ -108,45 +117,50 @@ def getOptionParser():
     parser.add_option_group(devEnvOptionGroup)
     return parser
 
+
 def generateJobConfigTemplate(reader, writer):
-    readerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n" % (reader,reader,reader)
-    writerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n " % (writer,writer,writer)
-    print readerRef
-    print writerRef
+    readerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n" % (
+        reader, reader, reader)
+    writerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n " % (
+        writer, writer, writer)
+    print(readerRef)
+    print(writerRef)
     jobGuid = 'Please save the following configuration as a json file and  use\n     python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json \nto run the job.\n'
-    print jobGuid
-    jobTemplate={
-      "job": {
-        "setting": {
-          "speed": {
-            "channel": ""
-          }
-        },
-        "content": [
-          {
-            "reader": {},
-            "writer": {}
-          }
-        ]
-      }
+    print(jobGuid)
+    jobTemplate = {
+        "job": {
+            "setting": {
+                "speed": {
+                    "channel": ""
+                }
+            },
+            "content": [
+                {
+                    "reader": {},
+                    "writer": {}
+                }
+            ]
+        }
     }
-    readerTemplatePath = "%s/plugin/reader/%s/plugin_job_template.json" % (DATAX_HOME,reader)
-    writerTemplatePath = "%s/plugin/writer/%s/plugin_job_template.json" % (DATAX_HOME,writer)
+    readerTemplatePath = "%s/plugin/reader/%s/plugin_job_template.json" % (DATAX_HOME, reader)
+    writerTemplatePath = "%s/plugin/writer/%s/plugin_job_template.json" % (DATAX_HOME, writer)
     try:
-      readerPar = readPluginTemplate(readerTemplatePath);
-    except Exception, e:
-       print "Read reader[%s] template error: can\'t find file %s" % (reader,readerTemplatePath)
+        readerPar = readPluginTemplate(readerTemplatePath)
+    except:
+        print("Read reader[%s] template error: can\'t find file %s" % (reader, readerTemplatePath))
     try:
-      writerPar = readPluginTemplate(writerTemplatePath);
-    except Exception, e:
-      print "Read writer[%s] template error: : can\'t find file %s" % (writer,writerTemplatePath)
-    jobTemplate['job']['content'][0]['reader'] = readerPar;
-    jobTemplate['job']['content'][0]['writer'] = writerPar;
-    print json.dumps(jobTemplate, indent=4, sort_keys=True)
+        writerPar = readPluginTemplate(writerTemplatePath)
+    except:
+        print("Read writer[%s] template error: : can\'t find file %s" % (writer, writerTemplatePath))
+    jobTemplate['job']['content'][0]['reader'] = readerPar
+    jobTemplate['job']['content'][0]['writer'] = writerPar
+    print(json.dumps(jobTemplate, indent=4, sort_keys=True))
+
 
 def readPluginTemplate(plugin):
     with open(plugin, 'r') as f:
-            return json.load(f)
+        return json.load(f)
+
 
 def isUrl(path):
     if not path:
@@ -168,7 +182,7 @@ def buildStartCommand(options, args):
 
     if options.remoteDebug:
         tempJVMCommand = tempJVMCommand + " " + REMOTE_DEBUG_CONFIG
-        print 'local ip: ', getLocalIp()
+        print('local ip: ', getLocalIp())
 
     if options.loglevel:
         tempJVMCommand = tempJVMCommand + " " + ("-Dloglevel=%s" % (options.loglevel))
@@ -198,11 +212,11 @@ def buildStartCommand(options, args):
 
 
 def printCopyright():
-    print '''
+    print('''
 DataX (%s), From Alibaba !
 Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
 
-''' % DATAX_VERSION
+''' % DATAX_VERSION)
     sys.stdout.flush()
 
 
@@ -211,7 +225,7 @@ if __name__ == "__main__":
     parser = getOptionParser()
     options, args = parser.parse_args(sys.argv[1:])
     if options.reader is not None and options.writer is not None:
-        generateJobConfigTemplate(options.reader,options.writer)
+        generateJobConfigTemplate(options.reader, options.writer)
         sys.exit(RET_STATE['OK'])
     if len(args) != 1:
         parser.print_help()

drdsreader/doc/drdsreader.md

@@ -50,7 +50,7 @@ DRDS的插件目前DataX只适配了Mysql引擎的场景，DRDS对于DataX而言
                         // 数据库连接密码
                         "password": "root",
                         "column": [
-                            "id"，"name"
+                            "id","name"
                         ],
                         "connection": [
                             {

hbase11xwriter/doc/hbase11xwriter.md

@@ -203,19 +203,20 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：要写入的hbase字段。index：指定该列对应reader端column的索引，从0开始；name：指定hbase表中的列，必须为 列族:列名 的格式；type：指定写入数据类型，用于转换HBase byte[]。配置格式如下：
 	
 	```
-"column": [
-              {
-                "index":1,
-                "name": "cf1:q1",
-                "type": "string"
-              },
-              {
-                "index":2,
-                "name": "cf1:q2",
-                "type": "string"
-              }
-           ］ 	
-		            
+
+	"column": [
+		      {
+			"index":1,
+			"name": "cf1:q1",
+			"type": "string"
+		      },
+		      {
+			"index":2,
+			"name": "cf1:q2",
+			"type": "string"
+		      }
+		   ］ 	
+
 	```
 
 	* 必选：是<br />
@@ -227,17 +228,17 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：要写入的hbase的rowkey列。index：指定该列对应reader端column的索引，从0开始，若为常量index为－1；type：指定写入数据类型，用于转换HBase byte[]；value：配置常量，常作为多个字段的拼接符。hbasewriter会将rowkeyColumn中所有列按照配置顺序进行拼接作为写入hbase的rowkey，不能全为常量。配置格式如下：
 	
 	```
-"rowkeyColumn": [
-                {
-                  "index":0,
-                  "type":"string"
-                },
-                {
-                  "index":-1,
-                  "type":"string",
-                  "value":"_"
-                }
-            ] 	
+	"rowkeyColumn": [
+			{
+			  "index":0,
+			  "type":"string"
+			},
+			{
+			  "index":-1,
+			  "type":"string",
+			  "value":"_"
+			}
+		    ] 	
 		            
 	```
 
@@ -250,19 +251,19 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：指定写入hbase的时间戳。支持：当前时间、指定时间列，指定时间，三者选一。若不配置表示用当前时间。index：指定对应reader端column的索引，从0开始，需保证能转换为long,若是Date类型，会尝试用yyyy-MM-dd HH:mm:ss和yyyy-MM-dd HH:mm:ss SSS去解析；若为指定时间index为－1；value：指定时间的值,long值。配置格式如下：
 	
 	```
-"versionColumn":{
-	"index":1
-}
+	"versionColumn":{
+		"index":1
+	}
 		            
 	```
 	
 	或者
 	
 	```
-"versionColumn":{
-	"index":－1,
-	"value":123456789
-}
+	"versionColumn":{
+		"index":－1,
+		"value":123456789
+	}
 		            
 	```
 

introduction.md

@@ -36,6 +36,7 @@ DataX本身作为离线数据同步框架，采用Framework + plugin架构构建
 | ------------ | ---------- | :-------: | :-------: |:-------: |
 | RDBMS 关系型数据库 | MySQL      |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mysqlreader/doc/mysqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mysqlwriter/doc/mysqlwriter.md)|
 |              | Oracle     |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/oraclereader/doc/oraclereader.md) 、[写](https://github.com/alibaba/DataX/blob/master/oraclewriter/doc/oraclewriter.md)|
+|              | OceanBase  |     √     |     √     |[读](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase) 、[写](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase)|
 |              | SQLServer  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/sqlserverreader/doc/sqlserverreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/sqlserverwriter/doc/sqlserverwriter.md)|
 |              | PostgreSQL |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/postgresqlreader/doc/postgresqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/postgresqlwriter/doc/postgresqlwriter.md)|
 |              | DRDS |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/drdsreader/doc/drdsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/drdswriter/doc/drdswriter.md)|

mongodbreader/doc/mongodbreader.md

@@ -8,7 +8,7 @@ MongoDBReader 插件利用 MongoDB 的java客户端MongoClient进行MongoDB的
 MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的JOB程序按照指定的规则对MongoDB中的数据进行分片，并行读取，然后将MongoDB支持的类型通过逐一判断转换成Datax支持的类型。
 
 #### 3 功能说明
-* 该示例从ODPS读一份数据到MongoDB。
+* 该示例从MongoDB读一份数据到ODPS。
 
 	    {
 	    "job": {
@@ -127,6 +127,7 @@ MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的J
 * address： MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】
 * userName：MongoDB的用户名。【选填】
 * userPassword： MongoDB的密码。【选填】
+* authDb: MongoDB认证数据库【选填】
 * collectionName： MonogoDB的集合名。【必填】
 * column：MongoDB的文档列名。【必填】
 * name：Column的名字。【必填】
@@ -147,4 +148,4 @@ MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的J
 
 
 #### 6 性能报告
-#### 7 测试报告
+#### 7 测试报告

mongodbwriter/doc/mongodbwriter.md

@@ -139,7 +139,7 @@ MongoDBWriter通过Datax框架获取Reader生成的数据，然后将Datax支持
 * splitter：特殊分隔符，当且仅当要处理的字符串要用分隔符分隔为字符数组时，才使用这个参数，通过这个参数指定的分隔符，将字符串分隔存储到MongoDB的数组中。【选填】
 * upsertInfo：指定了传输数据时更新的信息。【选填】
 * isUpsert：当设置为true时，表示针对相同的upsertKey做更新操作。【选填】
-* upsertKey：upsertKey指定了没行记录的业务主键。用来做更新时使用。【选填】
+* upsertKey：upsertKey指定了每行记录的业务主键。用来做更新时使用。【选填】
 
 #### 5 类型转换
 
@@ -154,4 +154,4 @@ MongoDBWriter通过Datax框架获取Reader生成的数据，然后将Datax支持
 
 
 #### 6 性能报告
-#### 7 测试报告
+#### 7 测试报告

mysqlreader/doc/mysqlreader.md

@@ -165,7 +165,7 @@ MysqlReader插件实现了从Mysql读取数据。在底层实现上，MysqlReade
 
 	  支持常量配置，用户需要按照Mysql SQL语法格式:
 	  ["id", "\`table\`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"]
-	  id为普通列名，\`table\`为包含保留在的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
+	  id为普通列名，\`table\`为包含保留字的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
 
 	* 必选：是 <br />
 

oceanbasev10reader/pom.xml

@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>oceanbasev10reader</artifactId>
+    <groupId>com.alibaba.datax</groupId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+		<dependency>
+			<groupId>com.alibaba.datax</groupId>
+			<artifactId>plugin-rdbms-util</artifactId>
+			<version>${datax-project-version}</version>
+		</dependency>
+		<dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>5.1.40</version>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.16</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/java</directory>
+                <includes>
+                    <include>**/*.properties</include>
+                </includes>
+            </resource>
+        </resources>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

oceanbasev10reader/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+                <include>plugin_job_template.json</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>oceanbasev10reader-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs/</directory>
+            <includes>
+                <include>*.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/reader/oceanbasev10reader/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/Config.java

@@ -0,0 +1,16 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader;
+
+public interface Config {
+    // queryTimeoutSecond
+    String QUERY_TIMEOUT_SECOND = "memstoreCheckIntervalSecond";
+
+    int DEFAULT_QUERY_TIMEOUT_SECOND = 60 * 60 * 48;// 2天
+
+    // readBatchSize
+    String READ_BATCH_SIZE = "readBatchSize";
+
+    int DEFAULT_READ_BATCH_SIZE = 100000;// 10万
+
+    String RETRY_LIMIT = "retryLimit";
+    int DEFAULT_RETRY_LIMIT = 10;
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/OceanBaseReader.java

@@ -0,0 +1,130 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader;
+
+import java.sql.Connection;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.spi.Reader;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.Constant;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderJob;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderTask;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+
+public class OceanBaseReader extends Reader {
+
+    public static class Job extends Reader.Job {
+        private Configuration originalConfig = null;
+        private ReaderJob readerJob;
+        private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+        @Override
+        public void init() {
+            this.originalConfig = super.getPluginJobConf();
+
+            Integer userConfigedFetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE);
+            if (userConfigedFetchSize != null) {
+                LOG.warn("The [fetchSize] is not recognized, please use readBatchSize instead.");
+            }
+            this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);
+            setDatabaseType(originalConfig);
+            this.readerJob = new ReaderJob();
+            this.readerJob.init(this.originalConfig);
+        }
+
+        @Override
+        public void prepare() {
+            //ObReaderUtils.DATABASE_TYPE获取当前数据库的语法模式
+        }
+
+        @Override
+        public void preCheck() {
+            init();
+            this.readerJob.preCheck(this.originalConfig, ObReaderUtils.databaseType);
+
+        }
+
+        @Override
+        public List<Configuration> split(int adviceNumber) {
+            return this.readerJob.split(this.originalConfig, adviceNumber);
+        }
+
+        @Override
+        public void post() {
+            this.readerJob.post(this.originalConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.readerJob.destroy(this.originalConfig);
+        }
+
+        private void setDatabaseType(Configuration config) {
+            String username = config.getString(Key.USERNAME);
+            String password = config.getString(Key.PASSWORD);
+            List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+            Configuration connConf = Configuration.from(conns.get(0).toString());
+            List<String> jdbcUrls = connConf.getList(Key.JDBC_URL, String.class);
+            String jdbcUrl = jdbcUrls.get(0);
+            if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+                String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+                if (ss.length != 3) {
+                    LOG.warn("unrecognized jdbc url: " + jdbcUrl);
+                    return;
+                }
+                username = ss[1].trim() + ":" + username;
+                jdbcUrl = ss[2];
+            }
+            // Use ob-client to get compatible mode.
+            try {
+                String obJdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:");
+                Connection conn = DBUtil.getConnection(DataBaseType.OceanBase, obJdbcUrl, username, password);
+                String compatibleMode = ObReaderUtils.getCompatibleMode(conn);
+                if (ObReaderUtils.isOracleMode(compatibleMode)) {
+                    ObReaderUtils.compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
+                }
+
+            } catch (Exception e) {
+                LOG.warn("error in get compatible mode, using mysql as default: " + e.getMessage());
+            }
+        }
+    }
+
+    public static class Task extends Reader.Task {
+        private Configuration readerSliceConfig;
+        private ReaderTask commonRdbmsReaderTask;
+        private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+        @Override
+        public void init() {
+            this.readerSliceConfig = super.getPluginJobConf();
+            this.commonRdbmsReaderTask = new ReaderTask(super.getTaskGroupId(), super.getTaskId());
+            this.commonRdbmsReaderTask.init(this.readerSliceConfig);
+
+        }
+
+        @Override
+        public void startRead(RecordSender recordSender) {
+            int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);
+            this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender, super.getTaskPluginCollector(),
+                    fetchSize);
+        }
+
+        @Override
+        public void post() {
+            this.commonRdbmsReaderTask.post(this.readerSliceConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.commonRdbmsReaderTask.destroy(this.readerSliceConfig);
+        }
+    }
+
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderJob.java

@@ -0,0 +1,65 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
+
+import java.util.List;
+
+import com.alibaba.datax.common.constant.CommonConstant;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+import com.alibaba.fastjson.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ReaderJob extends CommonRdbmsReader.Job {
+    private Logger LOG = LoggerFactory.getLogger(OceanBaseReader.Task.class);
+
+    public ReaderJob() {
+        super(ObReaderUtils.databaseType);
+    }
+
+    @Override
+    public void init(Configuration originalConfig) {
+        //将config中的column和table中的关键字进行转义
+        List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
+        ObReaderUtils.escapeDatabaseKeywords(columns);
+        originalConfig.set(Key.COLUMN, columns);
+
+        List<JSONObject> conns = originalConfig.getList(com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, JSONObject.class);
+        for (int i = 0; i < conns.size(); i++) {
+            JSONObject conn = conns.get(i);
+            Configuration connConfig = Configuration.from(conn.toString());
+            List<String> tables = connConfig.getList(Key.TABLE, String.class);
+            ObReaderUtils.escapeDatabaseKeywords(tables);
+            originalConfig.set(String.format("%s[%d].%s", com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, i, Key.TABLE), tables);
+        }
+        super.init(originalConfig);
+    }
+
+    @Override
+    public List<Configuration> split(Configuration originalConfig, int adviceNumber) {
+        List<Configuration> list = super.split(originalConfig, adviceNumber);
+        for (Configuration config : list) {
+            String jdbcUrl = config.getString(Key.JDBC_URL);
+            String obRegionName = getObRegionName(jdbcUrl);
+            config.set(CommonConstant.LOAD_BALANCE_RESOURCE_MARK, obRegionName);
+        }
+        return list;
+    }
+
+    private String getObRegionName(String jdbcUrl) {
+        if (jdbcUrl.startsWith(Constant.OB10_SPLIT_STRING)) {
+            String[] ss = jdbcUrl.split(Constant.OB10_SPLIT_STRING_PATTERN);
+            if (ss.length >= 2) {
+                String tenant = ss[1].trim();
+                String[] sss = tenant.split(":");
+                return sss[0];
+            }
+        }
+        return null;
+    }
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java

@@ -0,0 +1,297 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.statistics.PerfRecord;
+import com.alibaba.datax.common.statistics.PerfTrace;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.reader.Constant;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.RdbmsException;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.Config;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.TaskContext;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+
+public class ReaderTask extends CommonRdbmsReader.Task {
+    private static final Logger LOG = LoggerFactory.getLogger(ReaderTask.class);
+    private int taskGroupId = -1;
+    private int taskId = -1;
+
+    private String username;
+    private String password;
+    private String jdbcUrl;
+    private String mandatoryEncoding;
+    private int queryTimeoutSeconds;// 查询超时 默认48小时
+    private int readBatchSize;
+    private int retryLimit = 0;
+    private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+    private boolean reuseConn = false;
+
+    public ReaderTask(int taskGroupId, int taskId) {
+        super(ObReaderUtils.databaseType, taskGroupId, taskId);
+        this.taskGroupId = taskGroupId;
+        this.taskId = taskId;
+    }
+
+    @Override
+    public void init(Configuration readerSliceConfig) {
+        /* for database connection */
+        username = readerSliceConfig.getString(Key.USERNAME);
+        password = readerSliceConfig.getString(Key.PASSWORD);
+        jdbcUrl = readerSliceConfig.getString(Key.JDBC_URL);
+        queryTimeoutSeconds = readerSliceConfig.getInt(Config.QUERY_TIMEOUT_SECOND,
+                Config.DEFAULT_QUERY_TIMEOUT_SECOND);
+        // ob10的处理
+        if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+            String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+            if (ss.length == 3) {
+                LOG.info("this is ob1_0 jdbc url.");
+                username = ss[1].trim() + ":" + username;
+                jdbcUrl = ss[2];
+            }
+        }
+
+        jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:") + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
+        if (ObReaderUtils.compatibleMode.equals(ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE)) {
+            compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
+        }
+        LOG.info("this is ob1_0 jdbc url. user=" + username + " :url=" + jdbcUrl);
+        mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, "");
+        retryLimit = readerSliceConfig.getInt(Config.RETRY_LIMIT, Config.DEFAULT_RETRY_LIMIT);
+        LOG.info("retryLimit: " + retryLimit);
+    }
+
+    private void buildSavePoint(TaskContext context) {
+        if (!ObReaderUtils.isUserSavePointValid(context)) {
+            LOG.info("user save point is not valid, set to null.");
+            context.setUserSavePoint(null);
+        }
+    }
+
+    /**
+     * 如果isTableMode && table有PK
+     * <p>
+     * 则支持断点续读 (若pk不在原始的columns中,则追加到尾部,但不传给下游)
+     * <p>
+     * 否则,则使用旧模式
+     */
+    @Override
+    public void startRead(Configuration readerSliceConfig, RecordSender recordSender,
+                          TaskPluginCollector taskPluginCollector, int fetchSize) {
+        String querySql = readerSliceConfig.getString(Key.QUERY_SQL);
+        String table = readerSliceConfig.getString(Key.TABLE);
+        PerfTrace.getInstance().addTaskDetails(taskId, table + "," + jdbcUrl);
+        List<String> columns = readerSliceConfig.getList(Key.COLUMN_LIST, String.class);
+        String where = readerSliceConfig.getString(Key.WHERE);
+        boolean weakRead = readerSliceConfig.getBool(Key.WEAK_READ, true); // default true, using weak read
+        String userSavePoint = readerSliceConfig.getString(Key.SAVE_POINT, null);
+        reuseConn = readerSliceConfig.getBool(Key.REUSE_CONN, false);
+        String partitionName = readerSliceConfig.getString(Key.PARTITION_NAME, null);
+        // 从配置文件中取readBatchSize,若无则用默认值
+        readBatchSize = readerSliceConfig.getInt(Config.READ_BATCH_SIZE, Config.DEFAULT_READ_BATCH_SIZE);
+        // 不能少于1万
+        if (readBatchSize < 10000) {
+            readBatchSize = 10000;
+        }
+        TaskContext context = new TaskContext(table, columns, where, fetchSize);
+        context.setQuerySql(querySql);
+        context.setWeakRead(weakRead);
+        context.setCompatibleMode(compatibleMode);
+        if (partitionName != null) {
+            context.setPartitionName(partitionName);
+        }
+        // Add the user save point into the context
+        context.setUserSavePoint(userSavePoint);
+        PerfRecord allPerf = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.RESULT_NEXT_ALL);
+        allPerf.start();
+        boolean isTableMode = readerSliceConfig.getBool(Constant.IS_TABLE_MODE);
+        try {
+            startRead0(isTableMode, context, recordSender, taskPluginCollector);
+        } finally {
+            ObReaderUtils.close(null, null, context.getConn());
+        }
+        allPerf.end(context.getCost());
+        // 目前大盘是依赖这个打印，而之前这个Finish read record是包含了sql查询和result next的全部时间
+        LOG.info("finished read record by Sql: [{}\n] {}.", context.getQuerySql(), jdbcUrl);
+    }
+
+    private void startRead0(boolean isTableMode, TaskContext context, RecordSender recordSender,
+                            TaskPluginCollector taskPluginCollector) {
+        // 不是table模式 直接使用原来的做法
+        if (!isTableMode) {
+            doRead(recordSender, taskPluginCollector, context);
+            return;
+        }
+        // check primary key index
+        Connection conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
+        ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+        context.setConn(conn);
+        try {
+            ObReaderUtils.initIndex(conn, context);
+            ObReaderUtils.matchPkIndexs(conn, context);
+        } catch (Throwable e) {
+            LOG.warn("fetch PkIndexs fail,table=" + context.getTable(), e);
+        }
+        // 如果不是table 且 pk不存在 则仍然使用原来的做法
+        if (context.getPkIndexs() == null) {
+            doRead(recordSender, taskPluginCollector, context);
+            return;
+        }
+
+        // setup the user defined save point
+        buildSavePoint(context);
+
+        // 从这里开始就是 断点续读功能
+        // while(true) {
+        // 正常读 (需 order by pk asc)
+        // 如果遇到失败,分两种情况:
+        // a)已读出记录,则开始走增量读逻辑
+        // b)未读出记录,则走正常读逻辑(仍然需要order by pk asc)
+        // 正常结束 则 break
+        // }
+        context.setReadBatchSize(readBatchSize);
+        String getFirstQuerySql = ObReaderUtils.buildFirstQuerySql(context);
+        String appendQuerySql = ObReaderUtils.buildAppendQuerySql(conn, context);
+        LOG.warn("start table scan key : {}", context.getIndexName() == null ? "primary" : context.getIndexName());
+        context.setQuerySql(getFirstQuerySql);
+        boolean firstQuery = true;
+        // 原来打算firstQuery时 limit 1 减少
+        // 后来经过对比发现其实是多余的,因为:
+        // 1.假如走gmt_modified辅助索引,则直接索引扫描 不需要topN的order by
+        // 2.假如不走辅助索引,而是pk table scan,则减少排序规模并没有好处,因为下一次仍然要排序
+        // 减少这个多余的优化tip 可以让代码更易读
+        int retryCount = 0;
+        while (true) {
+            try {
+                boolean finish = doRead(recordSender, taskPluginCollector, context);
+                if (finish) {
+                    break;
+                }
+            } catch (Throwable e) {
+                if (retryLimit == ++retryCount) {
+                    throw RdbmsException.asQueryException(ObReaderUtils.databaseType, new Exception(e),
+                            context.getQuerySql(), context.getTable(), username);
+                }
+                LOG.error("read fail, retry count " + retryCount + ", sleep 60 second, save point:" +
+                        context.getSavePoint() + ", error: " + e.getMessage());
+                ObReaderUtils.sleep(60000); // sleep 10s
+            }
+            // 假如原来的查询有查出数据,则改成增量查询
+            if (firstQuery && context.getPkIndexs() != null && context.getSavePoint() != null) {
+                context.setQuerySql(appendQuerySql);
+                firstQuery = false;
+            }
+        }
+        DBUtil.closeDBResources(null, context.getConn());
+    }
+
+    private boolean isConnectionAlive(Connection conn) {
+        if (conn == null) {
+            return false;
+        }
+        Statement stmt = null;
+        ResultSet rs = null;
+        String sql = "select 1" + (compatibleMode == ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE ? " from dual" : "");
+        try {
+            stmt = conn.createStatement();
+            rs = stmt.executeQuery(sql);
+            rs.next();
+        } catch (Exception ex) {
+            LOG.info("connection is not alive: " + ex.getMessage());
+            return false;
+        } finally {
+            DBUtil.closeDBResources(rs, stmt, null);
+        }
+        return true;
+    }
+
+    private boolean doRead(RecordSender recordSender, TaskPluginCollector taskPluginCollector, TaskContext context) {
+        LOG.info("exe sql: {}", context.getQuerySql());
+        Connection conn = context.getConn();
+        if (reuseConn && isConnectionAlive(conn)) {
+            LOG.info("connection is alive, will reuse this connection.");
+        } else {
+            LOG.info("Create new connection for reader.");
+            conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
+            ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+            context.setConn(conn);
+        }
+        PreparedStatement ps = null;
+        ResultSet rs = null;
+        PerfRecord perfRecord = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.SQL_QUERY);
+        perfRecord.start();
+        try {
+            ps = conn.prepareStatement(context.getQuerySql(),
+                    ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+            if (context.getPkIndexs() != null && context.getSavePoint() != null) {
+                Record savePoint = context.getSavePoint();
+                List<Column> point = ObReaderUtils.buildPoint(savePoint, context.getPkIndexs());
+                ObReaderUtils.binding(ps, point);
+                if (LOG.isWarnEnabled()) {
+                    List<String> pointForLog = new ArrayList<String>();
+                    for (Column c : point) {
+                        pointForLog.add(c.asString());
+                    }
+                    LOG.warn("{} save point : {}", context.getTable(), StringUtils.join(pointForLog, ','));
+                }
+            }
+            // 打开流式接口
+            ps.setFetchSize(context.getFetchSize());
+            rs = ps.executeQuery();
+            ResultSetMetaData metaData = rs.getMetaData();
+            int columnNumber = metaData.getColumnCount();
+            long lastTime = System.nanoTime();
+            int count = 0;
+            for (; rs.next(); count++) {
+                context.addCost(System.nanoTime() - lastTime);
+                Record row = buildRecord(recordSender, rs, metaData, columnNumber, mandatoryEncoding,
+                        taskPluginCollector);
+                // // 如果第一个record重复了,则不需要发送
+                // if (count == 0 &&
+                // ObReaderUtils.isPkEquals(context.getSavePoint(), row,
+                // context.getPkIndexs())) {
+                // continue;
+                // }
+                // 如果是querySql
+                if (context.getTransferColumnNumber() == -1
+                        || row.getColumnNumber() == context.getTransferColumnNumber()) {
+                    recordSender.sendToWriter(row);
+                } else {
+                    Record newRow = recordSender.createRecord();
+                    for (int i = 0; i < context.getTransferColumnNumber(); i++) {
+                        newRow.addColumn(row.getColumn(i));
+                    }
+                    recordSender.sendToWriter(newRow);
+                }
+                context.setSavePoint(row);
+                lastTime = System.nanoTime();
+            }
+            LOG.info("end of sql: {}, " + count + "rows are read.", context.getQuerySql());
+            return context.getReadBatchSize() <= 0 || count < readBatchSize;
+        } catch (Exception e) {
+            ObReaderUtils.close(null, null, context.getConn());
+            context.setConn(null);
+            LOG.error("reader data fail", e);
+            throw RdbmsException.asQueryException(ObReaderUtils.databaseType, e, context.getQuerySql(),
+                    context.getTable(), username);
+        } finally {
+            perfRecord.end();
+            if (reuseConn) {
+                ObReaderUtils.close(rs, ps, null);
+            } else {
+                ObReaderUtils.close(rs, ps, conn);
+            }
+        }
+    }
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java

@@ -0,0 +1,177 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
+
+import java.sql.Connection;
+import java.util.Collections;
+import java.util.List;
+
+import com.alibaba.datax.common.element.Record;
+
+public class TaskContext {
+    private Connection conn;
+    private final String table;
+    private String indexName;
+    // 辅助索引的字段列表
+    private List<String> secondaryIndexColumns = Collections.emptyList();
+    private String querySql;
+    private final String where;
+    private final int fetchSize;
+    private long readBatchSize = -1;
+    private boolean weakRead = true;
+    private String userSavePoint;
+    private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+
+    public String getPartitionName() {
+        return partitionName;
+    }
+
+    public void setPartitionName(String partitionName) {
+        this.partitionName = partitionName;
+    }
+
+    private String partitionName;
+
+    // 断点续读的保存点
+    private volatile Record savePoint;
+
+    // pk在column中的index,用于绑定变量时从savePoint中读取值
+    // 如果这个值为null,则表示 不是断点续读的场景
+    private int[] pkIndexs;
+
+    private final List<String> columns;
+
+    private String[] pkColumns;
+
+    private long cost;
+
+    private final int transferColumnNumber;
+
+    public TaskContext(String table, List<String> columns, String where, int fetchSize) {
+        super();
+        this.table = table;
+        this.columns = columns;
+        // 针对只有querySql的场景
+        this.transferColumnNumber = columns == null ? -1 : columns.size();
+        this.where = where;
+        this.fetchSize = fetchSize;
+    }
+
+    public Connection getConn() {
+        return conn;
+    }
+
+    public void setConn(Connection conn) {
+        this.conn = conn;
+    }
+
+    public String getIndexName() {
+        return indexName;
+    }
+
+    public void setIndexName(String indexName) {
+        this.indexName = indexName;
+    }
+
+    public List<String> getSecondaryIndexColumns() {
+        return secondaryIndexColumns;
+    }
+
+    public void setSecondaryIndexColumns(List<String> secondaryIndexColumns) {
+        this.secondaryIndexColumns = secondaryIndexColumns;
+    }
+
+    public String getQuerySql() {
+        if (readBatchSize == -1 || ObReaderUtils.isOracleMode(compatibleMode)) {
+            return querySql;
+        } else {
+            return querySql + " limit " + readBatchSize;
+        }
+    }
+
+    public void setQuerySql(String querySql) {
+        this.querySql = querySql;
+    }
+
+    public String getWhere() {
+        return where;
+    }
+
+    public Record getSavePoint() {
+        return savePoint;
+    }
+
+    public void setSavePoint(Record savePoint) {
+        this.savePoint = savePoint;
+    }
+
+    public int[] getPkIndexs() {
+        return pkIndexs;
+    }
+
+    public void setPkIndexs(int[] pkIndexs) {
+        this.pkIndexs = pkIndexs;
+    }
+
+    public List<String> getColumns() {
+        return columns;
+    }
+
+    public String[] getPkColumns() {
+        return pkColumns;
+    }
+
+    public void setPkColumns(String[] pkColumns) {
+        this.pkColumns = pkColumns;
+    }
+
+    public String getTable() {
+        return table;
+    }
+
+    public int getFetchSize() {
+        return fetchSize;
+    }
+
+    public long getCost() {
+        return cost;
+    }
+
+    public void addCost(long cost) {
+        this.cost += cost;
+    }
+
+    public int getTransferColumnNumber() {
+        return transferColumnNumber;
+    }
+
+    public long getReadBatchSize() {
+        return readBatchSize;
+    }
+
+    public void setReadBatchSize(long readBatchSize) {
+        this.readBatchSize = readBatchSize;
+    }
+
+    public boolean getWeakRead() {
+        return weakRead;
+    }
+
+    public void setWeakRead(boolean weakRead) {
+        this.weakRead = weakRead;
+    }
+
+    public String getUserSavePoint() {
+        return userSavePoint;
+    }
+
+    public void setUserSavePoint(String userSavePoint) {
+        this.userSavePoint = userSavePoint;
+    }
+
+    public String getCompatibleMode() {
+        return compatibleMode;
+    }
+
+    public void setCompatibleMode(String compatibleMode) {
+        this.compatibleMode = compatibleMode;
+    }
+}

oceanbasev10reader/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "oceanbasev10reader",
+    "class": "com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader",
+    "description": "read data from oceanbase with SQL interface",
+    "developer": "oceanbase"
+}

oceanbasev10writer/pom.xml

@@ -0,0 +1,126 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>oceanbasev10writer</artifactId>
+
+    <groupId>com.alibaba.datax</groupId>
+    <version>0.0.1-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>plugin-rdbms-util</artifactId>
+            <version>${datax-project-version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+            <version>4.0.4.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+        <!-- 
+        <dependency>
+            <groupId>com.alipay.oceanbase</groupId>
+            <artifactId>oceanbase-partition</artifactId>
+            <version>0.0.5</version>
+        </dependency>
+       -->
+
+        <dependency>
+            <groupId>com.alipay.oceanbase</groupId>
+            <artifactId>oceanbase-connector-java</artifactId>
+            <version>3.2.0</version>
+            <scope>system</scope>
+            <systemPath>${basedir}/src/main/libs/oceanbase-connector-java-3.2.0.jar</systemPath>
+            <exclusions>
+            <exclusion>
+                <groupId>com.alipay.oceanbase</groupId>
+                <artifactId>oceanbase-client</artifactId>
+            </exclusion>
+            </exclusions>
+        </dependency>
+
+		<dependency>
+		  <groupId>log4j</groupId>
+		  <artifactId>log4j</artifactId>
+		  <version>1.2.16</version>
+		</dependency>
+		<dependency>
+      		<groupId>org.json</groupId>
+      		<artifactId>json</artifactId>
+      		<version>20160810</version>
+		</dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/java</directory>
+                <includes>
+                    <include>**/*.properties</include>
+                </includes>
+            </resource>
+        </resources>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

oceanbasev10writer/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+ 				<include>plugin_job_template.json</include>
+ 			</includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>oceanbasev10writer-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs</directory>
+            <includes>
+                <include>*.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/writer/oceanbasev10writer/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/Config.java

@@ -0,0 +1,62 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+public interface Config {
+
+	String MEMSTORE_THRESHOLD = "memstoreThreshold";
+
+	double DEFAULT_MEMSTORE_THRESHOLD = 0.9d;
+
+	String MEMSTORE_CHECK_INTERVAL_SECOND = "memstoreCheckIntervalSecond";
+
+	long DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND = 30;
+
+	int DEFAULT_BATCH_SIZE = 100;
+	int MAX_BATCH_SIZE = 4096;
+
+	String FAIL_TRY_COUNT  = "failTryCount";
+
+	int DEFAULT_FAIL_TRY_COUNT = 10000;
+
+	String WRITER_THREAD_COUNT = "writerThreadCount";
+
+	int DEFAULT_WRITER_THREAD_COUNT = 1;
+
+	String CONCURRENT_WRITE = "concurrentWrite";
+
+	boolean DEFAULT_CONCURRENT_WRITE = true;
+
+	String OB_VERSION = "obVersion";
+	String TIMEOUT = "timeout";
+
+	String PRINT_COST = "printCost";
+	boolean DEFAULT_PRINT_COST = false;
+
+	String COST_BOUND = "costBound";
+	long DEFAULT_COST_BOUND = 20;
+
+	String MAX_ACTIVE_CONNECTION = "maxActiveConnection";
+	int DEFAULT_MAX_ACTIVE_CONNECTION = 2000;
+
+	String WRITER_SUB_TASK_COUNT = "writerSubTaskCount";
+	int DEFAULT_WRITER_SUB_TASK_COUNT = 1;
+	int MAX_WRITER_SUB_TASK_COUNT = 4096;
+
+	String OB_WRITE_MODE = "obWriteMode";
+    String OB_COMPATIBLE_MODE = "obCompatibilityMode";
+    String OB_COMPATIBLE_MODE_ORACLE = "ORACLE";
+    String OB_COMPATIBLE_MODE_MYSQL = "MYSQL";
+
+	String OCJ_GET_CONNECT_TIMEOUT = "ocjGetConnectTimeout";
+	int DEFAULT_OCJ_GET_CONNECT_TIMEOUT = 5000; // 5s
+
+	String OCJ_PROXY_CONNECT_TIMEOUT = "ocjProxyConnectTimeout";
+	int DEFAULT_OCJ_PROXY_CONNECT_TIMEOUT = 5000; // 5s
+
+	String OCJ_CREATE_RESOURCE_TIMEOUT = "ocjCreateResourceTimeout";
+	int DEFAULT_OCJ_CREATE_RESOURCE_TIMEOUT = 60000; // 60s
+
+	String OB_UPDATE_COLUMNS = "obUpdateColumns";
+
+	String USE_PART_CALCULATOR = "usePartCalculator";
+	boolean DEFAULT_USE_PART_CALCULATOR = false;
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java

@@ -0,0 +1,258 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.DbUtils;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 2016-04-07
+ * <p>
+ * 专门针对OceanBase1.0的Writer
+ * 
+ * @author biliang.wbl
+ *
+ */
+public class OceanBaseV10Writer extends Writer {
+	private static DataBaseType DATABASE_TYPE = DataBaseType.OceanBase;
+
+	/**
+	 * Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
+	 * <p/>
+	 * 整个 Writer 执行流程是：
+	 * 
+	 * <pre>
+	 * Job类init-->prepare-->split
+	 * 
+	 *                          Task类init-->prepare-->startWrite-->post-->destroy
+	 *                          Task类init-->prepare-->startWrite-->post-->destroy
+	 * 
+	 *                                                                            Job类post-->destroy
+	 * </pre>
+	 */
+	public static class Job extends Writer.Job {
+		private Configuration originalConfig = null;
+		private CommonRdbmsWriter.Job commonJob;
+		private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常在这里对用户的配置进行校验：是否缺失必填项？有无错误值？有没有无关配置项？...
+		 * 并给出清晰的报错/警告提示。校验通常建议采用静态工具类进行，以保证本类结构清晰。
+		 */
+		@Override
+		public void init() {
+			this.originalConfig = super.getPluginJobConf();
+			checkCompatibleMode(originalConfig);
+			//将config中的column和table中的关键字进行转义
+			List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
+			ObWriterUtils.escapeDatabaseKeywords(columns);
+			originalConfig.set(Key.COLUMN, columns);
+
+			List<JSONObject> conns = originalConfig.getList(Constant.CONN_MARK, JSONObject.class);
+			for (int i = 0; i < conns.size(); i++) {
+				JSONObject conn = conns.get(i);
+				Configuration connConfig = Configuration.from(conn.toString());
+				List<String> tables = connConfig.getList(Key.TABLE, String.class);
+				ObWriterUtils.escapeDatabaseKeywords(tables);
+				originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE), tables);
+			}
+			this.commonJob = new CommonRdbmsWriter.Job(DATABASE_TYPE);
+			this.commonJob.init(this.originalConfig);
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+		 */
+		// 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
+		@Override
+		public void prepare() {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				this.commonJob.prepare(this.originalConfig);
+				final String version = fetchServerVersion(originalConfig);
+				originalConfig.set(Config.OB_VERSION, version);
+			}
+
+			String username = originalConfig.getString(Key.USERNAME);
+			String password = originalConfig.getString(Key.PASSWORD);
+
+			// 获取presql配置，并执行
+			List<String> preSqls = originalConfig.getList(Key.PRE_SQL, String.class);
+			if (preSqls == null || preSqls.size() == 0) {
+				return;
+			}
+
+			List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+			for (Object connConfObject : conns) {
+				Configuration connConf = Configuration.from(connConfObject.toString());
+				// 这里的 jdbcUrl 已经 append 了合适后缀参数
+				String jdbcUrl = connConf.getString(Key.JDBC_URL);
+
+				List<String> tableList = connConf.getList(Key.TABLE, String.class);
+				for (String table : tableList) {
+					List<String> renderedPreSqls = WriterUtil.renderPreOrPostSqls(preSqls, table);
+					if (null != renderedPreSqls && !renderedPreSqls.isEmpty()) {
+						Connection conn = DBUtil.getConnection(DATABASE_TYPE, jdbcUrl, username, password);
+						LOG.info("Begin to execute preSqls:[{}]. context info:{}.",
+								StringUtils.join(renderedPreSqls, ";"), jdbcUrl);
+						WriterUtil.executeSqls(conn, renderedPreSqls, jdbcUrl, DATABASE_TYPE);
+						ObWriterUtils.asyncClose(null, null, conn);
+					}
+				}
+			}
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("After job prepare(), originalConfig now is:[\n{}\n]", originalConfig.toJSON());
+			}
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常采用工具静态类完成把 Job 配置切分成多个 Task 配置的工作。 这里的
+		 * mandatoryNumber 是强制必须切分的份数。
+		 */
+		@Override
+		public List<Configuration> split(int mandatoryNumber) {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				return this.commonJob.split(this.originalConfig, mandatoryNumber);
+			}
+			Configuration simplifiedConf = this.originalConfig;
+
+			List<Configuration> splitResultConfigs = new ArrayList<Configuration>();
+			for (int j = 0; j < mandatoryNumber; j++) {
+				splitResultConfigs.add(simplifiedConf.clone());
+			}
+			return splitResultConfigs;
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之后的后续处理，可以在此处完成。
+		 */
+		@Override
+		public void post() {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				commonJob.post(this.originalConfig);
+				return;
+			}
+			String username = originalConfig.getString(Key.USERNAME);
+			String password = originalConfig.getString(Key.PASSWORD);
+			List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+			List<String> postSqls = originalConfig.getList(Key.POST_SQL, String.class);
+			if (postSqls == null || postSqls.size() == 0) {
+				return;
+			}
+
+			for (Object connConfObject : conns) {
+				Configuration connConf = Configuration.from(connConfObject.toString());
+				String jdbcUrl = connConf.getString(Key.JDBC_URL);
+				List<String> tableList = connConf.getList(Key.TABLE, String.class);
+
+				for (String table : tableList) {
+					List<String> renderedPostSqls = WriterUtil.renderPreOrPostSqls(postSqls, table);
+					if (null != renderedPostSqls && !renderedPostSqls.isEmpty()) {
+						// 说明有 postSql 配置，则此处删除掉
+						Connection conn = DBUtil.getConnection(DATABASE_TYPE, jdbcUrl, username, password);
+						LOG.info("Begin to execute postSqls:[{}]. context info:{}.",
+								StringUtils.join(renderedPostSqls, ";"), jdbcUrl);
+						WriterUtil.executeSqls(conn, renderedPostSqls, jdbcUrl, DATABASE_TYPE);
+						ObWriterUtils.asyncClose(null, null, conn);
+					}
+				}
+			}
+			originalConfig.remove(Key.POST_SQL);
+		}
+		
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常配合 Job 中的 post() 方法一起完成 Job 的资源释放。
+		 */
+		@Override
+		public void destroy() {
+			this.commonJob.destroy(this.originalConfig);
+		}
+
+		private String fetchServerVersion(Configuration config) {
+			final String fetchVersionSql = "show variables like 'version'";
+			return DbUtils.fetchSingleValueWithRetry(config, fetchVersionSql);
+		}
+
+		private void checkCompatibleMode(Configuration configure) {
+			final String fetchCompatibleModeSql = "SHOW VARIABLES LIKE 'ob_compatibility_mode'";
+			String compatibleMode = DbUtils.fetchSingleValueWithRetry(configure, fetchCompatibleModeSql);
+			ObWriterUtils.setCompatibleMode(compatibleMode);
+			configure.set(Config.OB_COMPATIBLE_MODE, compatibleMode);
+		}
+	}
+
+	public static class Task extends Writer.Task {
+		private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+		private Configuration writerSliceConfig;
+		private CommonRdbmsWriter.Task writerTask;
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处通过对 taskConfig 配置的读取，进而初始化一些资源为
+		 * startWrite()做准备。
+		 */
+		@Override
+		public void init() {
+			this.writerSliceConfig = super.getPluginJobConf();
+			int tableNumber = writerSliceConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				// always use concurrentTableWriter
+				this.writerTask = new ConcurrentTableWriterTask(DATABASE_TYPE);
+			} else {
+				throw new RuntimeException("writing to multi-tables is not supported.");
+			}
+			LOG.info("tableNumber:" + tableNumber + ",writerTask Class:" + writerTask.getClass().getName());
+			this.writerTask.init(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task
+		 * 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+		 */
+		@Override
+		public void prepare() {
+			this.writerTask.prepare(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。
+		 */
+		@Override
+		public void startWrite(RecordReceiver recordReceiver) {
+			this.writerTask.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task 中有需要进行数据同步之后的后续处理，可以在此处完成。
+		 */
+		@Override
+		public void post() {
+			this.writerTask.post(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：通常配合Task 中的 post() 方法一起完成 Task 的资源释放。
+		 */
+		@Override
+		public void destroy() {
+			this.writerTask.destroy(this.writerSliceConfig);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ConnHolder.java

@@ -0,0 +1,37 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+
+import java.sql.Connection;
+
+public abstract class ConnHolder {
+
+    protected final Configuration config;
+    protected Connection conn;
+
+    public ConnHolder(Configuration config) {
+        this.config = config;
+    }
+
+    public abstract Connection initConnection();
+
+    public Configuration getConfig() {
+        return config;
+    }
+
+    public Connection getConn() {
+        return conn;
+    }
+
+    public Connection reconnect() {
+        DBUtil.closeDBResources(null, conn);
+        return initConnection();
+    }
+
+    public abstract String getJdbcUrl();
+
+    public abstract String getUserName();
+
+    public abstract void destroy();
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DataBaseWriterBuffer.java

@@ -0,0 +1,101 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 
+ * @author oceanbase
+ *
+ */
+public class DataBaseWriterBuffer {
+	private static final Logger LOG = LoggerFactory.getLogger(DataBaseWriterBuffer.class);
+
+	private final ConnHolder connHolder;
+	private final String dbName;
+	private Map<String, LinkedList<Record>> tableBuffer = new HashMap<String, LinkedList<Record>>();
+	private long lastCheckMemstoreTime;
+
+	public DataBaseWriterBuffer(Configuration config,String jdbcUrl, String userName, String password,String dbName){
+		this.connHolder = new ObClientConnHolder(config, jdbcUrl, userName, password);
+		this.dbName=dbName;
+	}
+	
+	public ConnHolder getConnHolder(){
+		return connHolder;
+	}
+
+	public void initTableBuffer(List<String> tableList) {
+		for (String table : tableList) {
+			tableBuffer.put(table, new LinkedList<Record>());
+		}
+	}
+	
+	public List<String> getTableList(){
+		return new ArrayList<String>(tableBuffer.keySet());
+	}
+
+	public void addRecord(Record record, String tableName) {
+		LinkedList<Record> recordList = tableBuffer.get(tableName);
+		if (recordList == null) {
+			throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR,
+					String.format("The [table] calculated based on the rules does not exist. The calculated [tableName]=%s, [db]=%s. Please check the rules you configured.",
+							tableName, connHolder.getJdbcUrl()));
+		}
+		recordList.add(record);
+	}
+
+	public Map<String, LinkedList<Record>> getTableBuffer() {
+		return tableBuffer;
+	}
+
+	public String getDbName() {
+		return dbName;
+	}
+
+	public long getLastCheckMemstoreTime() {
+		return lastCheckMemstoreTime;
+	}
+
+	public void setLastCheckMemstoreTime(long lastCheckMemstoreTime) {
+		this.lastCheckMemstoreTime = lastCheckMemstoreTime;
+	}
+
+	/**
+	 * 检查当前DB的memstore使用状态
+	 * <p>
+	 * 若超过阈值,则休眠
+	 *
+	 * @param memstoreCheckIntervalSecond
+	 * @param memstoreThreshold
+	 */
+	public synchronized void checkMemstore(long memstoreCheckIntervalSecond, double memstoreThreshold) {
+		long now = System.currentTimeMillis();
+		if (now - getLastCheckMemstoreTime() < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+
+		LOG.debug(String.format("checking memstore usage: lastCheckTime=%d, now=%d, check interval=%d, threshold=%f",
+				getLastCheckMemstoreTime(), now, memstoreCheckIntervalSecond, memstoreThreshold));
+
+		Connection conn = getConnHolder().getConn();
+		while (ObWriterUtils.isMemstoreFull(conn, memstoreThreshold)) {
+			LOG.warn("OB memstore is full,sleep 60 seconds, jdbc=" + getConnHolder().getJdbcUrl()
+					+ ",threshold=" + memstoreThreshold);
+			ObWriterUtils.sleep(60000);
+		}
+		setLastCheckMemstoreTime(now);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OBDataSourceV10.java

@@ -0,0 +1,190 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alipay.oceanbase.obproxy.datasource.ObGroupDataSource;
+import com.alipay.oceanbase.obproxy.exception.ConnectionPropertiesNotSupportedException;
+import com.alipay.oceanbase.obproxy.util.StringParser.IllegalFormatException;
+import com.google.common.collect.Maps;
+
+public class OBDataSourceV10 {
+	private static final Logger LOG = LoggerFactory.getLogger(OBDataSourceV10.class);
+
+	private static final Map<String, DataSourceHolder> dataSources = Maps.newHashMap();
+
+	private static int ocjGetConnectionTimeout = 0;
+	private static int ocjGlobalProxyroGetConnectionTimeout = 0;
+	private static int ocjMaxWaitOfCreateClusterResourceMs = 0;
+
+	private static Configuration taskConfig;
+
+	public static String genKey(String fullUserName, String dbName) {
+		//username@tenantName#clusterName/dbName
+		return fullUserName + "/" + dbName;
+	}
+	
+	public static synchronized void init(Configuration configuration,
+			final String fullUsername,
+			final String password,
+			final String dbName) {
+		taskConfig = configuration;
+		final String rsUrl = "";
+		final String dataSourceKey = genKey(fullUsername, dbName);
+		final int maxActiveConnection = configuration.getInt(Config.MAX_ACTIVE_CONNECTION, Config.DEFAULT_MAX_ACTIVE_CONNECTION);
+		if (dataSources.containsKey(dataSourceKey)) {
+			dataSources.get(dataSourceKey).increseRefercnce();
+		} else {
+			long timeout = configuration.getInt(Config.TIMEOUT, 30);
+			if (timeout < 30) {
+				timeout = 30;
+			}
+			if (ocjGetConnectionTimeout == 0) {
+				ocjGetConnectionTimeout = configuration.getInt(Config.OCJ_GET_CONNECT_TIMEOUT,
+						Config.DEFAULT_OCJ_GET_CONNECT_TIMEOUT);
+				ocjGlobalProxyroGetConnectionTimeout = configuration.getInt(Config.OCJ_PROXY_CONNECT_TIMEOUT,
+						Config.DEFAULT_OCJ_PROXY_CONNECT_TIMEOUT);
+				ocjMaxWaitOfCreateClusterResourceMs = configuration.getInt(Config.OCJ_CREATE_RESOURCE_TIMEOUT,
+						Config.DEFAULT_OCJ_CREATE_RESOURCE_TIMEOUT);
+
+				LOG.info(String.format("initializing OCJ with ocjGetConnectionTimeout=%d, " +
+								"ocjGlobalProxyroGetConnectionTimeout=%d, ocjMaxWaitOfCreateClusterResourceMs=%d",
+						ocjGetConnectionTimeout, ocjGlobalProxyroGetConnectionTimeout, ocjMaxWaitOfCreateClusterResourceMs));
+			}
+			DataSourceHolder holder = null;
+			try {
+				holder = new DataSourceHolder(rsUrl, fullUsername, password, dbName, maxActiveConnection, timeout);
+				dataSources.put(dataSourceKey, holder);
+			} catch (ConnectionPropertiesNotSupportedException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (IllegalArgumentException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (IllegalFormatException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (SQLException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			}
+		}
+	}
+
+	public static synchronized void destory(final String dataSourceKey){
+		DataSourceHolder holder = dataSources.get(dataSourceKey);
+		holder.decreaseReference();
+		if (holder.canClose()) {
+			dataSources.remove(dataSourceKey);
+			holder.close();
+			LOG.info(String.format("close datasource success [%s]", dataSourceKey));
+		}
+	}
+	
+	public static Connection getConnection(final String url) {
+		Connection conn = null;
+		try {
+			conn = dataSources.get(url).getconnection();
+		} catch (SQLException e) {
+			e.printStackTrace();
+		}
+		return conn;
+	}
+
+	private static Map<String, String> buildJdbcProperty() {
+	    Map<String, String> property = new HashMap<String, String>();
+	    property.put("useServerPrepStmts", "false");
+	    property.put("characterEncoding", "UTF-8");
+	    property.put("useLocalSessionState", "false");
+	    property.put("rewriteBatchedStatements", "true");
+	    property.put("socketTimeout", "25000");
+
+	    return property;
+	}
+
+	private static class DataSourceHolder {
+		private volatile int reference;
+		private final ObGroupDataSource groupDataSource;
+		public static final Map<String, String> jdbcProperty = buildJdbcProperty();;
+
+		public DataSourceHolder(final String rsUrl,
+				final String fullUsername,
+				final String password,
+				final String dbName,
+				final int maxActive,
+				final long timeout) throws ConnectionPropertiesNotSupportedException, IllegalFormatException, IllegalArgumentException, SQLException {
+			this.reference = 1;
+			this.groupDataSource = new ObGroupDataSource();
+		    this.groupDataSource.setUrl(rsUrl);
+		    this.groupDataSource.setFullUsername(fullUsername);
+		    this.groupDataSource.setPassword(password);
+		    this.groupDataSource.setDatabase(dbName);
+		    this.groupDataSource.setConnectionProperties(jdbcProperty);
+		    this.groupDataSource.setGetConnectionTimeout(ocjGetConnectionTimeout);
+		    this.groupDataSource.setGlobalProxyroGetConnectionTimeout(ocjGlobalProxyroGetConnectionTimeout);
+		    this.groupDataSource.setMaxWaitOfCreateClusterResourceMs(ocjMaxWaitOfCreateClusterResourceMs);
+		    this.groupDataSource.setMaxActive(maxActive);
+		    this.groupDataSource.setGlobalSlowQueryThresholdUs(3000000);      // 3s, sql with response time more than 3s will be logged
+		    this.groupDataSource.setGlobalCleanLogFileEnabled(true);          // enable log cleanup
+		    this.groupDataSource.setGlobalLogFileSizeThreshold(17179869184L); // 16G, log file total size
+		    this.groupDataSource.setGlobalCleanLogFileInterval(10000);        // 10s, check interval
+		    this.groupDataSource.setInitialSize(1);
+
+			List<String> initSqls = new ArrayList<String>();
+			if (taskConfig != null) {
+				List<String> sessionConfig = taskConfig.getList(Key.SESSION, new ArrayList(), String.class);
+				if (sessionConfig != null || sessionConfig.size() > 0) {
+					initSqls.addAll(sessionConfig);
+				}
+			}
+            // set up for writing timestamp columns
+			if (ObWriterUtils.isOracleMode()) {
+				initSqls.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS';");
+				initSqls.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF';");
+				initSqls.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD';");
+			}
+
+			this.groupDataSource.setConnectionInitSqls(initSqls);
+
+		    this.groupDataSource.init();
+			// this.groupDataSource;
+		    LOG.info("Create GroupDataSource rsUrl=[{}], fullUserName=[{}], dbName=[{}], getConnectionTimeout= {}ms, maxActive={}",
+		    		rsUrl, fullUsername, dbName, 5000, maxActive);
+		}
+		
+		public Connection getconnection() throws SQLException {
+			return groupDataSource.getConnection();
+		}
+
+		public synchronized void increseRefercnce() {
+			this.reference++;
+		}
+
+		public synchronized void decreaseReference() {
+			this.reference--;
+		}
+
+		public synchronized boolean canClose() {
+			return reference == 0;
+		}
+
+		public synchronized void close() {
+			if (this.canClose()) {
+				groupDataSource.destroy();
+			}
+		}
+	}
+
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java

@@ -0,0 +1,55 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+
+/**
+ * wrap oceanbase java client
+ * @author oceanbase
+ */
+
+public class OCJConnHolder extends ConnHolder {
+	private ServerConnectInfo connectInfo;
+	private String dataSourceKey;
+
+	public OCJConnHolder (Configuration config, ServerConnectInfo connInfo) {
+		super(config);
+		this.connectInfo = connInfo;
+		this.dataSourceKey = OBDataSourceV10.genKey(connectInfo.getFullUserName(), connectInfo.databaseName);
+		OBDataSourceV10.init(config, connectInfo.getFullUserName(), connectInfo.password, connectInfo.databaseName);
+	}
+
+	@Override
+	public Connection initConnection() {
+		conn = OBDataSourceV10.getConnection(dataSourceKey);
+		return conn;
+	}
+
+	@Override
+	public Connection reconnect() {
+		DBUtil.closeDBResources(null, conn);
+		return initConnection();
+	}
+
+	@Override
+	public Connection getConn() {
+		return conn;
+	}
+
+	@Override
+	public String getJdbcUrl() {
+		return connectInfo.jdbcUrl;
+	}
+
+	@Override
+	public String getUserName() {
+		return connectInfo.userName;
+	}
+	
+	public void destroy() {
+		OBDataSourceV10.destory(this.dataSourceKey);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java

@@ -0,0 +1,63 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+/**
+ * 数据库连接代理对象,负责创建连接，重新连接
+ *
+ * @author oceanbase
+ *
+ */
+public class ObClientConnHolder extends ConnHolder {
+	private final String jdbcUrl;
+	private final String userName;
+	private final String password;
+
+	public ObClientConnHolder(Configuration config, String jdbcUrl, String userName, String password) {
+		super(config);
+		this.jdbcUrl = jdbcUrl;
+		this.userName = userName;
+		this.password = password;
+	}
+
+	// Connect to ob with obclient and obproxy
+	@Override
+	public Connection initConnection() {
+		String BASIC_MESSAGE = String.format("jdbcUrl:[%s]", this.jdbcUrl);
+		DataBaseType dbType = DataBaseType.OceanBase;
+		if (ObWriterUtils.isOracleMode()) {
+		    // set up for writing timestamp columns
+		    List<String> sessionConfig = config.getList(Key.SESSION, new ArrayList<String>(), String.class);
+			sessionConfig.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS'");
+			sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF'");
+			sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD'");
+		    config.set(Key.SESSION, sessionConfig);
+		}
+		conn = DBUtil.getConnection(dbType, jdbcUrl, userName, password);
+		DBUtil.dealWithSessionConfig(conn, config, dbType, BASIC_MESSAGE);
+		return conn;
+	}
+
+	@Override
+	public String getJdbcUrl() {
+		return jdbcUrl;
+	}
+
+	@Override
+	public String getUserName() {
+		return userName;
+	}
+
+	@Override
+	public void destroy() {
+		DBUtil.closeDBResources(null, conn);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObDataSourceErrorCode.java

@@ -0,0 +1,31 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import com.alibaba.datax.common.spi.ErrorCode;
+
+public enum ObDataSourceErrorCode implements ErrorCode {
+	DESC("ObDataSourceError code","connect error");
+
+    private final String code;
+    private final String describe;
+
+    private ObDataSourceErrorCode(String code, String describe) {
+        this.code = code;
+        this.describe = describe;
+    }
+
+    @Override
+    public String getCode() {
+        return this.code;
+    }
+
+    @Override
+    public String getDescription() {
+        return this.describe;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Code:[%s], Describe:[%s]. ", this.code,
+                this.describe);
+    }
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java

@@ -0,0 +1,79 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ServerConnectInfo {
+	
+	public String clusterName;
+	public String tenantName;
+	public String userName;
+	public String password;
+	public String databaseName;
+	public String ipPort;
+	public String jdbcUrl;
+
+	public ServerConnectInfo(final String jdbcUrl, final String username, final String password) {
+		if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+			String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+			if (ss.length != 3) {
+				throw new RuntimeException("jdbc url format is not correct: " + jdbcUrl);
+			}
+			this.userName = username;
+			this.clusterName = ss[1].trim().split(":")[0];
+			this.tenantName  = ss[1].trim().split(":")[1];
+			this.jdbcUrl = ss[2].replace("jdbc:mysql:", "jdbc:oceanbase:");
+		} else {
+			this.jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:");
+			if (username.contains("@") && username.contains("#")) {
+				this.userName = username.substring(0, username.indexOf("@"));
+				this.tenantName = username.substring(username.indexOf("@") + 1, username.indexOf("#"));
+				this.clusterName = username.substring(username.indexOf("#") + 1);
+			} else if (username.contains(":")) {
+				String[] config = username.split(":");
+				if (config.length != 3) {
+					throw new RuntimeException ("username format is not correct: " + username);
+				}
+				this.clusterName = config[0];
+				this.tenantName = config[1];
+				this.userName = config[2];
+			} else {
+				this.clusterName = null;
+				this.tenantName = null;
+				this.userName = username;
+			}
+		}
+
+		this.password = password;
+		parseJdbcUrl(jdbcUrl);
+	}
+
+	private void parseJdbcUrl(final String jdbcUrl) {
+		Pattern pattern = Pattern.compile("//([\\w\\.\\-]+:\\d+)/([\\w-]+)\\?");
+		Matcher matcher = pattern.matcher(jdbcUrl);
+		if (matcher.find()) {
+			String ipPort = matcher.group(1);
+			String dbName = matcher.group(2);
+			this.ipPort = ipPort;
+			this.databaseName = dbName;
+		} else {
+			throw new RuntimeException("Invalid argument:" + jdbcUrl);
+		}
+	}
+
+	public String toString() {
+		StringBuffer strBuffer = new StringBuffer();
+		return strBuffer.append("clusterName:").append(clusterName).append(", tenantName:").append(tenantName)
+				.append(", userName:").append(userName).append(", databaseName:").append(databaseName)
+				.append(", ipPort:").append(ipPort).append(", jdbcUrl:").append(jdbcUrl).toString();
+	}
+
+	public String getFullUserName() {
+		StringBuilder builder = new StringBuilder(userName);
+		if (tenantName != null && clusterName != null) {
+			builder.append("@").append(tenantName).append("#").append(clusterName);
+		}
+
+		return builder.toString();
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ColumnMetaCache.java

@@ -0,0 +1,41 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Triple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+
+public class ColumnMetaCache {
+    private static final Logger LOG = LoggerFactory.getLogger(ColumnMetaCache.class);
+	
+	private static String tableName;
+	private static Triple<List<String>, List<Integer>, List<String>> columnMeta = null;
+
+	public ColumnMetaCache() {
+		
+	}
+	
+	public static void init(Connection connection, final String tableName, final List<String> columns) throws SQLException {
+		if (columnMeta == null) {
+			synchronized(ColumnMetaCache.class) {
+				ColumnMetaCache.tableName = tableName;
+				if (columnMeta == null) {
+					columnMeta = DBUtil.getColumnMetaData(connection,
+            				tableName, StringUtils.join(columns, ","));
+					LOG.info("fetch columnMeta of table {} success", tableName);
+				}
+			}
+		}
+	}
+	
+	public static Triple<List<String>, List<Integer>, List<String>> getColumnMeta() {
+		return columnMeta;
+	}
+	
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java

@@ -0,0 +1,522 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import com.alipay.oceanbase.obproxy.data.TableEntryKey;
+import com.alipay.oceanbase.obproxy.util.ObPartitionIdCalculator;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+//import java.sql.PreparedStatement;
+
+public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
+    private static final Logger LOG = LoggerFactory.getLogger(ConcurrentTableWriterTask.class);
+
+	// memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入
+	private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD;
+	// memstore检查的间隔
+	private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND;
+	// 最后一次检查
+	private long lastCheckMemstoreTime;
+    
+    private static AtomicLong totalTask = new AtomicLong(0);
+    private long taskId = -1;
+    
+    private AtomicBoolean isMemStoreFull = new AtomicBoolean(false);
+    private ConnHolder checkConnHolder;
+
+    public ConcurrentTableWriterTask(DataBaseType dataBaseType) {
+		super(dataBaseType);
+		taskId = totalTask.getAndIncrement();
+	}
+
+	private ObPartitionIdCalculator partCalculator = null;
+
+	private HashMap<Long, List<Record>> groupInsertValues;
+	List<Record> unknownPartRecords = new ArrayList<Record>();
+//	private List<Record> unknownPartRecords;
+	private List<Integer> partitionKeyIndexes;
+	
+	private ConcurrentTableWriter concurrentWriter = null;
+	
+	private ConnHolder connHolder;
+	
+	private boolean allTaskInQueue = false;
+	
+	private Lock lock = new ReentrantLock();
+	private Condition condition = lock.newCondition();
+	
+	private long startTime;
+	private boolean isOb2 = false;
+	private String obWriteMode = "update";
+	private boolean isOracleCompatibleMode = false;
+	private String obUpdateColumns = null;
+	private List<Pair<String, int[]>> deleteColPos;
+	private String dbName;
+
+	@Override
+	public void init(Configuration config) {
+		super.init(config);
+		// OceanBase 所有操作都是 insert into on duplicate key update 模式
+		// writeMode应该使用enum来定义
+		this.writeMode = "update";
+        obWriteMode = config.getString(Config.OB_WRITE_MODE, "update");
+		ServerConnectInfo connectInfo = new ServerConnectInfo(jdbcUrl, username, password);
+		dbName = connectInfo.databaseName;
+		//init check memstore
+		this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD);
+		this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND,
+				Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND);
+        this.isOracleCompatibleMode = ObWriterUtils.isOracleMode();
+
+        LOG.info("configure url is unavailable, use obclient for connections.");
+        this.checkConnHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl,
+				connectInfo.getFullUserName(), connectInfo.password);
+        this.connHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl,
+				connectInfo.getFullUserName(), connectInfo.password);
+        checkConnHolder.initConnection();
+        if (isOracleCompatibleMode) {
+			connectInfo.databaseName = connectInfo.databaseName.toUpperCase();
+			//在转义的情况下不翻译
+			if (!(table.startsWith("\"") && table.endsWith("\""))) {
+				table = table.toUpperCase();
+			}
+
+			LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s",
+					connectInfo.databaseName, table));
+        }
+
+        if (config.getBool(Config.USE_PART_CALCULATOR, Config.DEFAULT_USE_PART_CALCULATOR)) {
+            initPartCalculator(connectInfo);
+        } else {
+            LOG.info("Disable partition calculation feature.");
+        }
+
+		obUpdateColumns = config.getString(Config.OB_UPDATE_COLUMNS, null);
+		groupInsertValues = new HashMap<Long, List<Record>>();
+		partitionKeyIndexes = new ArrayList<Integer>();
+		rewriteSql();
+
+		if (null == concurrentWriter) {
+			concurrentWriter = new ConcurrentTableWriter(config, connectInfo, writeRecordSql);
+			allTaskInQueue = false;
+		}
+
+		String version = config.getString(Config.OB_VERSION);
+		int pIdx = version.lastIndexOf('.');
+		if ((Float.valueOf(version.substring(0, pIdx)) >= 2.1f)) {
+			isOb2 = true;
+		}
+	}
+
+	private void initPartCalculator(ServerConnectInfo connectInfo) {
+		int retry = 0;
+		LOG.info(String.format("create tableEntryKey with clusterName %s, tenantName %s, databaseName %s, tableName %s",
+				connectInfo.clusterName, connectInfo.tenantName, connectInfo.databaseName, table));
+		TableEntryKey tableEntryKey = new TableEntryKey(connectInfo.clusterName, connectInfo.tenantName,
+				connectInfo.databaseName, table);
+		do {
+			try {
+				if (retry > 0) {
+					int sleep = retry > 8 ? 500 : (1 << retry);
+					TimeUnit.SECONDS.sleep(sleep);
+					LOG.info("retry create new part calculator, the {} times", retry);
+				}
+				LOG.info("create partCalculator with address: " + connectInfo.ipPort);
+				partCalculator = new ObPartitionIdCalculator(connectInfo.ipPort, tableEntryKey);
+			} catch (Exception ex) {
+				++retry;
+				LOG.warn("create new part calculator failed, retry {}: {}", retry, ex.getMessage());
+			}
+		} while (partCalculator == null && retry < 3); // try 3 times
+	}
+
+	public boolean isFinished() {
+		return allTaskInQueue && concurrentWriter.checkFinish();
+	}
+	
+	public boolean allTaskInQueue() {
+		return allTaskInQueue;
+	}
+	
+	public void setPutAllTaskInQueue() {
+		this.allTaskInQueue = true;
+		LOG.info("ConcurrentTableWriter has put all task in queue, queueSize = {},  total = {}, finished = {}",
+				concurrentWriter.getTaskQueueSize(),
+				concurrentWriter.getTotalTaskCount(),
+				concurrentWriter.getFinishTaskCount());
+	}
+	
+	private void rewriteSql() {
+		Connection conn = connHolder.initConnection();
+		if (isOracleCompatibleMode && obWriteMode.equalsIgnoreCase("update")) {
+			// change obWriteMode to insert so the insert statement will be generated.
+			obWriteMode = "insert";
+			deleteColPos = ObWriterUtils.buildDeleteSql(conn, dbName, table, columns);
+		}
+		this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns);
+		LOG.info("writeRecordSql :{}", this.writeRecordSql);
+	}
+	
+	public void prepare(Configuration writerSliceConfig) {
+		super.prepare(writerSliceConfig);
+		calPartitionKeyIndex(partitionKeyIndexes);
+		concurrentWriter.start();
+	}
+
+	private void calPartitionKeyIndex(List<Integer> partKeyIndexes) {
+		partKeyIndexes.clear();
+		if (null == partCalculator) {
+			LOG.error("partCalculator is null");
+			return;
+		}
+		for (int i = 0; i < columns.size(); ++i) {
+			if (partCalculator.isPartitionKeyColumn(columns.get(i))) {
+			    LOG.info(columns.get(i) + " is partition key.");
+				partKeyIndexes.add(i);
+			}
+		}
+	}
+
+	private Long calPartitionId(List<Integer> partKeyIndexes, Record record) {
+	    if (partCalculator == null) {
+	        return null;
+	    }
+		for (Integer i : partKeyIndexes) {
+			partCalculator.addColumn(columns.get(i), record.getColumn(i).asString());
+		}
+		return partCalculator.calculate();
+	}
+	
+	@Override
+    public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) {
+        this.taskPluginCollector = taskPluginCollector;
+
+        // 用于写入数据的时候的类型根据目的表字段类型转换
+        int retryTimes = 0;
+        boolean needRetry = false;
+        do {
+        	try {
+        		if (retryTimes > 0) {
+        			TimeUnit.SECONDS.sleep((1 << retryTimes));
+    				DBUtil.closeDBResources(null, connection);
+        			connection = DBUtil.getConnection(dataBaseType, jdbcUrl, username, password);
+        			LOG.warn("getColumnMetaData of table {} failed, retry the {} times ...", this.table, retryTimes);
+        		}
+        		ColumnMetaCache.init(connection, this.table, this.columns);
+        		this.resultSetMetaData = ColumnMetaCache.getColumnMeta();
+            	needRetry = false;
+        	} catch (SQLException e) {
+        		needRetry = true;
+        		++retryTimes;
+        		e.printStackTrace();
+        		LOG.warn("fetch column meta of [{}] failed..., retry {} times", this.table, retryTimes);
+        	} catch (InterruptedException e) {
+				LOG.warn("startWriteWithConnection interrupt, ignored");
+			} finally {
+        	}
+        } while (needRetry && retryTimes < 100);
+
+        try {
+            Record record;
+            startTime = System.currentTimeMillis();
+            while ((record = recordReceiver.getFromReader()) != null) {
+                if (record.getColumnNumber() != this.columnNumber) {
+                    // 源头读取字段列数与目的表字段写入列数不相等，直接报错
+                	LOG.error("column not equal {} != {}, record = {}",
+                			this.columnNumber, record.getColumnNumber(), record.toString());
+                    throw DataXException
+                            .asDataXException(
+                                    DBUtilErrorCode.CONF_ERROR,
+                                    String.format("Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: %d. ErrorCode: %d",
+                                            record.getColumnNumber(),
+                                            this.columnNumber));
+                }
+                addRecordToCache(record);
+            }
+            addLeftRecords();
+            waitTaskFinish();
+        } catch (Exception e) {
+            throw DataXException.asDataXException(
+                    DBUtilErrorCode.WRITE_DATA_ERROR, e);
+        } finally {
+            DBUtil.closeDBResources(null, null, connection);
+        }
+    }
+
+	public PreparedStatement fillStatement(PreparedStatement preparedStatement, Record record)
+			throws SQLException {
+		return fillPreparedStatement(preparedStatement, record);
+	}
+
+    public PreparedStatement fillStatementIndex(PreparedStatement preparedStatement,
+                                                int prepIdx, int columnIndex, Column column) throws SQLException {
+        int columnSqltype = this.resultSetMetaData.getMiddle().get(columnIndex);
+        String typeName = this.resultSetMetaData.getRight().get(columnIndex);
+        return fillPreparedStatementColumnType(preparedStatement, prepIdx, columnSqltype, typeName, column);
+    }
+
+    public void collectDirtyRecord(Record record, SQLException e) {
+		taskPluginCollector.collectDirtyRecord(record, e);
+	}
+
+	public void insertOneRecord(Connection connection, List<Record> buffer) {
+		doOneInsert(connection, buffer);
+	}
+
+	private void addLeftRecords() {
+		//不需要刷新Cache，已经是最后一批数据了
+		for (List<Record> groupValues : groupInsertValues.values()) {
+			if (groupValues.size() > 0 ) {
+				addRecordsToWriteQueue(groupValues);
+			}
+		}
+		if (unknownPartRecords.size() > 0) {
+			addRecordsToWriteQueue(unknownPartRecords);
+		}
+	}
+	
+	private void addRecordToCache(final Record record) {
+		Long partId =null;
+		try {
+			partId = calPartitionId(partitionKeyIndexes, record);
+		} catch (Exception e1) {
+		    LOG.warn("fail to get partition id: " + e1.getMessage() + ", record: " + record);
+		}
+
+        if (partId == null && isOb2) {
+            LOG.debug("fail to calculate parition id, just put into the default buffer.");
+            partId = Long.MAX_VALUE;
+        }
+
+		if (partId != null) {
+			List<Record> groupValues = groupInsertValues.get(partId);
+			if (groupValues == null) {
+				groupValues = new ArrayList<Record>(batchSize);
+				groupInsertValues.put(partId, groupValues);
+			}
+			groupValues.add(record);
+			if (groupValues.size() >= batchSize) {
+				groupValues = addRecordsToWriteQueue(groupValues);
+				groupInsertValues.put(partId, groupValues);
+			}
+		} else {
+			LOG.debug("add unknown part record {}", record);
+			unknownPartRecords.add(record);
+			if (unknownPartRecords.size() >= batchSize) {
+				unknownPartRecords = addRecordsToWriteQueue(unknownPartRecords);
+			}
+
+		}
+	}
+
+	/**
+	 *
+	 * @param records
+	 * @return 返回一个新的Cache用于存储接下来的数据
+	 */
+	private List<Record> addRecordsToWriteQueue(List<Record> records) {
+		int i = 0;
+		while (true) {
+			if (i > 0) {
+				LOG.info("retry add batch record the {} times", i);
+			}
+			try {
+				concurrentWriter.addBatchRecords(records);
+				break;
+			} catch (InterruptedException e) {
+				i++;
+				LOG.info("Concurrent table writer is interrupted");
+			}
+		}
+		return new ArrayList<Record>(batchSize);
+	}
+	private void checkMemStore() {
+		Connection checkConn = checkConnHolder.reconnect();
+		long now = System.currentTimeMillis();
+		if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+		boolean isFull = ObWriterUtils.isMemstoreFull(checkConn, memstoreThreshold);
+		this.isMemStoreFull.set(isFull);
+		if (isFull) {
+			LOG.warn("OB memstore is full,sleep 30 seconds, threshold=" + memstoreThreshold);
+		}
+		lastCheckMemstoreTime = now;
+	}
+	
+	public boolean isMemStoreFull() {
+		return isMemStoreFull.get();
+	}
+	
+	public void printEveryTime() {
+		long cost = System.currentTimeMillis() - startTime;
+		if (cost > 10000) { //10s
+			print();
+			startTime = System.currentTimeMillis();
+		}
+	}
+	
+	public void print() {
+		LOG.debug("Statistic total task {}, finished {}, queue Size {}",
+				concurrentWriter.getTotalTaskCount(),
+				concurrentWriter.getFinishTaskCount(),
+				concurrentWriter.getTaskQueueSize());
+		concurrentWriter.printStatistics();
+	}
+	
+	public void waitTaskFinish() {
+		setPutAllTaskInQueue();
+		lock.lock();
+		try {
+			while (!concurrentWriter.checkFinish()) {
+				condition.await(15, TimeUnit.SECONDS);
+				print();
+				checkMemStore();
+			}
+		} catch (InterruptedException e) {
+			LOG.warn("Concurrent table writer wait task finish interrupt");
+		} finally {
+			lock.unlock();
+		}
+		LOG.debug("wait all InsertTask finished ...");
+	}
+	
+	public void singalTaskFinish() {
+		lock.lock();
+		condition.signal();
+		lock.unlock();
+	}
+	
+	@Override
+	public void destroy(Configuration writerSliceConfig) {
+	   if(concurrentWriter!=null) {
+		concurrentWriter.destory();
+		}
+		// 把本级持有的conn关闭掉
+		DBUtil.closeDBResources(null, connHolder.getConn());
+        DBUtil.closeDBResources(null, checkConnHolder.getConn());
+        checkConnHolder.destroy();
+		super.destroy(writerSliceConfig);
+	}
+	
+	public class ConcurrentTableWriter {
+		private BlockingQueue<List<Record>> queue;
+		private List<InsertTask> insertTasks;
+		private Configuration config;
+		private ServerConnectInfo connectInfo;
+		private String rewriteRecordSql;
+		private AtomicLong totalTaskCount;
+		private AtomicLong finishTaskCount;
+		private final int threadCount;
+
+		public ConcurrentTableWriter(Configuration config, ServerConnectInfo connInfo, String rewriteRecordSql) {
+			threadCount = config.getInt(Config.WRITER_THREAD_COUNT, Config.DEFAULT_WRITER_THREAD_COUNT);
+			queue = new LinkedBlockingQueue<List<Record>>(threadCount << 1);
+			insertTasks = new ArrayList<InsertTask>(threadCount);
+			this.config = config;
+			this.connectInfo = connInfo;
+			this.rewriteRecordSql = rewriteRecordSql;
+			this.totalTaskCount = new AtomicLong(0);
+			this.finishTaskCount = new AtomicLong(0);
+		}
+		
+		public long getTotalTaskCount() {
+			return totalTaskCount.get();
+		}
+		
+		public long getFinishTaskCount() {
+			return finishTaskCount.get();
+		}
+		
+		public int getTaskQueueSize() {
+			return queue.size();
+		}
+		
+		public void increFinishCount() {
+			finishTaskCount.incrementAndGet();
+		}
+		
+		//should check after put all the task in the queue
+		public boolean checkFinish() {
+			long finishCount = finishTaskCount.get();
+			long totalCount = totalTaskCount.get();
+			return finishCount == totalCount;
+		}
+		
+		public synchronized void start() {
+			for (int i = 0; i < threadCount; ++i) {
+			    LOG.info("start {} insert task.", (i+1));
+				InsertTask insertTask = new InsertTask(taskId, queue, config, connectInfo, rewriteRecordSql, deleteColPos);
+				insertTask.setWriterTask(ConcurrentTableWriterTask.this);
+				insertTask.setWriter(this);
+				insertTasks.add(insertTask);
+			}
+			WriterThreadPool.executeBatch(insertTasks);
+		}
+		
+		public void printStatistics() {
+			long insertTotalCost = 0;
+			long insertTotalCount = 0;
+			for (InsertTask task: insertTasks) {
+				insertTotalCost += task.getTotalCost();
+				insertTotalCount += task.getInsertCount();
+			}
+			long avgCost = 0;
+			if (insertTotalCount != 0) {
+				avgCost = insertTotalCost / insertTotalCount;
+			}
+			ConcurrentTableWriterTask.LOG.debug("Insert {} times, totalCost {} ms, average {} ms",
+					insertTotalCount, insertTotalCost, avgCost);
+		}
+
+		public void addBatchRecords(final List<Record> records) throws InterruptedException {
+			boolean isSucc = false;
+			while (!isSucc) {
+				isSucc = queue.offer(records, 5, TimeUnit.SECONDS);
+				checkMemStore();
+			}
+			totalTaskCount.incrementAndGet();
+		}
+		
+		public synchronized void destory() {
+			if (insertTasks != null) {
+				for(InsertTask task : insertTasks) {
+					task.setStop();
+				}
+				for(InsertTask task: insertTasks) {
+					task.destroy();
+				}
+			}
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java

@@ -0,0 +1,286 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask.ConcurrentTableWriter;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+public class InsertTask implements Runnable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(InsertTask.class);
+
+	private ConcurrentTableWriterTask writerTask;
+	private ConcurrentTableWriter writer;
+
+	private String writeRecordSql;
+	private long totalCost = 0;
+	private long insertCount = 0;
+
+	private Queue<List<Record>> queue;
+	private boolean isStop;
+	private ConnHolder connHolder;
+
+	private final long taskId;
+	private ServerConnectInfo connInfo;
+
+	// 失败重试次数
+	private int failTryCount = Config.DEFAULT_FAIL_TRY_COUNT;
+	private boolean printCost = Config.DEFAULT_PRINT_COST;
+	private long costBound = Config.DEFAULT_COST_BOUND;
+	private List<Pair<String, int[]>> deleteMeta;
+
+	public InsertTask(
+			final long taskId,
+			Queue<List<Record>> recordsQueue,
+			Configuration config,
+			ServerConnectInfo connectInfo,
+			String writeRecordSql,
+			List<Pair<String, int[]>> deleteMeta) {
+		this.taskId = taskId;
+		this.queue = recordsQueue;
+		this.connInfo = connectInfo;
+		failTryCount = config.getInt(Config.FAIL_TRY_COUNT, Config.DEFAULT_FAIL_TRY_COUNT);
+		printCost = config.getBool(Config.PRINT_COST, Config.DEFAULT_PRINT_COST);
+		costBound = config.getLong(Config.COST_BOUND, Config.DEFAULT_COST_BOUND);
+		this.connHolder = new ObClientConnHolder(config, connInfo.jdbcUrl,
+                                         connInfo.getFullUserName(), connInfo.password);
+		this.writeRecordSql = writeRecordSql;
+		this.isStop = false;
+		this.deleteMeta = deleteMeta;
+		connHolder.initConnection();
+	}
+	
+	void setWriterTask(ConcurrentTableWriterTask writerTask) {
+		this.writerTask = writerTask;
+	}
+	
+	void setWriter(ConcurrentTableWriter writer) {
+		this.writer = writer;
+	}
+
+	private boolean isStop() { return isStop; }
+	public void setStop() { isStop = true; }
+	public long getTotalCost() { return totalCost; }
+	public long getInsertCount() { return insertCount; }
+	
+	@Override
+	public void run() {
+		Thread.currentThread().setName(String.format("%d-insertTask-%d", taskId, Thread.currentThread().getId()));
+		LOG.debug("Task {} start to execute...", taskId);
+		while (!isStop()) {
+			try {
+				List<Record> records = queue.poll();
+				if (null != records) {
+					doMultiInsert(records, this.printCost, this.costBound);
+
+				} else if (writerTask.isFinished()) {
+					writerTask.singalTaskFinish();
+					LOG.debug("not more task, thread exist ...");
+					break;
+				} else {
+					TimeUnit.MILLISECONDS.sleep(5);
+				}
+			} catch (InterruptedException e) {
+				LOG.debug("TableWriter is interrupt");
+			} catch (Exception e) {
+				LOG.warn("ERROR UNEXPECTED {}", e);
+			}
+		}
+		LOG.debug("Thread exist...");
+	}
+	
+	public void destroy() {
+	    connHolder.destroy();
+	};
+	
+	public void calStatistic(final long cost) {
+		writer.increFinishCount();
+		++insertCount;
+		totalCost += cost;
+		if (this.printCost && cost > this.costBound) {
+			LOG.info("slow multi insert cost {}ms", cost);
+		}
+	}
+
+	private void doDelete(Connection conn, final List<Record> buffer) throws SQLException {
+		if(deleteMeta == null || deleteMeta.size() == 0) {
+			return;
+		}
+		for (int i = 0; i < deleteMeta.size(); i++) {
+			String deleteSql = deleteMeta.get(i).getKey();
+			int[] valueIdx = deleteMeta.get(i).getValue();
+			PreparedStatement ps = null;
+			try {
+				ps = conn.prepareStatement(deleteSql);
+				StringBuilder builder = new StringBuilder();
+				for (Record record : buffer) {
+					int bindIndex = 0;
+					for (int idx : valueIdx) {
+						writerTask.fillStatementIndex(ps, bindIndex++, idx, record.getColumn(idx));
+						builder.append(record.getColumn(idx).asString()).append(",");
+					}
+					ps.addBatch();
+				}
+				LOG.debug("delete values: " + builder.toString());
+				ps.executeBatch();
+			} catch (SQLException ex) {
+				LOG.error("SQL Exception when delete records with {}", deleteSql, ex);
+				throw ex;
+			} finally {
+				DBUtil.closeDBResources(ps, null);
+			}
+		}
+	}
+
+	public void doMultiInsert(final List<Record> buffer, final boolean printCost, final long restrict) {
+		checkMemstore();
+		Connection conn = connHolder.getConn();
+		boolean success = false;
+		long cost = 0;
+		long startTime = 0;
+		try {
+			for (int i = 0; i < failTryCount; ++i) {
+				if (i > 0) {
+					try {
+						int sleep = i >= 9 ? 500 : 1 << i;//不明白为什么要sleep 500s
+						TimeUnit.SECONDS.sleep(sleep);
+					} catch (InterruptedException e) {
+						LOG.info("thread interrupted ..., ignore");
+					}
+					conn = connHolder.getConn();
+					LOG.info("retry {}, start do batch insert, size={}", i, buffer.size());
+					checkMemstore();
+				}
+				startTime = System.currentTimeMillis();
+				PreparedStatement ps = null;
+				try {
+					conn.setAutoCommit(false);
+
+					// do delete if necessary
+					doDelete(conn, buffer);
+
+					ps = conn.prepareStatement(writeRecordSql);
+					for (Record record : buffer) {
+						ps = writerTask.fillStatement(ps, record);
+						ps.addBatch();
+					}
+					ps.executeBatch();
+					conn.commit();
+					success = true;
+					cost = System.currentTimeMillis() - startTime;
+					calStatistic(cost);
+					break;
+				} catch (SQLException e) {
+					LOG.warn("Insert fatal error SqlState ={}, errorCode = {}, {}", e.getSQLState(), e.getErrorCode(), e);
+					if (i == 0 || i > 10 ) {
+						for (Record record : buffer) {
+							LOG.warn("ERROR : record {}", record);
+						}
+					}
+					// 按照错误码分类，分情况处理
+					// 如果是OB系统级异常,则需要重建连接
+					boolean fatalFail = ObWriterUtils.isFatalError(e);
+					if (fatalFail) {
+						ObWriterUtils.sleep(300000);
+						connHolder.reconnect();
+						// 如果是可恢复的异常,则重试
+					} else if (ObWriterUtils.isRecoverableError(e)) {
+						conn.rollback();
+						ObWriterUtils.sleep(60000);
+					} else {// 其它异常直接退出,采用逐条写入方式
+						conn.rollback();
+						ObWriterUtils.sleep(1000);
+						break;
+					}
+				} catch (Exception e) {
+					e.printStackTrace();
+					LOG.warn("Insert error unexpected {}", e);
+				} finally {
+					DBUtil.closeDBResources(ps, null);
+				}
+			}
+		} catch (SQLException e) {
+			LOG.warn("ERROR:retry failSql State ={}, errorCode = {}, {}", e.getSQLState(), e.getErrorCode(), e);
+		}
+
+		if (!success) {
+			try {
+				LOG.info("do one insert");
+				conn = connHolder.reconnect();
+				doOneInsert(conn, buffer);
+				cost = System.currentTimeMillis() - startTime;
+				calStatistic(cost);
+			} finally {
+			}
+		}
+	}
+
+	// process one row, delete before insert
+	private void doOneInsert(Connection connection, List<Record> buffer) {
+		List<PreparedStatement> deletePstmtList = new ArrayList();
+		PreparedStatement preparedStatement = null;
+		try {
+			connection.setAutoCommit(false);
+			if (deleteMeta != null && deleteMeta.size() > 0) {
+				for (int i = 0; i < deleteMeta.size(); i++) {
+					String deleteSql = deleteMeta.get(i).getKey();
+					deletePstmtList.add(connection.prepareStatement(deleteSql));
+				}
+			}
+
+			preparedStatement = connection.prepareStatement(this.writeRecordSql);
+			for (Record record : buffer) {
+				try {
+					for (int i = 0; i < deletePstmtList.size(); i++) {
+						PreparedStatement deleteStmt = deletePstmtList.get(i);
+						int[] valueIdx = deleteMeta.get(i).getValue();
+						int bindIndex = 0;
+						for (int idx : valueIdx) {
+							writerTask.fillStatementIndex(deleteStmt, bindIndex++, idx, record.getColumn(idx));
+						}
+						deleteStmt.execute();
+					}
+					preparedStatement = writerTask.fillStatement(preparedStatement, record);
+					preparedStatement.execute();
+					connection.commit();
+				} catch (SQLException e) {
+					writerTask.collectDirtyRecord(record, e);
+				} finally {
+					// 此处不应该关闭statement，后续的数据还需要用到
+				}
+			}
+		} catch (Exception e) {
+			throw DataXException.asDataXException(
+					DBUtilErrorCode.WRITE_DATA_ERROR, e);
+		} finally {
+			DBUtil.closeDBResources(preparedStatement, null);
+			for (PreparedStatement pstmt : deletePstmtList) {
+				DBUtil.closeDBResources(pstmt, null);
+			}
+		}
+	}
+
+	private void checkMemstore() {
+		while (writerTask.isMemStoreFull()) {
+			ObWriterUtils.sleep(30000);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/SingleTableWriterTask.java

@@ -0,0 +1,152 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+public class SingleTableWriterTask extends CommonRdbmsWriter.Task {
+
+	// memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入
+	private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD;
+
+	// memstore检查的间隔
+	private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND;
+
+	// 最后一次检查
+	private long lastCheckMemstoreTime;
+
+	// 失败重试次数
+	private int failTryCount = Config.DEFAULT_FAIL_TRY_COUNT;
+
+	private ConnHolder connHolder;
+	private String obWriteMode = "update";
+	private boolean isOracleCompatibleMode = false;
+	private String obUpdateColumns = null;
+
+	public SingleTableWriterTask(DataBaseType dataBaseType) {
+		super(dataBaseType);
+	}
+
+	@Override
+	public void init(Configuration config) {
+		super.init(config);
+		this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD);
+		this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND,
+				Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND);
+		failTryCount = config.getInt(Config.FAIL_TRY_COUNT, Config.DEFAULT_FAIL_TRY_COUNT);
+		// OceanBase 所有操作都是 insert into on duplicate key update 模式
+		// writeMode应该使用enum来定义
+		this.writeMode = "update";
+		this.connHolder = new ObClientConnHolder(config, jdbcUrl, username, password);
+		//ob1.0里面，
+		this.batchSize = Math.min(128, config.getInt(Key.BATCH_SIZE, 128));
+		LOG.info("In Write OceanBase 1.0, Real Batch Size : " + this.batchSize);
+
+		isOracleCompatibleMode = ObWriterUtils.isOracleMode();
+		LOG.info("isOracleCompatibleMode=" + isOracleCompatibleMode);
+
+		obUpdateColumns = config.getString(Config.OB_UPDATE_COLUMNS, null);
+
+        obWriteMode = config.getString(Config.OB_WRITE_MODE, "update");
+        if (isOracleCompatibleMode) {
+            obWriteMode = "insert";
+        }
+		rewriteSql();
+	}
+
+	private void rewriteSql() {
+		Connection conn = connHolder.initConnection();
+		this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns);
+	}
+
+	protected void doBatchInsert(Connection conn, List<Record> buffer) throws SQLException {
+		doBatchInsert(buffer);
+	}
+
+	private void doBatchInsert(List<Record> buffer) {
+		Connection conn = connHolder.getConn();
+		// 检查内存
+		checkMemstore(conn);
+		boolean success = false;
+		try {
+			for (int i = 0; i < failTryCount; i++) {
+				PreparedStatement ps = null;
+				try {
+					conn.setAutoCommit(false);
+					ps = conn.prepareStatement(this.writeRecordSql);
+					for (Record record : buffer) {
+						ps = fillPreparedStatement(ps, record);
+						ps.addBatch();
+					}
+					ps.executeBatch();
+					conn.commit();
+					// 标记执行正常,且退出for循环
+					success = true;
+					break;
+				} catch (SQLException e) {
+					// 如果是OB系统级异常,则需要重建连接
+					boolean fatalFail = ObWriterUtils.isFatalError(e);
+					if (fatalFail) {
+						LOG.warn("Fatal exception in OB. Roll back this write and hibernate for five minutes. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						ObWriterUtils.sleep(300000);
+						DBUtil.closeDBResources(null, conn);
+						conn = connHolder.reconnect();
+						// 如果是可恢复的异常,则重试
+					} else if (ObWriterUtils.isRecoverableError(e)) {
+						LOG.warn("Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						conn.rollback();
+						ObWriterUtils.sleep(60000);
+						// 其它异常直接退出,采用逐条写入方式
+					} else {
+						LOG.warn("Exception in OB. Roll back this write and hibernate for one second. Write and submit the records one by one. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						conn.rollback();
+						ObWriterUtils.sleep(1000);
+						break;
+					}
+				} finally {
+					DBUtil.closeDBResources(ps, null);
+				}
+			}
+		} catch (SQLException e) {
+			LOG.warn("Exception in OB. Roll back this write. Write and submit the records one by one. SQLState: {}. ErrorCode: {}",
+					e.getSQLState(), e.getErrorCode(), e);
+		}
+		if (!success) {
+			doOneInsert(conn, buffer);
+		}
+	}
+
+	private void checkMemstore(Connection conn) {
+		long now = System.currentTimeMillis();
+		if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+		while (ObWriterUtils.isMemstoreFull(conn, memstoreThreshold)) {
+			LOG.warn("OB memstore is full,sleep 60 seconds, threshold=" + memstoreThreshold);
+			ObWriterUtils.sleep(60000);
+		}
+		lastCheckMemstoreTime = now;
+	}
+
+	@Override
+	public void destroy(Configuration writerSliceConfig) {
+		// 把本级持有的conn关闭掉
+		DBUtil.closeDBResources(null, connHolder.getConn());
+		super.destroy(writerSliceConfig);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java

@@ -0,0 +1,37 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WriterThreadPool {
+    private static final Logger LOG = LoggerFactory.getLogger(InsertTask.class);
+	
+	private static ExecutorService executorService = Executors.newCachedThreadPool();
+	
+	public WriterThreadPool() {
+	}
+	
+	public static ExecutorService getInstance() {
+		return executorService;
+	}
+	
+	public static synchronized void shutdown() {
+		LOG.info("start shutdown executor service...");
+		executorService.shutdown();
+		LOG.info("shutdown executor service success...");
+	}
+	
+	public static synchronized void execute(InsertTask task) {
+		executorService.execute(task);
+	}
+	
+	public static synchronized void executeBatch(List<InsertTask> tasks) {
+		for (InsertTask task : tasks) {
+			executorService.execute(task);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/util/DbUtils.java

@@ -0,0 +1,71 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.util;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class DbUtils {
+
+    protected static final Logger LOG = LoggerFactory.getLogger(DbUtils.class);
+
+    public static String fetchSingleValueWithRetry(Configuration config, String query) {
+        final String username = config.getString(Key.USERNAME);
+        final String password = config.getString(Key.PASSWORD);
+        String jdbcUrl = config.getString(Key.JDBC_URL);
+
+        if(jdbcUrl == null) {
+            List<Object> conns = config.getList(Constant.CONN_MARK, Object.class);
+            Configuration connConf = Configuration.from(conns.get(0).toString());
+            jdbcUrl = connConf.getString(Key.JDBC_URL);
+        }
+
+        Connection conn = null;
+        PreparedStatement stmt = null;
+        ResultSet result = null;
+        boolean need_retry = false;
+        String value = null;
+        int retry = 0;
+        do {
+            try {
+                if (retry > 0) {
+                    int sleep = retry > 9 ? 500 : 1 << retry;
+                    try {
+                        TimeUnit.SECONDS.sleep(sleep);
+                    } catch (InterruptedException e) {
+                    }
+                    LOG.warn("retry fetch value for {} the {} times", query, retry);
+                }
+                conn = DBUtil.getConnection(DataBaseType.OceanBase, jdbcUrl, username, password);
+                stmt = conn.prepareStatement(query);
+                result = stmt.executeQuery();
+                if (result.next()) {
+                    value = result.getString("Value");
+                } else {
+                    throw new RuntimeException("no values returned for " + query);
+                }
+                LOG.info("value for query [{}] is [{}]", query, value);
+                break;
+            } catch (SQLException e) {
+                need_retry = true;
+                ++retry;
+                LOG.warn("fetch value with {} error {}", query, e);
+            } finally {
+                DBUtil.closeDBResources(result, stmt, null);
+            }
+        } while (need_retry);
+
+        return value;
+    }
+}

oceanbasev10writer/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "oceanbasev10writer",
+    "class": "com.alibaba.datax.plugin.writer.oceanbasev10writer.OceanBaseV10Writer",
+    "description": "write data into oceanbase with sql interface",
+    "developer": "oceanbase"
+}

odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/DESCipher.java

@@ -40,7 +40,7 @@ public class DESCipher {
 	 * 　　
 	 */
 
-	public static final String KEY = "u4Gqu4Z8";
+	public static final String KEY = "DESDES";
 
 	private final static String DES = "DES";
 

odpswriter/src/main/java/com/alibaba/datax/plugin/writer/odpswriter/util/DESCipher.java

@@ -40,7 +40,7 @@ public class DESCipher {
 	 * 　　
 	 */
 
-	public static final String KEY = "u4Gqu4Z8";
+	public static final String KEY = "DESDES";
 
 	private final static String DES = "DES";
 

oscarwriter/pom.xml

@@ -39,12 +39,18 @@
             <artifactId>plugin-rdbms-util</artifactId>
             <version>${datax-project-version}</version>
         </dependency>
-        <dependency>
+    <!--    <dependency>
             <groupId>com.oscar</groupId>
             <artifactId>oscar</artifactId>
             <version>7.0.8</version>
             <scope>system</scope>
             <systemPath>${basedir}/src/main/lib/oscarJDBC.jar</systemPath>
+        </dependency>-->
+        <!--驱动文件找不到了，暂时替换成下面这个-->
+        <dependency>
+            <groupId>com.csicit.thirdparty</groupId>
+            <artifactId>oscar</artifactId>
+            <version>1.0.1</version>
         </dependency>
 
     </dependencies>

otsreader/pom.xml

@@ -34,6 +34,16 @@
             <groupId>com.aliyun.openservices</groupId>
             <artifactId>ots-public</artifactId>
             <version>2.2.4</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>log4j-api</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>log4j-core</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+            </exclusions>
         </dependency> 
         <dependency>
             <groupId>com.google.code.gson</groupId>

otsstreamreader/pom.xml

@@ -33,6 +33,16 @@
             <groupId>com.aliyun.openservices</groupId>
             <artifactId>tablestore-streamclient</artifactId>
             <version>1.0.0</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>log4j-api</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>log4j-core</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>com.google.code.gson</groupId>

otswriter/pom.xml

@@ -34,6 +34,16 @@
             <groupId>com.aliyun.openservices</groupId>
             <artifactId>ots-public</artifactId>
             <version>2.2.4</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>log4j-api</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>log4j-core</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                </exclusion>
+            </exclusions>
         </dependency> 
         <dependency>
             <groupId>com.google.code.gson</groupId>

package.xml

@@ -33,7 +33,7 @@
             <outputDirectory>datax</outputDirectory>
         </fileSet>
         <fileSet>
-            <directory>oceanbasereader/target/datax/</directory>
+            <directory>oceanbasev10reader/target/datax/</directory>
             <includes>
                 <include>**/*.*</include>
             </includes>
@@ -189,6 +189,13 @@
             </includes>
             <outputDirectory>datax</outputDirectory>
         </fileSet>
+        <fileSet>
+            <directory>tdenginewriter/target/datax/</directory>
+            <includes>
+                <include>**/*.*</include>
+            </includes>
+            <outputDirectory>datax</outputDirectory>
+        </fileSet>
         <fileSet>
             <directory>drdswriter/target/datax/</directory>
             <includes>
@@ -357,6 +364,13 @@
             </includes>
             <outputDirectory>datax</outputDirectory>
         </fileSet>
+        <fileSet>
+            <directory>tsdbreader/target/datax/</directory>
+            <includes>
+                <include>**/*.*</include>
+            </includes>
+            <outputDirectory>datax</outputDirectory>
+        </fileSet>
         <fileSet>
             <directory>adbpgwriter/target/datax/</directory>
             <includes>
@@ -385,5 +399,12 @@
             </includes>
             <outputDirectory>datax</outputDirectory>
         </fileSet>
+        <fileSet>
+            <directory>oceanbasev10writer/target/datax/</directory>
+            <includes>
+                <include>**/*.*</include>
+            </includes>
+            <outputDirectory>datax</outputDirectory>
+        </fileSet>
     </fileSets>
 </assembly>

plugin-rdbms-util/pom.xml

@@ -63,5 +63,5 @@
             <artifactId>guava</artifactId>
             <version>r05</version>
         </dependency>
-    </dependencies>
+	</dependencies>
 </project>

plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/Key.java

@@ -48,4 +48,11 @@ public final class Key {
 
     public static String SPLIT_FACTOR = "splitFactor";
 
-}
+    public final static String WEAK_READ = "weakRead";
+
+    public final static String SAVE_POINT = "savePoint";
+
+    public final static String REUSE_CONN = "reuseConn";
+
+    public final static String PARTITION_NAME = "partitionName";
+}

plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DBUtil.java

@@ -358,7 +358,7 @@ public final class DBUtil {
                                                    String url, String user, String pass, String socketTimeout) {
 
         //ob10的处理
-        if (url.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING) && dataBaseType == DataBaseType.MySql) {
+        if (url.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
             String[] ss = url.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
             if (ss.length != 3) {
                 throw DataXException
@@ -367,7 +367,7 @@ public final class DBUtil {
             }
             LOG.info("this is ob1_0 jdbc url.");
             user = ss[1].trim() +":"+user;
-            url = ss[2];
+            url = ss[2].replace("jdbc:mysql:", "jdbc:oceanbase:");
             LOG.info("this is ob1_0 jdbc url. user="+user+" :url="+url);
         }
 

plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DataBaseType.java

@@ -21,7 +21,8 @@ public enum DataBaseType {
     ADS("ads","com.mysql.jdbc.Driver"),
     ClickHouse("clickhouse", "ru.yandex.clickhouse.ClickHouseDriver"),
     KingbaseES("kingbasees", "com.kingbase8.Driver"),
-    Oscar("oscar", "com.oscar.Driver");
+    Oscar("oscar", "com.oscar.Driver"),
+    OceanBase("oceanbase", "com.alipay.oceanbase.jdbc.Driver");
 
 
     private String typeName;
@@ -42,6 +43,7 @@ public enum DataBaseType {
         switch (this) {
             case MySql:
             case DRDS:
+            case OceanBase:
                 suffix = "yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true";
                 if (jdbc.contains("?")) {
                     result = jdbc + "&" + suffix;
@@ -108,6 +110,14 @@ public enum DataBaseType {
                 break;
             case Oscar:
                 break;
+            case OceanBase:
+                suffix = "yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true";
+                if (jdbc.contains("?")) {
+                    result = jdbc + "&" + suffix;
+                } else {
+                    result = jdbc + "?" + suffix;
+                }
+                break;
             default:
                 throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
         }

plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/writer/CommonRdbmsWriter.java

@@ -402,13 +402,20 @@ public class CommonRdbmsWriter {
                 throws SQLException {
             for (int i = 0; i < this.columnNumber; i++) {
                 int columnSqltype = this.resultSetMetaData.getMiddle().get(i);
-                preparedStatement = fillPreparedStatementColumnType(preparedStatement, i, columnSqltype, record.getColumn(i));
+                String typeName = this.resultSetMetaData.getRight().get(i);
+                preparedStatement = fillPreparedStatementColumnType(preparedStatement, i, columnSqltype, typeName, record.getColumn(i));
             }
 
             return preparedStatement;
         }
 
-        protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex, int columnSqltype, Column column) throws SQLException {
+        protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex,
+                                                                    int columnSqltype, Column column) throws SQLException {
+            return fillPreparedStatementColumnType(preparedStatement, columnIndex, columnSqltype, null, column);
+        }
+
+        protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex,
+                                                                    int columnSqltype, String typeName, Column column) throws SQLException {
             java.util.Date utilDate;
             switch (columnSqltype) {
                 case Types.CHAR:
@@ -451,8 +458,11 @@ public class CommonRdbmsWriter {
 
                 // for mysql bug, see http://bugs.mysql.com/bug.php?id=35115
                 case Types.DATE:
-                    if (this.resultSetMetaData.getRight().get(columnIndex)
-                            .equalsIgnoreCase("year")) {
+                    if (typeName == null) {
+                        typeName = this.resultSetMetaData.getRight().get(columnIndex);
+                    }
+
+                    if (typeName.equalsIgnoreCase("year")) {
                         if (column.asBigInteger() == null) {
                             preparedStatement.setString(columnIndex + 1, null);
                         } else {

pom.xml

@@ -69,9 +69,11 @@
         <module>opentsdbreader</module>
         <module>cassandrareader</module>
         <module>gdbreader</module>
+        <module>oceanbasev10reader</module>
 
         <!-- writer -->
         <module>mysqlwriter</module>
+        <module>tdenginewriter</module>
         <module>drdswriter</module>
         <module>odpswriter</module>
         <module>txtfilewriter</module>
@@ -99,6 +101,7 @@
         <module>cassandrawriter</module>
         <module>clickhousewriter</module>
         <module>oscarwriter</module>
+        <module>oceanbasev10writer</module>
         <!-- common support module -->
         <module>plugin-rdbms-util</module>
         <module>plugin-unstructured-storage-util</module>
@@ -180,6 +183,17 @@
                 <version>1.9.5</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-api</artifactId>
+                <version>2.17.1</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-core</artifactId>
+                <version>2.17.1</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

postgresqlreader/doc/postgresqlreader.md

@@ -48,7 +48,7 @@ PostgresqlReader插件实现了从PostgreSQL读取数据。在底层实现上，
                         // 数据库连接密码
                         "password": "xx",
                         "column": [
-                            "id"，"name"
+                            "id","name"
                         ],
                         //切分主键
                         "splitPk": "id",

postgresqlwriter/src/main/java/com/alibaba/datax/plugin/writer/postgresqlwriter/PostgresqlWriter.java

@@ -67,6 +67,8 @@ public class PostgresqlWriter extends Writer {
 				public String calcValueHolder(String columnType){
 					if("serial".equalsIgnoreCase(columnType)){
 						return "?::int";
+					}else if("bigserial".equalsIgnoreCase(columnType)){
+						return "?::int8";
 					}else if("bit".equalsIgnoreCase(columnType)){
 						return "?::bit varying";
 					}

rdbmsreader/doc/rdbmsreader.md

@@ -138,7 +138,7 @@ RDBMSReader插件实现了从RDBMS读取数据。在底层实现上，RDBMSReade
   
 	**rdbmswriter如何增加新的数据库支持:**  
 	
-	- 进入rdbmsreader对应目录，这里${DATAX_HOME}为DataX主目录，即: ${DATAX_HOME}/plugin/reader/rdbmswriter
+	- 进入rdbmsreader对应目录，这里${DATAX_HOME}为DataX主目录，即: ${DATAX_HOME}/plugin/reader/rdbmsreader
 	- 在rdbmsreader插件目录下有plugin.json配置文件，在此文件中注册您具体的数据库驱动，具体放在drivers数组中。rdbmsreader插件在任务执行时会动态选择合适的数据库驱动连接数据库。
 
 

tdenginewriter/doc/tdenginewriter-CN.md

@@ -0,0 +1,405 @@
+# DataX TDengineWriter
+
+简体中文| [English](./tdenginewriter.md)
+
+## 1 快速介绍
+
+TDengineWriter插件实现了写入数据到TDengine数据库功能。可用于离线同步其它数据库的数据到TDengine。
+
+## 2 实现原理
+
+TDengineWriter 通过 DataX 框架获取 Reader生成的协议数据，根据reader的类型解析数据。目前有两种写入方式：
+
+1. 对于OpenTSDBReader, TDengineWriter通过JNI方式调用TDengine客户端库文件（taos.lib或taos.dll）中的方法，使用[schemaless的方式](https://www.taosdata.com/cn/documentation/insert#schemaless)写入。
+
+2. 对于其它数据源,会根据配置生成SQL语句, 通过[taos-jdbcdriver](https://www.taosdata.com/cn/documentation/connector/java)批量写入。
+
+这样区分的原因是OpenTSDBReader将opentsdb的数据统一读取为json字符串，Writer端接收到的数据只有1列。而其它Reader插件一般会把数据放在不同列。
+
+## 3 功能说明
+### 3.1 从OpenTSDB到TDengine
+#### 3.1.1 配置样例
+
+```json
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "opentsdbreader",
+          "parameter": {
+            "endpoint": "http://192.168.1.180:4242",
+            "column": [
+              "weather_temperature"
+            ],
+            "beginDateTime": "2021-01-01 00:00:00",
+            "endDateTime": "2021-01-01 01:00:00"
+          }
+        },
+        "writer": {
+          "name": "tdenginewriter",
+          "parameter": {
+            "host": "192.168.1.180",
+            "port": 6030,
+            "dbName": "test",
+            "username": "root",
+            "password": "taosdata"
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}
+```
+
+#### 3.1.2 参数说明
+
+| 参数      | 描述                 | 是否必选 | 默认值   |
+| --------- | -------------------- | -------- | -------- |
+| host      | TDengine实例的host   | 是       | 无       |
+| port      | TDengine实例的port   | 是       | 无       |
+| username      | TDengine实例的用户名 | 否       | root     |
+| password  | TDengine实例的密码   | 否       | taosdata |
+| dbName    | 目的数据库的名称     | 是       | 无       |
+| batchSize | 每次批量插入多少记录 | 否       | 1        |
+
+
+#### 3.1.3 类型转换
+
+目前，由于OpenTSDBReader将opentsdb的数据统一读取为json字符串，TDengineWriter 在做Opentsdb到TDengine的迁移时，按照以下类型进行处理：
+
+| OpenTSDB数据类型 | DataX 内部类型 | TDengine 数据类型 |
+| ---------------- | -------------- | ----------------- |
+| timestamp        | Date           | timestamp         |
+| Integer（value） | Double         | double            |
+| Float（value）   | Double         | double            |
+| String（value）  | String         | binary            |
+| Integer（tag）   | String         | binary            |
+| Float（tag）     | String         | binary            |
+| String（tag）    | String         | binary            |
+
+### 3.2 从MongoDB到TDengine
+
+#### 3.2.1 配置样例
+```json
+{
+    "job": {
+        "setting": {
+            "speed": {
+                "channel": 2
+            }
+        },
+        "content": [
+            {
+                "reader": {
+                    "name": "mongodbreader",
+                    "parameter": {
+                        "address": [
+                            "127.0.0.1:27017"
+                        ],
+                        "userName": "user",
+                        "mechanism": "SCRAM-SHA-1",
+                        "userPassword": "password",
+                        "authDb": "admin",
+                        "dbName": "test",
+                        "collectionName": "stock",
+                        "column": [
+                            {
+                              "name": "stockID",
+                              "type": "string"  
+                            },
+                            {
+                                "name": "tradeTime",
+                                "type": "date"
+                            },
+                            {
+                                "name": "lastPrice",
+                                "type": "double"
+                            },
+                            {
+                              "name": "askPrice1",
+                              "type": "double"
+                            },
+                            {
+                              "name": "bidPrice1",
+                              "type": "double"
+                            },
+                            {
+                                "name": "volume",
+                                "type": "int"
+                            }
+                        ]
+                    }
+                },
+                "writer": {
+                    "name": "tdenginewriter",
+                    "parameter": {
+                        "host": "localhost",
+                        "port": 6030,
+                        "dbName": "test",
+                        "username": "root",
+                        "password": "taosdata",
+                        "stable": "stock",
+                        "tagColumn": {
+                          "industry": "energy",
+                          "stockID": 0
+                        },
+                        "fieldColumn": {
+                          "lastPrice": 2,
+                          "askPrice1": 3,
+                          "bidPrice1": 4,
+                          "volume": 5
+                        },
+                        "timestampColumn": {
+                          "tradeTime": 1
+                        }
+                    }
+                }
+            }
+        ]
+    }
+}
+```
+
+**注：本配置的writer部分同样适用于关系型数据库**
+
+
+#### 3.2.2 参数说明
+| 参数            | 描述                 | 是否必选         | 默认值   | 备注               |
+| --------------- | -------------------- | ---------------- | -------- | ------------------ |
+| host            | TDengine实例的host   | 是               | 无       |
+| port            | TDengine实例的port   | 是               | 无       |
+| username            | TDengine实例的用户名 | 否               | root     |
+| password        | TDengine实例的密码   | 否               | taosdata |
+| dbName          | 目的数据库的名称     | 是               | 无       |
+| batchSize       | 每次批量插入多少记录 | 否               | 1000     |
+| stable          | 目标超级表的名称     | 是(OpenTSDB除外) | 无       |
+| tagColumn       | 格式:{tagName1: tagInd1, tagName2: tagInd2}, 标签列在写插件收到的Record中的位置和列名   | 否               | 无       | 位置索引均从0开始, tagInd如果为字符串, 表示固定标签值，不需要从源数据中获取  |
+| fieldColumn     | 格式:{fdName1: fdInd1, fdName2: fdInd2}, 字段列在写插件收到的Record中的位置和列名   | 否               | 无       |                    |
+| timestampColumn | 格式:{tsColName: tsColIndex}, 时间戳列在写插件收到的Record中的位置和列名 | 否               | 无       | 时间戳列只能有一个 |
+
+示例配置中tagColumn有一个industry，它的值是一个固定的字符串“energy”, 作用是给导入的所有数据加一个值为"energy"的固定标签industry。这个应用场景可以是：在源库中，有多个设备采集的数据分表存储，设备名就是表名，可以用这个机制把设备名称转化为标签。
+
+#### 3.2.3 自动建表规则
+##### 3.2.3.1 超级表创建规则
+
+如果配置了tagColumn、 fieldColumn和timestampColumn将会在插入第一条数据前，自动创建超级表。<br>
+数据列的类型从第1条记录自动推断, 标签列默认类型为`NCHAR(64)`, 比如示例配置，可能生成以下建表语句：
+
+```sql
+CREATE STABLE IF NOT EXISTS market_snapshot (
+  tadetime TIMESTAMP,
+  lastprice DOUBLE,
+  askprice1 DOUBLE,
+  bidprice1 DOUBLE,
+  volume INT
+)
+TAGS(
+  industry NCHAR(64),
+  stockID NCHAR(64)
+);
+```
+
+##### 3.2.3.2 子表创建规则
+
+子表结构与超级表相同，子表表名生成规则：
+1. 将标签的value 组合成为如下的字符串: `tag_value1!tag_value2!tag_value3`。
+2. 计算该字符串的 MD5 散列值 "md5_val"。
+3. "t_md5val"作为子表名。其中的 "t" 是固定的前缀。
+
+#### 3.2.4 用户提前建表
+
+如果你已经创建好目标超级表，那么tagColumn、 fieldColumn和timestampColumn三个字段均可省略, 插件将通过执行通过`describe stableName`获取表结构的信息。
+此时要求接收到的Record中Column的顺序和执行`describe stableName`返回的列顺序相同， 比如通过`describe stableName`返回以下内容：
+```
+             Field              |         Type         |   Length    |   Note   |
+=================================================================================
+ ts                             | TIMESTAMP            |           8 |          |
+ current                        | DOUBLE               |           8 |          |
+ location                      | BINARY                |           10 | TAG      |
+```
+那么插件收到的数据第1列必须代表时间戳，第2列必须代表电流，第3列必须代表位置。
+
+#### 3.2.5 注意事项
+
+1. tagColumn、 fieldColumn和timestampColumn三个字段用于描述目标表的结构信息，这三个配置字段必须同时存在或同时省略。
+2. 如果存在以上三个配置，且目标表也已经存在，则两者必须一致。**一致性**由用户自己保证，插件不做检查。不一致可能会导致插入失败或插入数据错乱。
+
+#### 3.2.6 类型转换
+
+| DataX 内部类型 | TDengine 数据类型 |
+|-------------- | ----------------- |
+|Long           | BIGINT            |
+|Double         | DOUBLE            |
+|String         | NCHAR(64)         |
+|Date           | TIMESTAMP         |
+|Boolean        | BOOL              |
+|Bytes          | BINARY(64)        |
+
+### 3.3 从关系型数据库到TDengine
+writer部分的配置规则和上述MongoDB的示例是一样的，这里给出一个MySQL的示例。
+
+#### 3.3.1 MySQL中表结构
+```sql
+CREATE TABLE IF NOT EXISTS weather(
+    station varchar(100),
+    latitude DOUBLE,
+    longtitude DOUBLE,
+    `date` DATE,
+    TMAX int,
+    TMIN int
+)
+```
+
+#### 3.3.2 配置文件示例
+
+```json 
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "mysqlreader",
+          "parameter": {
+            "username": "root",
+            "password": "passw0rd",
+            "column": [
+              "*"
+            ],
+            "splitPk": "station",
+            "connection": [
+              {
+                "table": [
+                  "weather"
+                ],
+                "jdbcUrl": [
+                  "jdbc:mysql://127.0.0.1:3306/test?useSSL=false&useUnicode=true&characterEncoding=utf8"
+                ]
+              }
+            ]
+          }
+        },
+        "writer": {
+          "name": "tdenginewriter",
+          "parameter": {
+            "host": "127.0.0.1",
+            "port": 6030,
+            "dbName": "test",
+            "username": "root",
+            "password": "taosdata",
+            "batchSize": 1000,
+            "stable": "weather",
+            "tagColumn": {
+              "station": 0
+            },
+            "fieldColumn": {
+              "latitude": 1,
+              "longtitude": 2,
+              "tmax": 4,
+              "tmin": 5
+            },
+            "timestampColumn":{
+              "date": 3
+            }
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}
+```
+
+
+## 4 性能报告
+
+### 4.1 环境准备
+
+#### 4.1.1 数据特征
+
+建表语句：
+
+单行记录类似于：
+
+#### 4.1.2 机器参数
+
+* 执行DataX的机器参数为:
+    1. cpu:
+    2. mem:
+    3. net: 千兆双网卡
+    4. disc: DataX 数据不落磁盘，不统计此项
+
+* TDengine数据库机器参数为:
+    1. cpu:
+    2. mem:
+    3. net: 千兆双网卡
+    4. disc:
+
+#### 4.1.3 DataX jvm 参数
+
+	-Xms1024m -Xmx1024m -XX:+HeapDumpOnOutOfMemoryError
+
+### 4.2 测试报告
+
+#### 4.2.1 单表测试报告
+
+| 通道数 | DataX速度(Rec/s) | DataX流量(MB/s) | DataX机器网卡流出流量(MB/s) | DataX机器运行负载 | DB网卡进入流量(MB/s) | DB运行负载 | DB TPS |
+| ------ | ---------------- | --------------- | --------------------------- | ----------------- | -------------------- | ---------- | ------ |
+| 1      |                  |                 |                             |                   |                      |            |        |
+| 4      |                  |                 |                             |                   |                      |            |        |
+| 8      |                  |                 |                             |                   |                      |            |        |
+| 16     |                  |                 |                             |                   |                      |            |        |
+| 32     |                  |                 |                             |                   |                      |            |        |
+
+说明：
+
+1. 这里的单表，主键类型为 bigint(20),自增。
+2. batchSize 和 通道个数，对性能影响较大。
+3. 16通道，4096批量提交时，出现 full gc 2次。
+
+#### 4.2.4 性能测试小结
+
+
+## 5 约束限制
+
+1. 本插件自动创建超级表时NCHAR类型的长度固定为64，对于包含长度大于64的字符串的数据源，将不支持。
+2. 标签列不能包含null值，如果包含会被过滤掉。
+
+## FAQ
+
+### 如何选取要同步的数据的范围？
+
+数据范围的选取在Reader插件端配置，对于不同的Reader插件配置方法往往不同。比如对于mysqlreader， 可以用sql语句指定数据范围。对于opentsdbreader, 用beginDateTime和endDateTime两个配置项指定数据范围。
+
+### 如何一次导入多张源表？
+
+如果Reader插件支持一次读多张表，Writer插件就能一次导入多张表。如果Reader不支持多多张表，可以建多个job，分别导入。Writer插件只负责写数据。
+
+### 一张源表导入之后对应TDengine中多少张表？
+
+这是由tagColumn决定的，如果所有tag列的值都相同，那么目标表只有一个。源表有多少不同的tag组合，目标超级表就有多少子表。
+
+### 源表和目标表的字段顺序一致吗？
+
+TDengine要求每个表第一列是时间戳列，后边是普通字段，最后是标签列。如果源表不是这个顺序，插件在自动建表时会自动调整。
+
+### 插件如何确定各列的数据类型？
+
+根据收到的第一批数据自动推断各列的类型。
+
+### 为什么插入10年前的数据会抛异常`TDengine ERROR (2350): failed to execute batch bind` ?
+
+因为创建数据库的时候，默认保留10年的数据。可以手动指定要保留多长时间的数据，比如:`CREATE DATABASE power KEEP 36500;`。
+
+### 如果编译的时候某些插件的依赖找不到怎么办？
+
+如果这个插件不是必须的，可以注释掉根目录下的pom.xml中的对应插件。

tdenginewriter/doc/tdenginewriter.md

@@ -0,0 +1,356 @@
+# DataX TDengineWriter
+
+[简体中文](./tdenginewriter-CN.md) | English
+
+## 1 Quick Introduction
+
+TDengineWriter Plugin writes data to [TDengine](https://www.taosdata.com/en/). It can be used to offline synchronize data from other databases to TDengine.
+
+## 2 Implementation
+
+TDengineWriter get records from DataX Framework that are generated from reader side. It has two whiting strategies:
+
+1. For data from OpenTSDBReader which is in json format, to leverage the new feature of TDengine Server that support writing json data directly called [schemaless writing](https://www.taosdata.com/cn/documentation/insert#schemaless), we use JNI to call functions in `taos.lib` or `taos.dll`.(Since the feature was not included in taos-jdbcdrive until version 2.0.36).
+2. For other data sources, we use [taos-jdbcdriver](https://www.taosdata.com/cn/documentation/connector/java) to write data. If the target table is not exists beforehand, then it will be created automatically according to your configuration.
+
+## 3 Features Introduction
+### 3.1 From OpenTSDB to TDengine
+#### 3.1.1 Sample Setting
+
+```json
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "opentsdbreader",
+          "parameter": {
+            "endpoint": "http://192.168.1.180:4242",
+            "column": [
+              "weather_temperature"
+            ],
+            "beginDateTime": "2021-01-01 00:00:00",
+            "endDateTime": "2021-01-01 01:00:00"
+          }
+        },
+        "writer": {
+          "name": "tdenginewriter",
+          "parameter": {
+            "host": "192.168.1.180",
+            "port": 6030,
+            "dbName": "test",
+            "username": "root",
+            "password": "taosdata"
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}
+```
+
+#### 3.1.2 Configuration
+
+| Parameter | Description                    | Required | Default  |
+| --------- | ------------------------------ | -------- | -------- |
+| host      | host of TDengine               | Yes      |          |
+| port      | port of TDengine               | Yes      |          |
+| username      | use name of TDengine           | No       | root     |
+| password  | password of TDengine           | No       | taosdata |
+| dbName    | name of target database        | No       |          |
+| batchSize | batch size of insert operation | No       | 1        |
+
+
+#### 3.1.3 Type Convert
+
+| OpenTSDB Type    | DataX Type | TDengine Type |
+| ---------------- | ---------- | ------------- |
+| timestamp        | Date       | timestamp     |
+| Integer（value） | Double     | double        |
+| Float（value）   | Double     | double        |
+| String（value）  | String     | binary        |
+| Integer（tag）   | String     | binary        |
+| Float（tag）     | String     | binary        |
+| String（tag）    | String     | binary        |
+
+### 3.2 From MongoDB to TDengine
+
+#### 3.2.1 Sample Setting
+```json
+{
+    "job": {
+        "setting": {
+            "speed": {
+                "channel": 2
+            }
+        },
+        "content": [
+            {
+                "reader": {
+                    "name": "mongodbreader",
+                    "parameter": {
+                        "address": [
+                            "127.0.0.1:27017"
+                        ],
+                        "userName": "user",
+                        "mechanism": "SCRAM-SHA-1",
+                        "userPassword": "password",
+                        "authDb": "admin",
+                        "dbName": "test",
+                        "collectionName": "stock",
+                        "column": [
+                            {
+                              "name": "stockID",
+                              "type": "string"  
+                            },
+                            {
+                                "name": "tradeTime",
+                                "type": "date"
+                            },
+                            {
+                                "name": "lastPrice",
+                                "type": "double"
+                            },
+                            {
+                              "name": "askPrice1",
+                              "type": "double"
+                            },
+                            {
+                              "name": "bidPrice1",
+                              "type": "double"
+                            },
+                            {
+                                "name": "volume",
+                                "type": "int"
+                            }
+                        ]
+                    }
+                },
+                "writer": {
+                    "name": "tdenginewriter",
+                    "parameter": {
+                        "host": "localhost",
+                        "port": 6030,
+                        "dbName": "test",
+                        "username": "root",
+                        "password": "taosdata",
+                        "stable": "stock",
+                        "tagColumn": {
+                          "industry": "energy",
+                          "stockID": 0
+                        },
+                        "fieldColumn": {
+                          "lastPrice": 2,
+                          "askPrice1": 3,
+                          "bidPrice1": 4,
+                          "volume": 5
+                        },
+                        "timestampColumn": {
+                          "tradeTime": 1
+                        }
+                    }
+                }
+            }
+        ]
+    }
+}
+```
+
+**Note：the writer part of this setting can also apply to other data source except for OpenTSDB **
+
+
+#### 3.2.2 Configuration
+
+| Parameter       | Description                                                     | Required                 | Default  | Remark              |
+| --------------- | --------------------------------------------------------------- | ------------------------ | -------- | ------------------- |
+| host            | host ofTDengine                                                 | Yes                      |          |
+| port            | port of TDengine                                                | Yes                      |          |
+| username            | username of TDengine                                           | No                       | root     |
+| password        | password of TDengine                                            | No                       | taosdata |
+| dbName          | name of target database                                         | Yes                      |          |
+| batchSize       | batch size of insert operation                                  | No                       | 1000     |
+| stable          | name of target super table                                      | Yes(except for OpenTSDB) |          |
+| tagColumn       | name and position of tag columns in the record from reader, format:{tagName1: tagInd1, tagName2: tagInd2}      | No                       |          | index starts with 0 |
+| fieldColumn     | name and position of data columns in the record from reader, format: {fdName1: fdInd1, fdName2: fdInd2}     | No                       |          |                     |
+| timestampColumn | name and position of timestamp column in the record from reader | No                       |          |                     |
+
+**Note**: You see that the value of tagColumn "industry" is a fixed string, this ia a good feature of this plugin. Think about this scenario: you have many tables with the structure and one table corresponds to one device. You want to use the device number as a tag in the target super table, then this feature is designed for you.
+
+#### 3.2.3 Auto table creating
+##### 3.2.3.1 Rules
+
+If all of `tagColumn`, `fieldColumn` and `timestampColumn` are offered in writer configuration, then target super table will be created automatically.
+The type of tag columns will always be `NCHAR(64)`. The sample setting above will produce following sql:
+
+```sql
+CREATE STABLE IF NOT EXISTS market_snapshot (
+  tadetime TIMESTAMP,
+  lastprice DOUBLE,
+  askprice1 DOUBLE,
+  bidprice1 DOUBLE,
+  volume INT
+)
+TAGS(
+  industry NCHAR(64),
+  stockID NCHAR(64)
+);
+```
+
+##### 3.2.3.2 Sub-table Creating Rules
+
+The structure of sub-tables are the same with structure of super table. The names of sub-tables are generated by rules below:
+1. combine value of tags like this:`tag_value1!tag_value2!tag_value3`.
+2. compute md5 hash hex of  above string, named `md5val`
+3. use "t_md5val" as sub-table name, in which "t" is fixed prefix.
+
+#### 3.2.4 Use Pre-created Table
+
+If you have created super table firstly, then all of tagColumn, fieldColumn and timestampColumn can be omitted. The writer plugin will get table schema by executing `describe stableName`.
+The order of columns of records received by this plugin must be the same as the order of columns returned by `describe stableName`. For example, if you have super table as below:
+```
+             Field              |         Type         |   Length    |   Note   |
+=================================================================================
+ ts                             | TIMESTAMP            |           8 |          |
+ current                        | DOUBLE               |           8 |          |
+ location                      | BINARY                |           10 | TAG      |
+```
+Then the first columns received by this writer plugin must represent timestamp, the second column must represent current with type double, the third column must represent location with internal type string.
+
+#### 3.2.5 Remarks
+
+1. Config keys --tagColumn, fieldColumn and timestampColumn, must be presented or omitted at the same time.  
+2. If above three config keys exist and the target table also exists, then the order of columns defined by the config file and the existed table must be the same.
+
+#### 3.2.6 Type Convert
+
+|DataX Type | TDengine Type |
+|-------------- | ----------------- |
+|Long           | BIGINT            |
+|Double         | DOUBLE            |
+|String         | NCHAR(64)         |
+|Date           | TIMESTAMP         |
+|Boolean        | BOOL              |
+|Bytes          | BINARY(64)        |
+
+### 3.3 From Relational Database to TDengine
+
+Take MySQl as example.
+
+#### 3.3.1 Table Structure in MySQL
+```sql
+CREATE TABLE IF NOT EXISTS weather(
+    station varchar(100),
+    latitude DOUBLE,
+    longtitude DOUBLE,
+    `date` DATE,
+    TMAX int,
+    TMIN int
+)
+```
+
+#### 3.3.2 Sample Setting
+
+```json 
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "mysqlreader",
+          "parameter": {
+            "username": "root",
+            "password": "passw0rd",
+            "column": [
+              "*"
+            ],
+            "splitPk": "station",
+            "connection": [
+              {
+                "table": [
+                  "weather"
+                ],
+                "jdbcUrl": [
+                  "jdbc:mysql://127.0.0.1:3306/test?useSSL=false&useUnicode=true&characterEncoding=utf8"
+                ]
+              }
+            ]
+          }
+        },
+        "writer": {
+          "name": "tdenginewriter",
+          "parameter": {
+            "host": "127.0.0.1",
+            "port": 6030,
+            "dbName": "test",
+            "username": "root",
+            "password": "taosdata",
+            "batchSize": 1000,
+            "stable": "weather",
+            "tagColumn": {
+              "station": 0
+            },
+            "fieldColumn": {
+              "latitude": 1,
+              "longtitude": 2,
+              "tmax": 4,
+              "tmin": 5
+            },
+            "timestampColumn":{
+              "date": 3
+            }
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}
+```
+
+
+## 4 Performance Test
+
+## 5 Restriction
+
+1. NCHAR type has fixed length 64 when auto creating stable.
+2. Rows have null tag values will be dropped.
+
+## FAQ
+
+### How to filter on source table？
+
+It depends on reader plugin. For different reader plugins, the way may be different.
+
+### How to import multiple source tables at once？
+
+It depends on reader plugin. If the reader plugin supports reading multiple tables at once, then there is no problem. 
+
+### How many sub-tables will be produced?
+
+The number of sub-tables is determined by tagColumns, equals to the number of different combinations of tag values. 
+
+### Do columns in source table and columns in target table must be in the same order?
+
+No. TDengine require the first column has timestamp type，which is followed by data columns, followed by tag columns. The writer plugin will create super table in this column order, regardless of origin column orders.
+
+### How dose the plugin infer the data type of incoming data?
+
+By the first batch of records it received.
+
+###  Why can't I insert data of 10 years ago? Do this will get error: `TDengine ERROR (2350): failed to execute batch bind`.
+
+Because the database you created only keep 10 years data by default, you can create table like this: `CREATE DATABASE power KEEP 36500;`, in order to enlarge the time period to 100 years.
+
+
+### What should I do if some dependencies of a plugin can't be found?
+
+I this plugin is not necessary for you, just remove it from pom.xml under project's root directory.

tdenginewriter/pom.xml

@@ -0,0 +1,107 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.alibaba.datax.tdenginewriter</groupId>
+    <artifactId>tdenginewriter</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.taosdata.jdbc</groupId>
+            <artifactId>taos-jdbcdriver</artifactId>
+            <version>2.0.34</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.taosdata.jdbc</groupId>
+            <artifactId>taos-jdbcdriver</artifactId>
+            <version>2.0.34</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit-version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons-lang3-version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.12.4</version>
+                <configuration>
+                    <!-- 包含哪些测试用例 -->
+                    <includes>
+                        <include>**/*Test.java</include>
+                    </includes>
+                    <!-- 不包含哪些测试用例 -->
+                    <excludes>
+                    </excludes>
+                    <testFailureIgnore>true</testFailureIgnore>
+                </configuration>
+            </plugin>
+
+        </plugins>
+    </build>
+
+</project>

tdenginewriter/src/main/assembly/package.xml

@@ -0,0 +1,34 @@
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+          xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+                <include>plugin_job_template.json</include>
+            </includes>
+            <outputDirectory>plugin/writer/tdenginewriter</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>tdenginewriter-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/tdenginewriter</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/writer/tdenginewriter/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/DataHandler.java

@@ -0,0 +1,12 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.plugin.RecordReceiver;
+
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+
+import java.util.Properties;
+
+public interface DataHandler {
+
+    long handle(RecordReceiver lineReceiver, Properties properties, TaskPluginCollector collector);
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/DataHandlerFactory.java

@@ -0,0 +1,10 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+public class DataHandlerFactory {
+
+    public static DataHandler build(String peerPluginName) {
+        if (peerPluginName.equals("opentsdbreader"))
+            return new OpentsdbDataHandler();
+        return new DefaultDataHandler();
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/DefaultDataHandler.java

@@ -0,0 +1,108 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.taosdata.jdbc.TSDBDriver;
+import com.taosdata.jdbc.TSDBPreparedStatement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Properties;
+
+/**
+ * 默认DataHandler
+ */
+public class DefaultDataHandler implements DataHandler {
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultDataHandler.class);
+
+    static {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+        } catch (ClassNotFoundException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public long handle(RecordReceiver lineReceiver, Properties properties, TaskPluginCollector collector) {
+        SchemaManager schemaManager = new SchemaManager(properties);
+        if (!schemaManager.configValid()) {
+            return 0;
+        }
+
+        try {
+            Connection conn = getTaosConnection(properties);
+            if (conn == null) {
+                return 0;
+            }
+            if (schemaManager.shouldGuessSchema()) {
+                // 无法从配置文件获取表结构信息，尝试从数据库获取
+                LOG.info(Msg.get("try_get_schema_from_db"));
+                boolean success = schemaManager.getFromDB(conn);
+                if (!success) {
+                    return 0;
+                }
+            } else {
+
+            }
+            int batchSize = Integer.parseInt(properties.getProperty(Key.BATCH_SIZE, "1000"));
+            if (batchSize < 5) {
+                // batchSize太小，会增加自动类型推断错误的概率，建议改大后重试
+                LOG.error(Msg.get("batch_size_too_small"));
+                return 0;
+            }
+            return write(lineReceiver, conn, batchSize, schemaManager, collector);
+        } catch (Exception e) {
+            LOG.error("write failed " + e.getMessage());
+            e.printStackTrace();
+        }
+        return 0;
+    }
+
+
+    private Connection getTaosConnection(Properties properties) throws SQLException {
+        // 检查必要参数
+        String host = properties.getProperty(Key.HOST);
+        String port = properties.getProperty(Key.PORT);
+        String dbname = properties.getProperty(Key.DBNAME);
+        String user = properties.getProperty(Key.USER);
+        String password = properties.getProperty(Key.PASSWORD);
+        if (host == null || port == null || dbname == null || user == null || password == null) {
+            String keys = String.join(" ", Key.HOST, Key.PORT, Key.DBNAME, Key.USER, Key.PASSWORD);
+            LOG.error("Required options missing, please check: " + keys);
+            return null;
+        }
+        String jdbcUrl = String.format("jdbc:TAOS://%s:%s/%s?user=%s&password=%s", host, port, dbname, user, password);
+        Properties connProps = new Properties();
+        connProps.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
+        LOG.info("TDengine connection established, host:{} port:{} dbname:{} user:{}", host, port, dbname, user);
+        return DriverManager.getConnection(jdbcUrl, connProps);
+    }
+
+    /**
+     * 使用SQL批量写入<br/>
+     *
+     * @return 成功写入记录数
+     * @throws SQLException
+     */
+    private long write(RecordReceiver lineReceiver, Connection conn, int batchSize, SchemaManager scm, TaskPluginCollector collector) throws SQLException {
+        Record record = lineReceiver.getFromReader();
+        if (record == null) {
+            return 0;
+        }
+        String pq = String.format("INSERT INTO ? USING %s TAGS(%s) (%s) values (%s)", scm.getStable(), scm.getTagValuesPlaceHolder(), scm.getJoinedFieldNames(), scm.getFieldValuesPlaceHolder());
+        LOG.info("Prepared SQL: {}", pq);
+        try (TSDBPreparedStatement stmt = (TSDBPreparedStatement) conn.prepareStatement(pq)) {
+            JDBCBatchWriter batchWriter = new JDBCBatchWriter(conn, stmt, scm, batchSize, collector);
+            do {
+                batchWriter.append(record);
+            } while ((record = lineReceiver.getFromReader()) != null);
+            batchWriter.flush();
+            return batchWriter.getCount();
+        }
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/JDBCBatchWriter.java

@@ -0,0 +1,244 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.taosdata.jdbc.TSDBPreparedStatement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * 使用JDBC原生写入接口批量写入。<br/>
+ * 有两个限制条件导致批量写入的代码逻辑过于复杂，以至于需要开发新的类来封装。<br/>
+ * 1. 用户必须提前把需要批量写入的数据搜集到ArrayList中
+ * 2. 每批写入的表名必须相同。
+ * 这个类的实现逻辑是：
+ * 1. 先把属于同一子表的Record缓存起来
+ * 2. 缓存的数量达到batchSize阈值，自动执行一次批量写入
+ * 3. 最后一批数据需要用户手动flush才能写入
+ */
+public class JDBCBatchWriter {
+    public static final Logger LOG = LoggerFactory.getLogger(JDBCBatchWriter.class);
+    private TSDBPreparedStatement stmt;
+    private SchemaManager scm;
+    private Connection conn;
+    private int batchSize;
+    private TaskPluginCollector collector;
+
+    // 缓存Record， key为tableName
+    Map<String, List<Record>> buf = new HashMap<>();
+    // 缓存表的标签值， key为tableName
+    Map<String, String[]> tableTagValues = new HashMap<>();
+    private long sucCount = 0;
+    private final int tsColIndex;
+    private List<String> fieldList;
+    // 每个record至少应该包含的列数，用于校验数据
+    private int minColNum = 0;
+    private Map<String, Integer> fieldIndexMap;
+    private List<Column.Type> fieldTypes = null;
+
+    public JDBCBatchWriter(Connection conn, TSDBPreparedStatement stmt, SchemaManager scm, int batchSize, TaskPluginCollector collector) {
+        this.conn = conn;
+        this.stmt = stmt;
+        this.scm = scm;
+        this.batchSize = batchSize;
+        this.collector = collector;
+        this.tsColIndex = scm.getTsColIndex();
+        this.fieldList = scm.getFieldList();
+        this.fieldIndexMap = scm.getFieldIndexMap();
+        this.minColNum = 1 + fieldList.size() + scm.getDynamicTagCount();
+
+    }
+
+    public void initFiledTypesAndTargetTable(List<Record> records) throws SQLException {
+        if (fieldTypes != null) {
+            return;
+        }
+        guessFieldTypes(records);
+        if (scm.shouldCreateTable()) {
+            scm.createSTable(conn, fieldTypes);
+        }
+    }
+
+    public void append(Record record) throws SQLException {
+        int columnNum = record.getColumnNumber();
+        if (columnNum < minColNum) {
+            // 实际列数小于期望列数
+            collector.collectDirtyRecord(record, Msg.get("column_number_error"));
+            return;
+        }
+        String[] tagValues = scm.getTagValuesFromRecord(record);
+        if (tagValues == null) {
+            // 标签列包含null
+            collector.collectDirtyRecord(record, Msg.get("tag_value_error"));
+            return;
+        }
+        if (!scm.hasTimestamp(record)) {
+            // 时间戳列为null或类型错误
+            collector.collectDirtyRecord(record, Msg.get("ts_value_error"));
+            return;
+        }
+        String tableName = scm.computeTableName(tagValues);
+        if (buf.containsKey(tableName)) {
+            List<Record> lis = buf.get(tableName);
+            lis.add(record);
+            if (lis.size() == batchSize) {
+                if (fieldTypes == null) {
+                    initFiledTypesAndTargetTable(lis);
+                }
+                executeBatch(tableName);
+                lis.clear();
+            }
+        } else {
+            List<Record> lis = new ArrayList<>(batchSize);
+            lis.add(record);
+            buf.put(tableName, lis);
+            tableTagValues.put(tableName, tagValues);
+        }
+    }
+
+    /**
+     * 只有String类型比较特别，测试发现值为null的列会转成String类型。所以Column的类型为String并不代表这一列的类型真的是String。
+     *
+     * @param records
+     */
+    private void guessFieldTypes(List<Record> records) {
+        fieldTypes = new ArrayList<>(fieldList.size());
+        for (int i = 0; i < fieldList.size(); ++i) {
+            int colIndex = fieldIndexMap.get(fieldList.get(i));
+            boolean ok = false;
+            for (int j = 0; j < records.size() && !ok; ++j) {
+                Column column = records.get(j).getColumn(colIndex);
+                Column.Type type = column.getType();
+                switch (type) {
+                    case LONG:
+                    case DOUBLE:
+                    case DATE:
+                    case BOOL:
+                    case BYTES:
+                        if (column.getRawData() != null) {
+                            fieldTypes.add(type);
+                            ok = true;
+                        }
+                        break;
+                    case STRING:
+                        // 只有非null且非空的String列，才会被真的当作String类型。
+                        String value = column.asString();
+                        if (value != null && !"".equals(value)) {
+                            fieldTypes.add(type);
+                            ok = true;
+                        }
+                        break;
+                    default:
+                        throw DataXException.asDataXException(TDengineWriterErrorCode.TYPE_ERROR, fieldTypes.get(i).toString());
+                }
+            }
+            if (!ok) {
+                // 根据采样的%d条数据，无法推断第%d列的数据类型
+                throw DataXException.asDataXException(TDengineWriterErrorCode.TYPE_ERROR, String.format(Msg.get("infer_column_type_error"), records.size(), i + 1));
+            }
+        }
+        LOG.info("Field Types: {}", fieldTypes);
+    }
+
+    /**
+     * 执行单表批量写入
+     *
+     * @param tableName
+     * @throws SQLException
+     */
+    private void executeBatch(String tableName) throws SQLException {
+        // 表名
+        stmt.setTableName(tableName);
+        List<Record> records = buf.get(tableName);
+        // 标签
+        String[] tagValues = tableTagValues.get(tableName);
+        LOG.debug("executeBatch {}", String.join(",", tagValues));
+        for (int i = 0; i < tagValues.length; ++i) {
+            stmt.setTagNString(i, tagValues[i]);
+        }
+        // 时间戳
+        ArrayList<Long> tsList = records.stream().map(r -> r.getColumn(tsColIndex).asDate().getTime()).collect(Collectors.toCollection(ArrayList::new));
+        stmt.setTimestamp(0, tsList);
+        // 字段
+        for (int i = 0; i < fieldList.size(); ) {
+            String fieldName = fieldList.get(i);
+            int index = fieldIndexMap.get(fieldName);
+            switch (fieldTypes.get(i)) {
+                case LONG:
+                    ArrayList<Long> lisLong = records.stream().map(r -> r.getColumn(index).asBigInteger().longValue()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setLong(++i, lisLong);
+                    break;
+                case DOUBLE:
+                    ArrayList<Double> lisDouble = records.stream().map(r -> r.getColumn(index).asDouble()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setDouble(++i, lisDouble);
+                    break;
+                case STRING:
+                    ArrayList<String> lisString = records.stream().map(r -> r.getColumn(index).asString()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setNString(++i, lisString, 64);
+                    break;
+                case DATE:
+                    ArrayList<Long> lisTs = records.stream().map(r -> r.getColumn(index).asBigInteger().longValue()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setTimestamp(++i, lisTs);
+                    break;
+                case BOOL:
+                    ArrayList<Boolean> lisBool = records.stream().map(r -> r.getColumn(index).asBoolean()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setBoolean(++i, lisBool);
+                    break;
+                case BYTES:
+                    ArrayList<String> lisBytes = records.stream().map(r -> r.getColumn(index).asString()).collect(Collectors.toCollection(ArrayList::new));
+                    stmt.setString(++i, lisBytes, 64);
+                    break;
+                default:
+                    throw DataXException.asDataXException(TDengineWriterErrorCode.TYPE_ERROR, fieldTypes.get(i).toString());
+            }
+        }
+        // 执行
+        stmt.columnDataAddBatch();
+        stmt.columnDataExecuteBatch();
+        // 更新计数器
+        sucCount += records.size();
+    }
+
+    /**
+     * 把缓存的Record全部写入
+     */
+    public void flush() throws SQLException {
+        if (fieldTypes == null) {
+            List<Record> records = new ArrayList<>();
+            for (List<Record> lis : buf.values()) {
+                records.addAll(lis);
+                if (records.size() > 100) {
+                    break;
+                }
+            }
+            if (records.size() > 0) {
+                initFiledTypesAndTargetTable(records);
+            } else {
+                return;
+            }
+        }
+        for (String tabName : buf.keySet()) {
+            if (buf.get(tabName).size() > 0) {
+                executeBatch(tabName);
+            }
+        }
+        stmt.columnDataCloseBatch();
+    }
+
+    /**
+     * @return 成功写入的数据量
+     */
+    public long getCount() {
+        return sucCount;
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/JniConnection.java

@@ -0,0 +1,89 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import java.util.Properties;
+
+public class JniConnection {
+
+    private static final long JNI_NULL_POINTER = 0L;
+    private static final int JNI_SUCCESSFUL = 0;
+    public static final String PROPERTY_KEY_CONFIG_DIR = "cfgdir";
+    public static final String PROPERTY_KEY_LOCALE = "locale";
+    public static final String PROPERTY_KEY_CHARSET = "charset";
+    public static final String PROPERTY_KEY_TIME_ZONE = "timezone";
+
+    private long conn;
+
+    static {
+        System.loadLibrary("taos");
+    }
+
+    public JniConnection(Properties props) throws Exception {
+        initImp(props.getProperty(PROPERTY_KEY_CONFIG_DIR, null));
+
+        String locale = props.getProperty(PROPERTY_KEY_LOCALE);
+        if (setOptions(0, locale) < 0) {
+            throw new Exception("Failed to set locale: " + locale + ". System default will be used.");
+        }
+        String charset = props.getProperty(PROPERTY_KEY_CHARSET);
+        if (setOptions(1, charset) < 0) {
+            throw new Exception("Failed to set charset: " + charset + ". System default will be used.");
+        }
+        String timezone = props.getProperty(PROPERTY_KEY_TIME_ZONE);
+        if (setOptions(2, timezone) < 0) {
+            throw new Exception("Failed to set timezone: " + timezone + ". System default will be used.");
+        }
+    }
+
+    public void open(String host, int port, String dbname, String user, String password) throws Exception {
+        if (this.conn != JNI_NULL_POINTER) {
+            close();
+            this.conn = JNI_NULL_POINTER;
+        }
+
+        this.conn = connectImp(host, port, dbname, user, password);
+        if (this.conn == JNI_NULL_POINTER) {
+            String errMsg = getErrMsgImp(0);
+            throw new Exception(errMsg);
+        }
+    }
+
+    public void insertOpentsdbJson(String json) throws Exception {
+        if (this.conn == JNI_NULL_POINTER) {
+            throw new Exception("JNI connection is NULL");
+        }
+
+        long result = insertOpentsdbJson(json, this.conn);
+        int errCode = getErrCodeImp(this.conn, result);
+        if (errCode != JNI_SUCCESSFUL) {
+            String errMsg = getErrMsgImp(result);
+            freeResultSetImp(this.conn, result);
+            throw new Exception(errMsg);
+        }
+        freeResultSetImp(this.conn, result);
+    }
+
+    public void close() throws Exception {
+        int code = this.closeConnectionImp(this.conn);
+        if (code != 0) {
+            throw new Exception("JNI closeConnection failed");
+        }
+        this.conn = JNI_NULL_POINTER;
+    }
+
+    private static native void initImp(String configDir);
+
+    private static native int setOptions(int optionIndex, String optionValue);
+
+    private native long connectImp(String host, int port, String dbName, String user, String password);
+
+    private native int getErrCodeImp(long connection, long pSql);
+
+    private native String getErrMsgImp(long pSql);
+
+    private native void freeResultSetImp(long connection, long pSql);
+
+    private native int closeConnectionImp(long connection);
+
+    private native long insertOpentsdbJson(String json, long pSql);
+
+}

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

@@ -0,0 +1,14 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+public class Key {
+    public static final String HOST = "host";
+    public static final String PORT = "port";
+    public static final String DBNAME = "dbName";
+    public static final String USER = "username";
+    public static final String PASSWORD = "password";
+    public static final String BATCH_SIZE = "batchSize";
+    public static final String STABLE = "stable";
+    public static final String TAG_COLUMN = "tagColumn";
+    public static final String FIELD_COLUMN = "fieldColumn";
+    public static final String TIMESTAMP_COLUMN = "timestampColumn";
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/Msg.java

@@ -0,0 +1,20 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import java.util.Locale;
+import java.util.ResourceBundle;
+
+/**
+ * i18n message util
+ */
+public class Msg {
+    private static ResourceBundle bundle;
+
+    static {
+        bundle = ResourceBundle.getBundle("tdenginewritermsg", Locale.getDefault());
+    }
+
+    public static String get(String key) {
+        return bundle.getString(key);
+    }
+
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/OpentsdbDataHandler.java

@@ -0,0 +1,99 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+public class OpentsdbDataHandler implements DataHandler {
+    private static final Logger LOG = LoggerFactory.getLogger(OpentsdbDataHandler.class);
+    private static final String DEFAULT_BATCH_SIZE = "1";
+
+    @Override
+    public long handle(RecordReceiver lineReceiver, Properties properties, TaskPluginCollector collector) {
+        // opentsdb json protocol use JNI and schemaless API to write
+        String host = properties.getProperty(Key.HOST);
+        int port = Integer.parseInt(properties.getProperty(Key.PORT));
+        String dbname = properties.getProperty(Key.DBNAME);
+        String user = properties.getProperty(Key.USER);
+        String password = properties.getProperty(Key.PASSWORD);
+
+        JniConnection conn = null;
+        long count = 0;
+        try {
+            conn = new JniConnection(properties);
+            conn.open(host, port, dbname, user, password);
+            LOG.info("TDengine connection established, host: " + host + ", port: " + port + ", dbname: " + dbname + ", user: " + user);
+            int batchSize = Integer.parseInt(properties.getProperty(Key.BATCH_SIZE, DEFAULT_BATCH_SIZE));
+            count = writeOpentsdb(lineReceiver, conn, batchSize);
+        } catch (Exception e) {
+            LOG.error(e.getMessage());
+            e.printStackTrace();
+        } finally {
+            try {
+                if (conn != null)
+                    conn.close();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            LOG.info("TDengine connection closed");
+        }
+
+        return count;
+    }
+
+    private long writeOpentsdb(RecordReceiver lineReceiver, JniConnection conn, int batchSize) {
+        long recordIndex = 1;
+        try {
+            Record record;
+            StringBuilder sb = new StringBuilder();
+            while ((record = lineReceiver.getFromReader()) != null) {
+                if (batchSize == 1) {
+                    String jsonData = recordToString(record);
+                    LOG.debug(">>> " + jsonData);
+                    conn.insertOpentsdbJson(jsonData);
+                } else if (recordIndex % batchSize == 1) {
+                    sb.append("[").append(recordToString(record)).append(",");
+                } else if (recordIndex % batchSize == 0) {
+                    sb.append(recordToString(record)).append("]");
+                    String jsonData = sb.toString();
+                    LOG.debug(">>> " + jsonData);
+                    conn.insertOpentsdbJson(jsonData);
+                    sb.delete(0, sb.length());
+                } else {
+                    sb.append(recordToString(record)).append(",");
+                }
+                recordIndex++;
+            }
+            if (sb.length() != 0 && sb.charAt(0) == '[') {
+                String jsonData = sb.deleteCharAt(sb.length() - 1).append("]").toString();
+                LOG.debug(">>> " + jsonData);
+                conn.insertOpentsdbJson(jsonData);
+            }
+        } catch (Exception e) {
+            LOG.error("TDengineWriter ERROR: " + e.getMessage());
+            throw DataXException.asDataXException(TDengineWriterErrorCode.RUNTIME_EXCEPTION, e);
+        }
+        return recordIndex - 1;
+    }
+
+    private String recordToString(Record record) {
+        int recordLength = record.getColumnNumber();
+        if (0 == recordLength) {
+            return "";
+        }
+        Column column;
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < recordLength; i++) {
+            column = record.getColumn(i);
+            sb.append(column.asString()).append("\t");
+        }
+        sb.setLength(sb.length() - 1);
+        return sb.toString();
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/SchemaManager.java

@@ -0,0 +1,271 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class SchemaManager {
+    private static final Logger LOG = LoggerFactory.getLogger(SchemaManager.class);
+
+    private String stable; // 目标超表名
+    private Map<String, String> fixedTagValue = new HashMap<>(); // 固定标签值 标签名 -> 标签值
+    private Map<String, Integer> tagIndexMap = new HashMap<>(); // 动态标签值  标签名 -> 列索引
+    private Map<String, Integer> fieldIndexMap = new HashMap<>(); // 字段名 -> 字段索引
+    private String tsColName; // 时间戳列名
+    private int tsColIndex = -1; // 时间戳列索引
+    private List<String> fieldList = new ArrayList<>();
+    private List<String> tagList = new ArrayList<>();
+    private boolean canInferSchemaFromConfig = false;
+
+
+    public SchemaManager() {
+    }
+
+    public SchemaManager(Properties properties) {
+        getFromConfig(properties);
+    }
+
+    private String mapDataxType(Column.Type type) {
+        switch (type) {
+            case LONG:
+                return "BIGINT";
+            case DOUBLE:
+                return "DOUBLE";
+            case STRING:
+                return "NCHAR(64)";
+            case DATE:
+                return "TIMESTAMP";
+            case BOOL:
+                return "BOOL";
+            case BYTES:
+                return "BINARY(64)";
+            default:
+                throw DataXException.asDataXException(TDengineWriterErrorCode.TYPE_ERROR, type.toString());
+        }
+    }
+
+    public void setStable(String stable) {
+        stable = stable;
+    }
+
+    public String getStable() {
+        return stable;
+    }
+
+    private void getFromConfig(Properties properties) {
+        stable = properties.getProperty(Key.STABLE);
+        if (stable == null) {
+            LOG.error("Config error: no stable");
+            return;
+        }
+        for (Object key : properties.keySet()) {
+            String k = (String) key;
+            String v = properties.getProperty(k);
+
+            String[] ps = k.split("\\.");
+            if (ps.length == 1) {
+                continue;
+            }
+            if (k.startsWith(Key.TAG_COLUMN)) {
+                String tagName = ps[1];
+                try {
+                    Integer tagIndex = Integer.parseInt(v);
+                    this.tagIndexMap.put(tagName, tagIndex);
+                    tagList.add(tagName);
+                } catch (NumberFormatException e) {
+                    fixedTagValue.put(tagName, v);
+                    tagList.add(tagName);
+                }
+            } else if (k.startsWith(Key.FIELD_COLUMN)) {
+                String fieldName = ps[1];
+                Integer fileIndex = Integer.parseInt(v);
+                fieldIndexMap.put(fieldName, fileIndex);
+            } else if (k.startsWith(Key.TIMESTAMP_COLUMN)) {
+                tsColName = ps[1];
+                tsColIndex = Integer.parseInt(v);
+            }
+        }
+        List<String> sortedFieldName = fieldIndexMap.entrySet().stream().sorted((x, y) -> x.getValue().compareTo(y.getValue())).map(e -> e.getKey()).collect(Collectors.toList());
+        fieldList.addAll(sortedFieldName); // 排序的目的是保证自动建表时列的顺序和输入数据的列的顺序保持一致
+        canInferSchemaFromConfig = tsColIndex > -1 && !(fixedTagValue.isEmpty() && tagIndexMap.isEmpty()) && !fieldIndexMap.isEmpty();
+        LOG.info("Config file parsed result：fixedTags=[{}] ,tags=[{}], fields=[{}], tsColName={}, tsIndex={}", String.join(",", fixedTagValue.keySet()), String.join(",", tagIndexMap.keySet()), String.join(",", fieldList), tsColName, tsColIndex);
+    }
+
+    public boolean shouldGuessSchema() {
+        return !canInferSchemaFromConfig;
+    }
+
+    public boolean shouldCreateTable() {
+        return canInferSchemaFromConfig;
+    }
+
+    public boolean configValid() {
+        boolean valid = (tagList.size() > 0 && fieldList.size() > 0 && tsColIndex > -1) || (tagList.size() == 0 && fieldList.size() == 0 && tsColIndex == -1);
+        if (!valid) {
+            LOG.error("Config error: tagColumn, fieldColumn and timestampColumn must be present together or absent together.");
+        }
+        return valid;
+    }
+
+    /**
+     * 通过执行`describe dbname.stable`命令，获取表的schema.<br/>
+     * describe命名返回有4列内容，分布是：Field,Type,Length,Note<br/>
+     *
+     * @return 成功返回true，如果超表不存在或其他错误则返回false
+     */
+    public boolean getFromDB(Connection conn) {
+        try {
+            List<String> stables = getSTables(conn);
+            if (!stables.contains(stable)) {
+                LOG.error("super table {} not exist， fail to get schema from database.", stable);
+                return false;
+            }
+        } catch (SQLException e) {
+            LOG.error(e.getMessage());
+            e.printStackTrace();
+            return false;
+        }
+        try (Statement stmt = conn.createStatement()) {
+            ResultSet rs = stmt.executeQuery("describe " + stable);
+            int colIndex = 0;
+            while (rs.next()) {
+                String name = rs.getString(1);
+                String type = rs.getString(2);
+                String note = rs.getString(4);
+                if ("TIMESTAMP".equals(type)) {
+                    tsColName = name;
+                    tsColIndex = colIndex;
+                } else if ("TAG".equals(note)) {
+                    tagIndexMap.put(name, colIndex);
+                    tagList.add(name);
+                } else {
+                    fieldIndexMap.put(name, colIndex);
+                    fieldList.add(name);
+                }
+                colIndex++;
+            }
+            LOG.info("table info：tags=[{}], fields=[{}], tsColName={}, tsIndex={}", String.join(",", tagIndexMap.keySet()), String.join(",", fieldList), tsColName, tsColIndex);
+            return true;
+        } catch (SQLException e) {
+            LOG.error(e.getMessage());
+            e.printStackTrace();
+            return false;
+        }
+    }
+
+    public static List<String> getSTables(Connection conn) throws SQLException {
+        List<String> stables = new ArrayList<>();
+        try (Statement stmt = conn.createStatement()) {
+            ResultSet rs = stmt.executeQuery("show stables");
+            while (rs.next()) {
+                String name = rs.getString(1);
+                stables.add(name);
+            }
+        }
+        return stables;
+    }
+
+    public void createSTable(Connection conn, List<Column.Type> fieldTypes) throws SQLException {
+        StringBuilder sb = new StringBuilder();
+        sb.append("CREATE STABLE IF NOT EXISTS ").append(stable).append("(");
+        sb.append(tsColName).append(" ").append("TIMESTAMP,");
+        for (int i = 0; i < fieldList.size(); ++i) {
+            String fieldName = fieldList.get(i);
+            Column.Type dxType = fieldTypes.get(i);
+            sb.append(fieldName).append(' ');
+            String tdType = mapDataxType(dxType);
+            sb.append(tdType).append(',');
+        }
+        sb.deleteCharAt(sb.length() - 1);
+        sb.append(") TAGS(");
+        for (String tagName : tagList) {
+            sb.append(tagName).append(" NCHAR(64),");
+        }
+        sb.deleteCharAt(sb.length() - 1);
+        sb.append(")");
+        String q = sb.toString();
+        LOG.info("run sql：" + q);
+        try (Statement stmt = conn.createStatement()) {
+            stmt.execute(q);
+        }
+    }
+
+    public String[] getTagValuesFromRecord(Record record) {
+        String[] tagValues = new String[tagList.size()];
+        for (int i = 0; i < tagList.size(); ++i) {
+            if (fixedTagValue.containsKey(tagList.get(i))) {
+                tagValues[i] = fixedTagValue.get(tagList.get(i));
+            } else {
+                int tagIndex = tagIndexMap.get(tagList.get(i));
+                tagValues[i] = record.getColumn(tagIndex).asString();
+            }
+            if (tagValues[i] == null) {
+                return null;
+            }
+        }
+        return tagValues;
+    }
+
+    public boolean hasTimestamp(Record record) {
+        Column column = record.getColumn(tsColIndex);
+        if (column.getType() == Column.Type.DATE && column.asDate() != null) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    public Map<String, Integer> getFieldIndexMap() {
+        return fieldIndexMap;
+    }
+
+    public List<String> getFieldList() {
+        return fieldList;
+    }
+
+    public String getJoinedFieldNames() {
+        return tsColName + ", " + String.join(", ", fieldList);
+    }
+
+    public int getTsColIndex() {
+        return tsColIndex;
+    }
+
+    public String getTagValuesPlaceHolder() {
+        return tagList.stream().map(x -> "?").collect(Collectors.joining(","));
+    }
+
+    public String getFieldValuesPlaceHolder() {
+        return "?, " + fieldList.stream().map(x -> "?").collect(Collectors.joining(", "));
+    }
+
+    /**
+     * 计算子表表名
+     * <ol>
+     * <li>将标签的value 组合成为如下的字符串: tag_value1!tag_value2!tag_value3。</li>
+     * <li>计算该字符串的 MD5 散列值 "md5_val"。</li>
+     * <li>"t_md5val"作为子表名。其中的 "t" 是固定的前缀。</li>
+     * </ol>
+     *
+     * @param tagValues
+     * @return
+     */
+    public String computeTableName(String[] tagValues) {
+        String s = String.join("!", tagValues);
+        return "t_" + DigestUtils.md5Hex(s);
+    }
+
+    public int getDynamicTagCount() {
+        return tagIndexMap.size();
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/TDengineWriter.java

@@ -0,0 +1,91 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+public class TDengineWriter extends Writer {
+
+    private static final String PEER_PLUGIN_NAME = "peerPluginName";
+
+    static {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+        } catch (ClassNotFoundException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static class Job extends Writer.Job {
+
+        private Configuration originalConfig;
+
+        @Override
+        public void init() {
+            this.originalConfig = super.getPluginJobConf();
+            this.originalConfig.set(PEER_PLUGIN_NAME, getPeerPluginName());
+        }
+
+        @Override
+        public void destroy() {
+
+        }
+
+        @Override
+        public List<Configuration> split(int mandatoryNumber) {
+            List<Configuration> writerSplitConfigs = new ArrayList<>();
+            for (int i = 0; i < mandatoryNumber; i++) {
+                writerSplitConfigs.add(this.originalConfig);
+            }
+            return writerSplitConfigs;
+        }
+    }
+
+    public static class Task extends Writer.Task {
+        private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+
+        private Configuration writerSliceConfig;
+
+        @Override
+        public void init() {
+            this.writerSliceConfig = getPluginJobConf();
+
+        }
+
+        @Override
+        public void destroy() {
+
+        }
+
+        @Override
+        public void startWrite(RecordReceiver lineReceiver) {
+            Set<String> keys = this.writerSliceConfig.getKeys();
+            Properties properties = new Properties();
+            for (String key : keys) {
+                String value = this.writerSliceConfig.getString(key);
+                properties.setProperty(key, value);
+            }
+            if (!keys.contains(Key.USER)) {
+                properties.setProperty(Key.USER, "root");
+            }
+            if (!keys.contains(Key.PASSWORD)) {
+                properties.setProperty(Key.PASSWORD, "taosdata");
+            }
+            LOG.debug("========================properties==========================\n" + properties);
+            String peerPluginName = this.writerSliceConfig.getString(PEER_PLUGIN_NAME);
+            LOG.debug("start to handle record from: " + peerPluginName);
+            DataHandler handler = DataHandlerFactory.build(peerPluginName);
+            long records = handler.handle(lineReceiver, properties, getTaskPluginCollector());
+            LOG.debug("handle data finished, records: " + records);
+        }
+
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/TDengineWriterErrorCode.java

@@ -0,0 +1,32 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import com.alibaba.datax.common.spi.ErrorCode;
+
+public enum TDengineWriterErrorCode implements ErrorCode {
+    RUNTIME_EXCEPTION("TDengineWriter-00", "运行时异常"),
+    TYPE_ERROR("TDengineWriter-00", "Datax类型无法正确映射到TDengine类型");
+
+    private final String code;
+    private final String description;
+
+    private TDengineWriterErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    @Override
+    public String getCode() {
+        return this.code;
+    }
+
+    @Override
+    public String getDescription() {
+        return this.description;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Code:[%s], Description:[%s]. ", this.code,
+                this.description);
+    }
+}

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/com_alibaba_datax_plugin_writer_tdenginewriter_JniConnection.h

@@ -0,0 +1,105 @@
+/* DO NOT EDIT THIS FILE - it is machine generated */
+#include <jni.h>
+/* Header for class com_alibaba_datax_plugin_writer_JniConnection */
+
+#ifndef _Included_com_alibaba_datax_plugin_writer_JniConnection
+#define _Included_com_alibaba_datax_plugin_writer_JniConnection
+#ifdef __cplusplus
+extern "C" {
+#endif
+#undef com_alibaba_datax_plugin_writer_JniConnection_JNI_NULL_POINTER
+#define com_alibaba_datax_plugin_writer_JniConnection_JNI_NULL_POINTER 0LL
+#undef com_alibaba_datax_plugin_writer_JniConnection_JNI_SUCCESSFUL
+#define com_alibaba_datax_plugin_writer_JniConnection_JNI_SUCCESSFUL 0L
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    initImp
+ * Signature: (Ljava/lang/String;)V
+ */
+JNIEXPORT void JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_initImp
+  (JNIEnv *, jclass, jstring);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    setOptions
+ * Signature: (ILjava/lang/String;)I
+ */
+JNIEXPORT jint JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_setOptions
+  (JNIEnv *, jclass, jint, jstring);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    getTsCharset
+ * Signature: ()Ljava/lang/String;
+ */
+JNIEXPORT jstring JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_getTsCharset
+  (JNIEnv *, jclass);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    connectImp
+ * Signature: (Ljava/lang/String;ILjava/lang/String;Ljava/lang/String;Ljava/lang/String;)J
+ */
+JNIEXPORT jlong JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_connectImp
+  (JNIEnv *, jobject, jstring, jint, jstring, jstring, jstring);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    executeQueryImp
+ * Signature: ([BJ)J
+ */
+JNIEXPORT jlong JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_executeQueryImp
+  (JNIEnv *, jobject, jbyteArray, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    getErrCodeImp
+ * Signature: (JJ)I
+ */
+JNIEXPORT jint JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_getErrCodeImp
+  (JNIEnv *, jobject, jlong, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    getErrMsgImp
+ * Signature: (J)Ljava/lang/String;
+ */
+JNIEXPORT jstring JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_getErrMsgImp
+  (JNIEnv *, jobject, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    getErrMsgByCode
+ * Signature: (J)Ljava/lang/String;
+ */
+JNIEXPORT jstring JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_getErrMsgByCode
+  (JNIEnv *, jobject, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    getAffectedRowsImp
+ * Signature: (JJ)I
+ */
+JNIEXPORT jint JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_getAffectedRowsImp
+  (JNIEnv *, jobject, jlong, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    closeConnectionImp
+ * Signature: (J)I
+ */
+JNIEXPORT jint JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_closeConnectionImp
+  (JNIEnv *, jobject, jlong);
+
+/*
+ * Class:     com_alibaba_datax_plugin_writer_JniConnection
+ * Method:    insertOpentsdbJson
+ * Signature: (Ljava/lang/String;J)J
+ */
+JNIEXPORT jlong JNICALL Java_com_alibaba_datax_plugin_writer_JniConnection_insertOpentsdbJson
+  (JNIEnv *, jobject, jstring, jlong);
+
+#ifdef __cplusplus
+}
+#endif
+#endif

tdenginewriter/src/main/resources/plugin.json

@@ -0,0 +1,9 @@
+{
+    "name": "tdenginewriter",
+    "class": "com.alibaba.datax.plugin.writer.tdenginewriter.TDengineWriter",
+    "description": {
+        "useScene": "data migration to tdengine",
+        "mechanism": "use JNI or taos-jdbc to write data to tdengine."
+    },
+    "developer": "zyyang-taosdata"
+}

tdenginewriter/src/main/resources/plugin_job_template.json

@@ -0,0 +1,24 @@
+{
+    "name": "tdenginewriter",
+    "parameter": {
+        "host": "127.0.0.1",
+        "port": 6030,
+        "dbname": "test",
+        "user": "root",
+        "password": "taosdata",
+        "batchSize": 1000,
+        "stable": "weather",
+        "tagColumn": {
+            "station": 0
+        },
+        "fieldColumn": {
+            "latitude": 1,
+            "longtitude": 2,
+            "tmax": 4,
+            "tmin": 5
+        },
+        "timestampColumn":{
+            "date": 3
+        }
+    }
+}

tdenginewriter/src/main/resources/tdenginewritermsg.properties

@@ -0,0 +1,6 @@
+try_get_schema_fromdb=fail to get  structure info of target table from configure file and will try to get it from database
+batch_size_too_small='batchSize' is too small, please increase it and try again
+column_number_error=number of columns is less than expected
+tag_value_error=tag columns include 'null' value
+ts_value_error=timestamp column type error or null
+infer_column_type_error=fail to infer column type: sample count %d, column index %d

tdenginewriter/src/main/resources/tdenginewritermsg_en_US.properties

@@ -0,0 +1,6 @@
+try_get_schema_fromdb=fail to get  structure info of target table from configure file and will try to get it from database
+batch_size_too_small='batchSize' is too small, please increase it and try again
+column_number_error=number of columns is less than expected
+tag_value_error=tag columns include 'null' value
+ts_value_error=timestamp column type error or null
+infer_column_type_error=fail to infer column type: sample count %d, column index %d

tdenginewriter/src/main/resources/tdenginewritermsg_zh_CN.properties

@@ -0,0 +1,6 @@
+try_get_schema_fromdb=\u65e0\u6cd5\u4ece\u914d\u7f6e\u6587\u4ef6\u83b7\u53d6\u8868\u7ed3\u6784\u4fe1\u606f\uff0c\u5c1d\u8bd5\u4ece\u6570\u636e\u5e93\u83b7\u53d6
+batch_size_too_small=batchSize\u592a\u5c0f\uff0c\u4f1a\u589e\u52a0\u81ea\u52a8\u7c7b\u578b\u63a8\u65ad\u9519\u8bef\u7684\u6982\u7387\uff0c\u5efa\u8bae\u6539\u5927\u540e\u91cd\u8bd5
+column_number_error=\u5b9e\u9645\u5217\u6570\u5c0f\u4e8e\u671f\u671b\u5217\u6570
+tag_value_error=\u6807\u7b7e\u5217\u5305\u542bnull
+ts_value_error=\u65f6\u95f4\u6233\u5217\u4e3anull\u6216\u7c7b\u578b\u9519\u8bef
+infer_column_type_error=\u6839\u636e\u91c7\u6837\u7684%d\u6761\u6570\u636e\uff0c\u65e0\u6cd5\u63a8\u65ad\u7b2c%d\u5217\u7684\u6570\u636e\u7c7b\u578b

tdenginewriter/src/test/java/com/alibaba/datax/plugin/writer/tdenginewriter/JniConnectionTest.java

@@ -0,0 +1,21 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+public class JniConnectionTest {
+
+    @Test
+    public void test() throws Exception {
+        JniConnection connection = new JniConnection(new Properties());
+
+        connection.open("192.168.56.105", 6030, "test", "root", "taosdata");
+
+        String json = "{\"metric\":\"weather_temperature\",\"timestamp\":1609430400000,\"value\":123,\"tags\":{\"location\":\"beijing\",\"id\":\"t123\"}}";
+        connection.insertOpentsdbJson(json);
+
+        connection.close();
+    }
+
+}

tdenginewriter/src/test/java/com/alibaba/datax/plugin/writer/tdenginewriter/MessageTest.java

@@ -0,0 +1,25 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import org.junit.Test;
+
+import java.util.Locale;
+import java.util.ResourceBundle;
+
+import org.junit.Assert;
+
+public class MessageTest {
+    @Test
+    public void testChineseMessage() {
+        Locale local = new Locale("zh", "CN");
+        ResourceBundle bundle = ResourceBundle.getBundle("tdenginewritermsg", local);
+        String msg = bundle.getString("try_get_schema_fromdb");
+        Assert.assertEquals("无法从配置文件获取表结构信息，尝试从数据库获取", msg);
+    }
+
+    @Test
+    public void  testDefaultMessage() {
+        ResourceBundle bundle = ResourceBundle.getBundle("tdenginewritermsg", Locale.getDefault());
+        String msg = bundle.getString("try_get_schema_fromdb");
+        System.out.println(msg);
+    }
+}

tdenginewriter/src/test/java/com/alibaba/datax/plugin/writer/tdenginewriter/TDengineWriterTest.java

@@ -0,0 +1,31 @@
+package com.alibaba.datax.plugin.writer.tdenginewriter;
+
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+public class TDengineWriterTest {
+
+
+    @Test
+    public void testGetSchema() throws ClassNotFoundException, SQLException {
+        Class.forName("com.taosdata.jdbc.TSDBDriver");
+        String jdbcUrl = String.format("jdbc:TAOS://%s:%s/%s?user=%s&password=%s", "wozai.fun", "6030", "test", "root", "taosdata");
+        Connection conn = DriverManager.getConnection(jdbcUrl);
+        SchemaManager schemaManager = new SchemaManager();
+        schemaManager.setStable("test1");
+        schemaManager.getFromDB(conn);
+    }
+
+    @Test
+    public void dropTestTable() throws ClassNotFoundException, SQLException {
+        Class.forName("com.taosdata.jdbc.TSDBDriver");
+        String jdbcUrl = String.format("jdbc:TAOS://%s:%s/%s?user=%s&password=%s", "wozai.fun", "6030", "test", "root", "taosdata");
+        Connection conn = DriverManager.getConnection(jdbcUrl);
+        Statement stmt = conn.createStatement();
+        stmt.execute("drop table market_snapshot");
+    }
+}

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/Constant.java

@@ -16,6 +16,8 @@ public final class Constant {
     static final String DEFAULT_DATA_FORMAT = "yyyy-MM-dd HH:mm:ss";
 
     public static final String METRIC_SPECIFY_KEY = "__metric__";
+    public static final String METRIC_SPECIFY_KEY_PREFIX = METRIC_SPECIFY_KEY + ".";
+    public static final int METRIC_SPECIFY_KEY_PREFIX_LENGTH = METRIC_SPECIFY_KEY_PREFIX.length();
     public static final String TS_SPECIFY_KEY = "__ts__";
     public static final String VALUE_SPECIFY_KEY = "__value__";
 

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/Key.java

@@ -17,14 +17,19 @@ public class Key {
     // RDB for MySQL / ADB etc.
     static final String SINK_DB_TYPE = "sinkDbType";
     static final String ENDPOINT = "endpoint";
+    static final String USERNAME = "username";
+    static final String PASSWORD = "password";
     static final String COLUMN = "column";
     static final String METRIC = "metric";
     static final String FIELD = "field";
     static final String TAG = "tag";
+    static final String COMBINE = "combine";
     static final String INTERVAL_DATE_TIME = "splitIntervalMs";
     static final String BEGIN_DATE_TIME = "beginDateTime";
     static final String END_DATE_TIME = "endDateTime";
+    static final String HINT = "hint";
 
+    static final Boolean COMBINE_DEFAULT_VALUE = false;
     static final Integer INTERVAL_DATE_TIME_DEFAULT_VALUE = 60;
     static final String TYPE_DEFAULT_VALUE = "TSDB";
     static final Set<String> TYPE_SET = new HashSet<>();

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/TSDBReader.java

@@ -60,6 +60,15 @@ public class TSDBReader extends Reader {
                         "The parameter [" + Key.ENDPOINT + "] is not set.");
             }
 
+            String username = originalConfig.getString(Key.USERNAME, null);
+            if (StringUtils.isBlank(username)) {
+                LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
+            }
+            String password = originalConfig.getString(Key.PASSWORD, null);
+            if (StringUtils.isBlank(password)) {
+                LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
+            }
+
             // tagK / field could be empty
             if ("TSDB".equals(type)) {
                 List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
@@ -76,7 +85,14 @@ public class TSDBReader extends Reader {
                             "The parameter [" + Key.COLUMN + "] is not set.");
                 }
                 for (String specifyKey : Constant.MUST_CONTAINED_SPECIFY_KEYS) {
-                    if (!columns.contains(specifyKey)) {
+                    boolean containSpecifyKey = false;
+                    for (String column : columns) {
+                        if (column.startsWith(specifyKey)) {
+                            containSpecifyKey = true;
+                            break;
+                        }
+                    }
+                    if (!containSpecifyKey) {
                         throw DataXException.asDataXException(
                                 TSDBReaderErrorCode.ILLEGAL_VALUE,
                                 "The parameter [" + Key.COLUMN + "] should contain "
@@ -99,6 +115,8 @@ public class TSDBReader extends Reader {
                         "The parameter [" + Key.INTERVAL_DATE_TIME + "] should be great than zero.");
             }
 
+            Boolean isCombine = originalConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
+
             SimpleDateFormat format = new SimpleDateFormat(Constant.DEFAULT_DATA_FORMAT);
             String startTime = originalConfig.getString(Key.BEGIN_DATE_TIME);
             Long startDate;
@@ -168,14 +186,14 @@ public class TSDBReader extends Reader {
                 startTime = format.parse(originalConfig.getString(Key.BEGIN_DATE_TIME)).getTime();
             } catch (ParseException e) {
                 throw DataXException.asDataXException(
-                        TSDBReaderErrorCode.ILLEGAL_VALUE, "解析[" + Key.BEGIN_DATE_TIME + "]失败.", e);
+                        TSDBReaderErrorCode.ILLEGAL_VALUE, "Analysis [" + Key.BEGIN_DATE_TIME + "] failed.", e);
             }
             long endTime;
             try {
                 endTime = format.parse(originalConfig.getString(Key.END_DATE_TIME)).getTime();
             } catch (ParseException e) {
                 throw DataXException.asDataXException(
-                        TSDBReaderErrorCode.ILLEGAL_VALUE, "解析[" + Key.END_DATE_TIME + "]失败.", e);
+                        TSDBReaderErrorCode.ILLEGAL_VALUE, "Analysis [" + Key.END_DATE_TIME + "] failed.", e);
             }
             if (TimeUtils.isSecond(startTime)) {
                 startTime *= 1000;
@@ -186,13 +204,14 @@ public class TSDBReader extends Reader {
             DateTime startDateTime = new DateTime(TimeUtils.getTimeInHour(startTime));
             DateTime endDateTime = new DateTime(TimeUtils.getTimeInHour(endTime));
 
+            final Boolean isCombine = originalConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
+
             if ("TSDB".equals(type)) {
-                // split by metric
-                for (String column : columns4TSDB) {
+                if (isCombine) {
                     // split by time in hour
                     while (startDateTime.isBefore(endDateTime)) {
                         Configuration clone = this.originalConfig.clone();
-                        clone.set(Key.COLUMN, Collections.singletonList(column));
+                        clone.set(Key.COLUMN, columns4TSDB);
 
                         clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
                         startDateTime = startDateTime.plusMillis(splitIntervalMs);
@@ -202,15 +221,30 @@ public class TSDBReader extends Reader {
 
                         LOG.info("Configuration: {}", JSON.toJSONString(clone));
                     }
+                } else {
+                    // split by time in hour
+                    while (startDateTime.isBefore(endDateTime)) {
+                        // split by metric
+                        for (String column : columns4TSDB) {
+                            Configuration clone = this.originalConfig.clone();
+                            clone.set(Key.COLUMN, Collections.singletonList(column));
+
+                            clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
+                            startDateTime = startDateTime.plusMillis(splitIntervalMs);
+                            // Make sure the time interval is [start, end).
+                            clone.set(Key.END_DATE_TIME, startDateTime.getMillis() - 1);
+                            configurations.add(clone);
+
+                            LOG.info("Configuration: {}", JSON.toJSONString(clone));
+                        }
+                    }
                 }
             } else {
-                // split by metric
-                for (String metric : metrics) {
-                    // split by time in hour
+                if (isCombine) {
                     while (startDateTime.isBefore(endDateTime)) {
                         Configuration clone = this.originalConfig.clone();
                         clone.set(Key.COLUMN, columns4RDB);
-                        clone.set(Key.METRIC, Collections.singletonList(metric));
+                        clone.set(Key.METRIC, metrics);
 
                         clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
                         startDateTime = startDateTime.plusMillis(splitIntervalMs);
@@ -220,6 +254,24 @@ public class TSDBReader extends Reader {
 
                         LOG.info("Configuration: {}", JSON.toJSONString(clone));
                     }
+                } else {
+                    // split by time in hour
+                    while (startDateTime.isBefore(endDateTime)) {
+                        // split by metric
+                        for (String metric : metrics) {
+                            Configuration clone = this.originalConfig.clone();
+                            clone.set(Key.COLUMN, columns4RDB);
+                            clone.set(Key.METRIC, Collections.singletonList(metric));
+
+                            clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
+                            startDateTime = startDateTime.plusMillis(splitIntervalMs);
+                            // Make sure the time interval is [start, end).
+                            clone.set(Key.END_DATE_TIME, startDateTime.getMillis() - 1);
+                            configurations.add(clone);
+
+                            LOG.info("Configuration: {}", JSON.toJSONString(clone));
+                        }
+                    }
                 }
             }
             return configurations;
@@ -247,6 +299,8 @@ public class TSDBReader extends Reader {
         private TSDBConnection conn;
         private Long startTime;
         private Long endTime;
+        private Boolean isCombine;
+        private Map<String, Object> hint;
 
         @Override
         public void init() {
@@ -265,11 +319,16 @@ public class TSDBReader extends Reader {
             this.tags = readerSliceConfig.getMap(Key.TAG);
 
             String address = readerSliceConfig.getString(Key.ENDPOINT);
+            String username = readerSliceConfig.getString(Key.USERNAME);
+            String password = readerSliceConfig.getString(Key.PASSWORD);
 
-            conn = new TSDBConnection(address);
+            conn = new TSDBConnection(address, username, password);
 
             this.startTime = readerSliceConfig.getLong(Key.BEGIN_DATE_TIME);
             this.endTime = readerSliceConfig.getLong(Key.END_DATE_TIME);
+
+            this.isCombine = readerSliceConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
+            this.hint = readerSliceConfig.getMap(Key.HINT);
         }
 
         @Override
@@ -283,29 +342,35 @@ public class TSDBReader extends Reader {
                 if ("TSDB".equals(type)) {
                     for (String metric : columns4TSDB) {
                         final Map<String, String> tags = this.tags == null ?
-                                    null : (Map<String, String>) this.tags.get(metric);
+                                null : (Map<String, String>) this.tags.get(metric);
                         if (fields == null || !fields.containsKey(metric)) {
-                            conn.sendDPs(metric, tags, this.startTime, this.endTime, recordSender);
+                            conn.sendDPs(metric, tags, this.startTime, this.endTime, recordSender, hint);
                         } else {
                             conn.sendDPs(metric, (List<String>) fields.get(metric),
-                                    tags, this.startTime, this.endTime, recordSender);
+                                    tags, this.startTime, this.endTime, recordSender, hint);
                         }
                     }
                 } else {
-                    for (String metric : metrics) {
+                    if (isCombine) {
                         final Map<String, String> tags = this.tags == null ?
-                                null : (Map<String, String>) this.tags.get(metric);
-                        if (fields == null || !fields.containsKey(metric)) {
-                            conn.sendRecords(metric, tags, startTime, endTime, columns4RDB, recordSender);
-                        } else {
-                            conn.sendRecords(metric, (List<String>) fields.get(metric),
-                                    tags, startTime, endTime, columns4RDB, recordSender);
+                                null : (Map<String, String>) this.tags.get(metrics.get(0));
+                        conn.sendRecords(metrics, tags, startTime, endTime, columns4RDB, recordSender, hint);
+                    } else {
+                        for (String metric : metrics) {
+                            final Map<String, String> tags = this.tags == null ?
+                                    null : (Map<String, String>) this.tags.get(metric);
+                            if (fields == null || !fields.containsKey(metric)) {
+                                conn.sendRecords(metric, tags, startTime, endTime, columns4RDB, isCombine, recordSender, hint);
+                            } else {
+                                conn.sendRecords(metric, (List<String>) fields.get(metric),
+                                        tags, startTime, endTime, columns4RDB, recordSender, hint);
+                            }
                         }
                     }
                 }
             } catch (Exception e) {
                 throw DataXException.asDataXException(
-                        TSDBReaderErrorCode.ILLEGAL_VALUE, "获取或发送数据点的过程中出错！", e);
+                        TSDBReaderErrorCode.ILLEGAL_VALUE, "Error in getting or sending data point！", e);
             }
         }
 

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/Connection4TSDB.java

@@ -22,6 +22,20 @@ public interface Connection4TSDB {
      */
     String address();
 
+    /**
+     * Get the address of Database.
+     *
+     * @return host+ip
+     */
+    String username();
+
+    /**
+     * Get the address of Database.
+     *
+     * @return host+ip
+     */
+    String password();
+
     /**
      * Get the version of Database.
      *
@@ -46,22 +60,27 @@ public interface Connection4TSDB {
     /**
      * Send data points for TSDB with single field.
      */
-    void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception;
+    void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception;
 
     /**
      * Send data points for TSDB with multi fields.
      */
-    void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception;
+    void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception;
 
     /**
      * Send data points for RDB with single field.
      */
-    void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception;
+    void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, Boolean isCombine, RecordSender recordSender, Map<String, Object> hint) throws Exception;
 
     /**
      * Send data points for RDB with multi fields.
      */
-    void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception;
+    void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception;
+
+    /**
+     * Send data points for RDB with single fields on combine mode.
+     */
+    void sendRecords(List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception;
 
     /**
      * Put data point.

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBConnection.java

@@ -19,9 +19,13 @@ import java.util.Map;
 public class TSDBConnection implements Connection4TSDB {
 
     private String address;
+    private String username;
+    private String password;
 
-    public TSDBConnection(String address) {
+    public TSDBConnection(String address, String username, String password) {
         this.address = address;
+        this.username = username;
+        this.password = password;
     }
 
     @Override
@@ -29,14 +33,24 @@ public class TSDBConnection implements Connection4TSDB {
         return address;
     }
 
+    @Override
+    public String username() {
+        return username;
+    }
+
+    @Override
+    public String password() {
+        return password;
+    }
+
     @Override
     public String version() {
-        return TSDBUtils.version(address);
+        return TSDBUtils.version(address, username, password);
     }
 
     @Override
     public String config() {
-        return TSDBUtils.config(address);
+        return TSDBUtils.config(address, username, password);
     }
 
     @Override
@@ -45,23 +59,28 @@ public class TSDBConnection implements Connection4TSDB {
     }
 
     @Override
-    public void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception {
-        TSDBDump.dump4TSDB(this, metric, tags, start, end, recordSender);
+    public void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception {
+        TSDBDump.dump4TSDB(this, metric, tags, start, end, recordSender, hint);
     }
 
     @Override
-    public void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception {
-        TSDBDump.dump4TSDB(this, metric, fields, tags, start, end, recordSender);
+    public void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception {
+        TSDBDump.dump4TSDB(this, metric, fields, tags, start, end, recordSender, hint);
     }
 
     @Override
-    public void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception {
-        TSDBDump.dump4RDB(this, metric, tags, start, end, columns4RDB, recordSender);
+    public void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, Boolean isCombine, RecordSender recordSender, Map<String, Object> hint) throws Exception {
+        TSDBDump.dump4RDB(this, metric, tags, start, end, columns4RDB, recordSender, hint);
     }
 
     @Override
-    public void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception {
-        TSDBDump.dump4RDB(this, metric, fields, tags, start, end, columns4RDB, recordSender);
+    public void sendRecords(List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception {
+        TSDBDump.dump4RDB(this, metrics, tags, start, end, columns4RDB, recordSender, hint);
+    }
+
+    @Override
+    public void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception {
+        TSDBDump.dump4RDB(this, metric, fields, tags, start, end, columns4RDB, recordSender, hint);
     }
 
     @Override

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBDump.java

@@ -9,10 +9,9 @@ import com.alibaba.fastjson.parser.Feature;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+
+import static com.alibaba.datax.plugin.reader.tsdbreader.Constant.METRIC_SPECIFY_KEY_PREFIX_LENGTH;
 
 /**
  * Copyright @ 2019 alibaba.com
@@ -37,10 +36,10 @@ final class TSDBDump {
     }
 
     static void dump4TSDB(TSDBConnection conn, String metric, Map<String, String> tags,
-                          Long start, Long end, RecordSender sender) throws Exception {
+                          Long start, Long end, RecordSender sender, Map<String, Object> hint) throws Exception {
         LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
 
-        String res = queryRange4SingleField(conn, metric, tags, start, end);
+        String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
         List<String> dps = getDps4TSDB(metric, res);
         if (dps == null || dps.isEmpty()) {
             return;
@@ -49,10 +48,10 @@ final class TSDBDump {
     }
 
     static void dump4TSDB(TSDBConnection conn, String metric, List<String> fields, Map<String, String> tags,
-                          Long start, Long end, RecordSender sender) throws Exception {
+                          Long start, Long end, RecordSender sender, Map<String, Object> hint) throws Exception {
         LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
 
-        String res = queryRange4MultiFields(conn, metric, fields, tags, start, end);
+        String res = queryRange4MultiFields(conn, metric, fields, tags, start, end, hint);
         List<String> dps = getDps4TSDB(metric, fields, res);
         if (dps == null || dps.isEmpty()) {
             return;
@@ -61,10 +60,10 @@ final class TSDBDump {
     }
 
     static void dump4RDB(TSDBConnection conn, String metric, Map<String, String> tags,
-                         Long start, Long end, List<String> columns4RDB, RecordSender sender) throws Exception {
+                         Long start, Long end, List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
         LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
 
-        String res = queryRange4SingleField(conn, metric, tags, start, end);
+        String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
         List<DataPoint4TSDB> dps = getDps4RDB(metric, res);
         if (dps == null || dps.isEmpty()) {
             return;
@@ -92,12 +91,71 @@ final class TSDBDump {
         }
     }
 
+    public static void dump4RDB(TSDBConnection conn, List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
+        LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metrics, start, end);
+
+        List<DataPoint4TSDB> dps = new LinkedList<>();
+        for (String metric : metrics) {
+            String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
+            final List<DataPoint4TSDB> dpList = getDps4RDB(metric, res);
+            if (dpList == null || dpList.isEmpty()) {
+                continue;
+            }
+            dps.addAll(dpList);
+        }
+        if (dps.isEmpty()) {
+            return;
+        }
+        Map<Long, Map<String, DataPoint4TSDB>> dpsCombinedByTs = new LinkedHashMap<>();
+        for (DataPoint4TSDB dp : dps) {
+            final long ts = dp.getTimestamp();
+            final Map<String, DataPoint4TSDB> dpsWithSameTs = dpsCombinedByTs.computeIfAbsent(ts, k -> new LinkedHashMap<>());
+            dpsWithSameTs.put(dp.getMetric(), dp);
+        }
+
+        for (Map.Entry<Long, Map<String, DataPoint4TSDB>> entry : dpsCombinedByTs.entrySet()) {
+            final Long ts = entry.getKey();
+            final Map<String, DataPoint4TSDB> metricAndDps = entry.getValue();
+            final Record record = sender.createRecord();
+            DataPoint4TSDB tmpDp = null;
+
+            for (final String column : columns4RDB) {
+                if (column.startsWith(Constant.METRIC_SPECIFY_KEY)) {
+                    final String m = column.substring(METRIC_SPECIFY_KEY_PREFIX_LENGTH);
+                    tmpDp = metricAndDps.get(m);
+                    if (tmpDp == null) {
+                        continue;
+                    }
+                    record.addColumn(getColumn(tmpDp.getValue()));
+                } else if (Constant.TS_SPECIFY_KEY.equals(column)) {
+                    record.addColumn(new LongColumn(ts));
+                } else if (Constant.VALUE_SPECIFY_KEY.equals(column)) {
+                    // combine 模式下，不应该定义 __value__ 字段，因为 __metric__.xxx 字段会输出对应的 value 值
+                    throw new RuntimeException("The " + Constant.VALUE_SPECIFY_KEY +
+                            " column should not be specified in combine mode!");
+                } else {
+                    // combine 模式下，应该确保 __metric__.xxx 字段的定义，放在 column 数组的最前面，以保证获取到 metric
+                    if (tmpDp == null) {
+                        throw new RuntimeException("These " + Constant.METRIC_SPECIFY_KEY_PREFIX +
+                                " column should be placed first in the column array in combine mode!");
+                    }
+                    final Object tagv = tmpDp.getTags().get(column);
+                    if (tagv == null) {
+                        continue;
+                    }
+                    record.addColumn(getColumn(tagv));
+                }
+            }
+            sender.sendToWriter(record);
+        }
+    }
+
     static void dump4RDB(TSDBConnection conn, String metric, List<String> fields,
                          Map<String, String> tags, Long start, Long end,
-                         List<String> columns4RDB, RecordSender sender) throws Exception {
+                         List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
         LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
 
-        String res = queryRange4MultiFields(conn, metric, fields, tags, start, end);
+        String res = queryRange4MultiFields(conn, metric, fields, tags, start, end, hint);
         List<DataPoint4TSDB> dps = getDps4RDB(metric, fields, res);
         if (dps == null || dps.isEmpty()) {
             return;
@@ -131,14 +189,16 @@ final class TSDBDump {
             valueColumn = new LongColumn((Long) value);
         } else if (value instanceof String) {
             valueColumn = new StringColumn((String) value);
+        } else if (value instanceof Integer) {
+            valueColumn = new LongColumn(((Integer)value).longValue());
         } else {
-            throw new Exception(String.format("value 不支持类型: [%s]", value.getClass().getSimpleName()));
+            throw new Exception(String.format("value not supported type: [%s]", value.getClass().getSimpleName()));
         }
         return valueColumn;
     }
 
     private static String queryRange4SingleField(TSDBConnection conn, String metric, Map<String, String> tags,
-                                                 Long start, Long end) throws Exception {
+                                                 Long start, Long end, Map<String, Object> hint) throws Exception {
         String tagKV = getFilterByTags(tags);
         String body = "{\n" +
                 "  \"start\": " + start + ",\n" +
@@ -148,14 +208,15 @@ final class TSDBDump {
                 "      \"aggregator\": \"none\",\n" +
                 "      \"metric\": \"" + metric + "\"\n" +
                 (tagKV == null ? "" : tagKV) +
+                (hint == null ? "" : (", \"hint\": " + JSON.toJSONString(hint))) +
                 "    }\n" +
                 "  ]\n" +
                 "}";
-        return HttpUtils.post(conn.address() + QUERY, body);
+        return HttpUtils.post(conn.address() + QUERY, conn.username(), conn.password(), body);
     }
 
     private static String queryRange4MultiFields(TSDBConnection conn, String metric, List<String> fields,
-                                                 Map<String, String> tags, Long start, Long end) throws Exception {
+                                                 Map<String, String> tags, Long start, Long end, Map<String, Object> hint) throws Exception {
         // fields
         StringBuilder fieldBuilder = new StringBuilder();
         fieldBuilder.append("\"fields\":[");
@@ -177,10 +238,11 @@ final class TSDBDump {
                 "      \"metric\": \"" + metric + "\",\n" +
                 fieldBuilder.toString() +
                 (tagKV == null ? "" : tagKV) +
+                (hint == null ? "" : (", \"hint\": " + JSON.toJSONString(hint))) +
                 "    }\n" +
                 "  ]\n" +
                 "}";
-        return HttpUtils.post(conn.address() + QUERY_MULTI_FIELD, body);
+        return HttpUtils.post(conn.address() + QUERY_MULTI_FIELD, conn.username(), conn.password(), body);
     }
 
     private static String getFilterByTags(Map<String, String> tags) {

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/util/HttpUtils.java

@@ -1,11 +1,13 @@
 package com.alibaba.datax.plugin.reader.tsdbreader.util;
 
 import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.client.fluent.Content;
 import org.apache.http.client.fluent.Request;
 import org.apache.http.entity.ContentType;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Base64;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -22,13 +24,18 @@ public final class HttpUtils {
     public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
     public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
 
+    private static final String CREDENTIALS_FORMAT = "%s:%s";
+    private static final String BASIC_AUTHENTICATION_FORMAT = "Basic %s";
+
     private HttpUtils() {
     }
 
-    public static String get(String url) throws Exception {
-        Content content = Request.Get(url)
+    public static String get(String url, String username, String password) throws Exception {
+        final Request request = Request.Get(url)
                 .connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL)
-                .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL)
+                .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+        addAuth(request, username, password);
+        Content content = request
                 .execute()
                 .returnContent();
         if (content == null) {
@@ -37,24 +44,21 @@ public final class HttpUtils {
         return content.asString(StandardCharsets.UTF_8);
     }
 
-    public static String post(String url, Map<String, Object> params) throws Exception {
-        return post(url, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+    public static String post(String url, String username, String password, Map<String, Object> params) throws Exception {
+        return post(url, username, password, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
     }
 
-    public static String post(String url, String params) throws Exception {
-        return post(url, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+    public static String post(String url, String username, String password, String params) throws Exception {
+        return post(url, username, password, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
     }
 
-    public static String post(String url, Map<String, Object> params,
+    public static String post(String url, String username, String password, String params,
                               int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
-        return post(url, JSON.toJSONString(params), connectTimeoutInMill, socketTimeoutInMill);
-    }
-
-    public static String post(String url, String params,
-                              int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
-        Content content = Request.Post(url)
+        Request request = Request.Post(url)
                 .connectTimeout(connectTimeoutInMill)
-                .socketTimeout(socketTimeoutInMill)
+                .socketTimeout(socketTimeoutInMill);
+        addAuth(request, username, password);
+        Content content = request
                 .addHeader("Content-Type", "application/json")
                 .bodyString(params, ContentType.APPLICATION_JSON)
                 .execute()
@@ -64,4 +68,20 @@ public final class HttpUtils {
         }
         return content.asString(StandardCharsets.UTF_8);
     }
+
+    private static void addAuth(Request request, String username, String password) {
+        String authorization = generateHttpAuthorization(username, password);
+        if (authorization != null) {
+            request.setHeader("Authorization", authorization);
+        }
+    }
+
+    private static String generateHttpAuthorization(String username, String password) {
+        if (StringUtils.isBlank(username) || StringUtils.isBlank(password)) {
+            return null;
+        }
+        String credentials = String.format(CREDENTIALS_FORMAT, username, password);
+        credentials = Base64.getEncoder().encodeToString(credentials.getBytes());
+        return String.format(BASIC_AUTHENTICATION_FORMAT, credentials);
+    }
 }

tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/util/TSDBUtils.java

@@ -1,11 +1,5 @@
 package com.alibaba.datax.plugin.reader.tsdbreader.util;
 
-import com.alibaba.datax.plugin.reader.tsdbreader.conn.DataPoint4TSDB;
-import com.alibaba.fastjson.JSON;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
 
 /**
  * Copyright @ 2019 alibaba.com
@@ -17,52 +11,28 @@ import java.util.List;
  */
 public final class TSDBUtils {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(TSDBUtils.class);
-
     private TSDBUtils() {
     }
 
-    public static String version(String address) {
+    public static String version(String address, String username, String password) {
         String url = String.format("%s/api/version", address);
         String rsp;
         try {
-            rsp = HttpUtils.get(url);
+            rsp = HttpUtils.get(url, username, password);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
         return rsp;
     }
 
-    public static String config(String address) {
+    public static String config(String address, String username, String password) {
         String url = String.format("%s/api/config", address);
         String rsp;
         try {
-            rsp = HttpUtils.get(url);
+            rsp = HttpUtils.get(url, username, password);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
         return rsp;
     }
-
-    public static boolean put(String address, List<DataPoint4TSDB> dps) {
-        return put(address, JSON.toJSON(dps));
-    }
-
-    public static boolean put(String address, DataPoint4TSDB dp) {
-        return put(address, JSON.toJSON(dp));
-    }
-
-    private static boolean put(String address, Object o) {
-        String url = String.format("%s/api/put", address);
-        String rsp;
-        try {
-            rsp = HttpUtils.post(url, o.toString());
-            // If successful, the returned content should be null.
-            assert rsp == null;
-        } catch (Exception e) {
-            LOGGER.error("Address: {}, DataPoints: {}", url, o);
-            throw new RuntimeException(e);
-        }
-        return true;
-    }
 }

tsdbreader/src/test/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBConnectionTest.java

@@ -19,12 +19,12 @@ public class TSDBConnectionTest {
 
     @Test
     public void testVersion() {
-        String version = new TSDBConnection(TSDB_ADDRESS).version();
+        String version = new TSDBConnection(TSDB_ADDRESS,null,null).version();
         Assert.assertNotNull(version);
     }
 
     @Test
     public void testIsSupported() {
-        Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS).isSupported());
+        Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS,null,null).isSupported());
     }
 }

tsdbreader/src/test/java/com/alibaba/datax/plugin/reader/tsdbreader/util/HttpUtilsTest.java

@@ -1,39 +0,0 @@
-package com.alibaba.datax.plugin.reader.tsdbreader.util;
-
-import org.junit.Assert;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Copyright @ 2019 alibaba.com
- * All right reserved.
- * Function：HttpUtils Test
- *
- * @author Benedict Jin
- * @since 2019-10-21
- */
-@Ignore
-public class HttpUtilsTest {
-
-    @Test
-    public void testSimpleCase() throws Exception {
-        String url = "https://httpbin.org/post";
-        Map<String, Object> params = new HashMap<>();
-        params.put("foo", "bar");
-
-        String rsp = HttpUtils.post(url, params);
-        System.out.println(rsp);
-        Assert.assertNotNull(rsp);
-    }
-
-    @Test
-    public void testGet() throws Exception {
-        String url = String.format("%s/api/version", Const.TSDB_ADDRESS);
-        String rsp = HttpUtils.get(url);
-        System.out.println(rsp);
-        Assert.assertNotNull(rsp);
-    }
-}

tsdbwriter/pom.xml

@@ -91,6 +91,13 @@
             <version>${fastjson.version}</version>
         </dependency>
 
+        <!-- tsdb -->
+        <dependency>
+            <groupId>com.aliyun</groupId>
+            <artifactId>hitsdb-client</artifactId>
+            <version>0.4.0-SNAPSHOT</version>
+        </dependency>
+
         <!-- test -->
         <dependency>
             <groupId>junit</groupId>

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/conn/Connection4TSDB.java

@@ -21,6 +21,28 @@ public interface Connection4TSDB {
      */
     String address();
 
+    /**
+     * Get the setted database name.
+     *
+     * @return database
+     */
+    String database();
+
+
+    /**
+     * Get the username of Database.
+     *
+     * @return username
+     */
+    String username();
+
+    /**
+     * Get the password of Database.
+     *
+     * @return password
+     */
+    String password();
+
     /**
      * Get the version of Database.
      *
@@ -69,17 +91,25 @@ public interface Connection4TSDB {
     boolean put(List<DataPoint4TSDB> dps);
 
     /**
-     * Put data points.
+     * Put data points with single field.
      *
      * @param dps data points
      * @return whether the data point is written successfully
      */
     boolean put(String dps);
 
+    /**
+     * Put data points with multi fields.
+     *
+     * @param dps data points
+     * @return whether the data point is written successfully
+     */
+    boolean mput(String dps);
+
     /**
      * Whether current version is supported.
      *
      * @return true: supported; false: not yet!
      */
     boolean isSupported();
-}
+}

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/conn/TSDBConnection.java

@@ -18,12 +18,18 @@ import java.util.List;
 public class TSDBConnection implements Connection4TSDB {
 
     private String address;
+    private String username;
+    private String password;
+    private String database;
 
-    public TSDBConnection(String address) {
+    public TSDBConnection(String address, String database, String username, String password) {
         if (StringUtils.isBlank(address)) {
             throw new RuntimeException("TSDBConnection init failed because address is blank!");
         }
         this.address = address;
+        this.database = database;
+        this.username = username;
+        this.password = password;
     }
 
     @Override
@@ -31,14 +37,29 @@ public class TSDBConnection implements Connection4TSDB {
         return address;
     }
 
+    @Override
+    public String username() {
+        return username;
+    }
+
+    @Override
+    public String database() {
+        return database;
+    }
+
+    @Override
+    public String password() {
+        return password;
+    }
+
     @Override
     public String version() {
-        return TSDBUtils.version(address);
+        return TSDBUtils.version(address, username, password);
     }
 
     @Override
     public String config() {
-        return TSDBUtils.config(address);
+        return TSDBUtils.config(address, username, password);
     }
 
     @Override
@@ -53,17 +74,22 @@ public class TSDBConnection implements Connection4TSDB {
 
     @Override
     public boolean put(DataPoint4TSDB dp) {
-        return TSDBUtils.put(address, dp);
+        return TSDBUtils.put(address, database, username, password, dp);
     }
 
     @Override
     public boolean put(List<DataPoint4TSDB> dps) {
-        return TSDBUtils.put(address, dps);
+        return TSDBUtils.put(address, database, username, password, dps);
     }
 
     @Override
     public boolean put(String dps) {
-        return TSDBUtils.put(address, dps);
+        return TSDBUtils.put(address, database, username, password, dps);
+    }
+
+    @Override
+    public boolean mput(String dps) {
+        return TSDBUtils.mput(address, database, username, password, dps);
     }
 
     @Override

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

@@ -10,8 +10,22 @@ package com.alibaba.datax.plugin.writer.tsdbwriter;
  */
 public class Key {
 
+    static final String SOURCE_DB_TYPE = "sourceDbType";
+    static final String MULTI_FIELD = "multiField";
+
+    // common
     static final String ENDPOINT = "endpoint";
+    static final String USERNAME = "username";
+    static final String PASSWORD = "password";
+    static final String IGNORE_WRITE_ERROR = "ignoreWriteError";
+    static final String DATABASE = "database";
+
+    // for tsdb
     static final String BATCH_SIZE = "batchSize";
     static final String MAX_RETRY_TIME = "maxRetryTime";
-    static final String IGNORE_WRITE_ERROR = "ignoreWriteError";
+
+    // for rdb
+    static final String COLUMN = "column";
+    static final String COLUMN_TYPE = "columnType";
+    static final String TABLE = "table";
 }

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/SourceDBType.java

@@ -0,0 +1,5 @@
+package com.alibaba.datax.plugin.writer.tsdbwriter;
+
+public enum SourceDBType {
+    TSDB, RDB
+}

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBConverter.java

@@ -0,0 +1,96 @@
+package com.alibaba.datax.plugin.writer.tsdbwriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.fastjson.JSON;
+import com.aliyun.hitsdb.client.value.request.MultiFieldPoint;
+import com.aliyun.hitsdb.client.value.request.Point;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class TSDBConverter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TSDBConverter.class);
+
+    private List<String> columnName;
+    private List<String> columnType;
+
+    TSDBConverter(List<String> columnName, List<String> columnType) {
+        this.columnName = columnName;
+        this.columnType = columnType;
+        LOG.info("columnName: {}, columnType: {}", JSON.toJSONString(columnName), JSON.toJSONString(columnType));
+    }
+
+    List<Point> transRecord2Point(List<Record> records) {
+        List<Point> dps = new ArrayList<Point>();
+        for (Record record : records) {
+            List<Point.MetricBuilder> metricBuilders = new ArrayList<Point.MetricBuilder>();
+            Map<String, String> tags = new HashMap<String, String>();
+            Long time = 0L;
+
+            for (int i = 0; i < columnType.size(); i++) {
+                String type = columnType.get(i);
+                String name = columnName.get(i);
+                Column column = record.getColumn(i);
+                if (TSDBModel.TSDB_TAG.equals(type)) {
+                    tags.put(name, column.asString());
+                } else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type)) {
+                    metricBuilders.add(new Point.MetricBuilder(name).value(column.asDouble()));
+                } else if (TSDBModel.TSDB_FIELD_STRING.equals(type)) {
+                    metricBuilders.add(new Point.MetricBuilder(name).value(column.asString()));
+                } else if (TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
+                    metricBuilders.add(new Point.MetricBuilder(name).value(column.asBoolean()));
+                } else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
+                    time = column.asLong();
+                } else if (TSDBModel.TSDB_METRIC_NUM.equals(type)) {
+                    // compatible with previous usage of TSDB_METRIC_NUM
+                    metricBuilders.add(new Point.MetricBuilder(name).value(column.asDouble()));
+                } else if (TSDBModel.TSDB_METRIC_STRING.equals(type)) {
+                    // compatible with previous usage of TSDB_METRIC_STRING
+                    metricBuilders.add(new Point.MetricBuilder(name).value(column.asString()));
+                }
+            }
+            for (Point.MetricBuilder metricBuilder : metricBuilders) {
+                dps.add(metricBuilder.tag(tags).timestamp(time).build(false));
+            }
+        }
+        return dps;
+    }
+
+    List<MultiFieldPoint> transRecord2MultiFieldPoint(List<Record> records, String tableName) {
+        List<MultiFieldPoint> dps = new ArrayList<MultiFieldPoint>();
+        for (Record record : records) {
+            MultiFieldPoint.MetricBuilder builder = MultiFieldPoint.metric(tableName);
+            for (int i = 0; i < columnType.size(); i++) {
+                String type = columnType.get(i);
+                String name = columnName.get(i);
+                Column column = record.getColumn(i);
+                if (TSDBModel.TSDB_TAG.equals(type)) {
+                    builder.tag(name, column.asString());
+                } else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type)) {
+                    builder.field(name, column.asDouble());
+                } else if (TSDBModel.TSDB_FIELD_STRING.equals(type)) {
+                    builder.field(name, column.asString());
+                } else if (TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
+                    builder.field(name, column.asBoolean());
+                } else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
+                    builder.timestamp(column.asLong());
+                } else if (TSDBModel.TSDB_METRIC_NUM.equals(type)) {
+                    // compatible with previous usage of TSDB_METRIC_NUM
+                    builder.field(name, column.asDouble());
+                } else if (TSDBModel.TSDB_METRIC_STRING.equals(type)) {
+                    // compatible with previous usage of TSDB_METRIC_STRING
+                    builder.field(name, column.asString());
+                }
+            }
+            MultiFieldPoint point = builder.build(false);
+            dps.add(point);
+        }
+        return dps;
+    }
+}

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBModel.java

@@ -0,0 +1,11 @@
+package com.alibaba.datax.plugin.writer.tsdbwriter;
+
+class TSDBModel {
+    static final String TSDB_METRIC_NUM = "metric_num";
+    static final String TSDB_METRIC_STRING = "metric_string";
+    static final String TSDB_TAG = "tag";
+    static final String TSDB_TIMESTAMP = "timestamp";
+    static final String TSDB_FIELD_DOUBLE = "field_double";
+    static final String TSDB_FIELD_STRING = "field_string";
+    static final String TSDB_FIELD_BOOL = "field_bool";
+}

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBWriter.java

@@ -7,12 +7,20 @@ import com.alibaba.datax.common.spi.Writer;
 import com.alibaba.datax.common.util.Configuration;
 import com.alibaba.datax.common.util.RetryUtil;
 import com.alibaba.datax.plugin.writer.conn.TSDBConnection;
+import com.aliyun.hitsdb.client.TSDB;
+import com.aliyun.hitsdb.client.TSDBClientFactory;
+import com.aliyun.hitsdb.client.TSDBConfig;
+import com.aliyun.hitsdb.client.value.request.MultiFieldPoint;
+import com.aliyun.hitsdb.client.value.request.Point;
+import com.aliyun.hitsdb.client.value.response.batch.IgnoreErrorsResult;
+import com.aliyun.hitsdb.client.value.response.batch.MultiFieldIgnoreErrorsResult;
+import com.aliyun.hitsdb.client.value.response.batch.SummaryResult;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.io.IOException;
+import java.util.*;
 import java.util.concurrent.Callable;
 
 /**
@@ -26,6 +34,9 @@ import java.util.concurrent.Callable;
 @SuppressWarnings("unused")
 public class TSDBWriter extends Writer {
 
+    private static SourceDBType DB_TYPE;
+    private static TSDB tsdb = null;
+
     public static class Job extends Writer.Job {
 
         private static final Logger LOG = LoggerFactory.getLogger(Job.class);
@@ -34,33 +45,99 @@ public class TSDBWriter extends Writer {
 
         @Override
         public void init() {
-            this.originalConfig = super.getPluginJobConf();
+            originalConfig = super.getPluginJobConf();
 
-            String address = this.originalConfig.getString(Key.ENDPOINT);
-            if (StringUtils.isBlank(address)) {
+            // check source db type
+            String sourceDbType = originalConfig.getString(Key.SOURCE_DB_TYPE);
+            if (StringUtils.isBlank(sourceDbType)) {
+                sourceDbType = SourceDBType.TSDB.name();
+                originalConfig.set(Key.SOURCE_DB_TYPE, sourceDbType);
+                LOG.info("The parameter [" + Key.SOURCE_DB_TYPE + "] will be default value: " + SourceDBType.TSDB);
+            }
+            try {
+                DB_TYPE = SourceDBType.valueOf(sourceDbType);
+            } catch (Exception e) {
                 throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
-                        "The parameter [" + Key.ENDPOINT + "] is not set.");
+                        "The parameter [" + Key.SOURCE_DB_TYPE +
+                                "] is invalid, which should be one of [" + Arrays.toString(SourceDBType.values()) + "].");
             }
 
-            Integer batchSize = this.originalConfig.getInt(Key.BATCH_SIZE);
-            if (batchSize == null || batchSize < 1) {
-                originalConfig.set(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
-                LOG.info("The parameter [" + Key.BATCH_SIZE +
-                        "] will be default value: " + Constant.DEFAULT_BATCH_SIZE);
-            }
+            // for tsdb
+            if (DB_TYPE == SourceDBType.TSDB) {
+                String address = originalConfig.getString(Key.ENDPOINT);
+                if (StringUtils.isBlank(address)) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
+                            "The parameter [" + Key.ENDPOINT + "] is not set.");
+                }
 
-            Integer retrySize = this.originalConfig.getInt(Key.MAX_RETRY_TIME);
-            if (retrySize == null || retrySize < 0) {
-                originalConfig.set(Key.MAX_RETRY_TIME, Constant.DEFAULT_TRY_SIZE);
-                LOG.info("The parameter [" + Key.MAX_RETRY_TIME +
-                        "] will be default value: " + Constant.DEFAULT_TRY_SIZE);
-            }
+                String username = originalConfig.getString(Key.USERNAME, null);
+                if (StringUtils.isBlank(username)) {
+                    LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
+                }
+                String password = originalConfig.getString(Key.PASSWORD, null);
+                if (StringUtils.isBlank(password)) {
+                    LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
+                }
+
+                Integer batchSize = originalConfig.getInt(Key.BATCH_SIZE);
+                if (batchSize == null || batchSize < 1) {
+                    originalConfig.set(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
+                    LOG.info("The parameter [" + Key.BATCH_SIZE +
+                            "] will be default value: " + Constant.DEFAULT_BATCH_SIZE);
+                }
+
+                Integer retrySize = originalConfig.getInt(Key.MAX_RETRY_TIME);
+                if (retrySize == null || retrySize < 0) {
+                    originalConfig.set(Key.MAX_RETRY_TIME, Constant.DEFAULT_TRY_SIZE);
+                    LOG.info("The parameter [" + Key.MAX_RETRY_TIME +
+                            "] will be default value: " + Constant.DEFAULT_TRY_SIZE);
+                }
+
+                Boolean ignoreWriteError = originalConfig.getBool(Key.IGNORE_WRITE_ERROR);
+                if (ignoreWriteError == null) {
+                    originalConfig.set(Key.IGNORE_WRITE_ERROR, Constant.DEFAULT_IGNORE_WRITE_ERROR);
+                    LOG.info("The parameter [" + Key.IGNORE_WRITE_ERROR +
+                            "] will be default value: " + Constant.DEFAULT_IGNORE_WRITE_ERROR);
+                }
+            } else if (DB_TYPE == SourceDBType.RDB) {
+                // for rdb
+                originalConfig.getNecessaryValue(Key.ENDPOINT, TSDBWriterErrorCode.REQUIRED_VALUE);
+                originalConfig.getNecessaryValue(Key.COLUMN_TYPE, TSDBWriterErrorCode.REQUIRED_VALUE);
+                originalConfig.getNecessaryValue(Key.COLUMN, TSDBWriterErrorCode.REQUIRED_VALUE);
+                String endpoint = originalConfig.getString(Key.ENDPOINT);
+                String[] split = endpoint.split(":");
+                if (split.length != 3) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
+                            "The parameter [" + Key.ENDPOINT + "] is invalid, which should be [http://IP:Port].");
+                }
+                String ip = split[1].substring(2);
+                int port = Integer.parseInt(split[2]);
+
+                String username = originalConfig.getString(Key.USERNAME, null);
+                if (StringUtils.isBlank(username)) {
+                    LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
+                }
+
+                String password = originalConfig.getString(Key.PASSWORD, null);
+                if (StringUtils.isBlank(password)) {
+                    LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
+                }
+
+                if (!StringUtils.isBlank(password) && !StringUtils.isBlank(username)) {
+                    tsdb = TSDBClientFactory.connect(TSDBConfig.address(ip, port).basicAuth(username, password).config());
+                } else {
+                    tsdb = TSDBClientFactory.connect(TSDBConfig.address(ip, port).config());
+                }
+
+                String database = originalConfig.getString(Key.DATABASE, null);
+                if (StringUtils.isBlank(database)) {
+                    LOG.info("The parameter [" + Key.DATABASE + "] is blank.");
+                } else {
+                    tsdb.useDatabase(database);
+                }
+
+                LOG.info("Tsdb config:" + originalConfig.toJSON());
 
-            Boolean ignoreWriteError = this.originalConfig.getBool(Key.IGNORE_WRITE_ERROR);
-            if (ignoreWriteError == null) {
-                originalConfig.set(Key.IGNORE_WRITE_ERROR, Constant.DEFAULT_IGNORE_WRITE_ERROR);
-                LOG.info("The parameter [" + Key.IGNORE_WRITE_ERROR +
-                        "] will be default value: " + Constant.DEFAULT_IGNORE_WRITE_ERROR);
             }
         }
 
@@ -72,7 +149,7 @@ public class TSDBWriter extends Writer {
         public List<Configuration> split(int mandatoryNumber) {
             ArrayList<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber);
             for (int i = 0; i < mandatoryNumber; i++) {
-                configurations.add(this.originalConfig.clone());
+                configurations.add(originalConfig.clone());
             }
             return configurations;
         }
@@ -83,6 +160,14 @@ public class TSDBWriter extends Writer {
 
         @Override
         public void destroy() {
+            if (DB_TYPE == SourceDBType.RDB) {
+                if (tsdb != null) {
+                    try {
+                        tsdb.close();
+                    } catch (IOException ignored) {
+                    }
+                }
+            }
         }
     }
 
@@ -91,18 +176,87 @@ public class TSDBWriter extends Writer {
         private static final Logger LOG = LoggerFactory.getLogger(Task.class);
 
         private TSDBConnection conn;
+        private boolean multiField;
         private int batchSize;
         private int retrySize;
         private boolean ignoreWriteError;
+        private String tableName;
+        private TSDBConverter tsdbConverter;
 
         @Override
         public void init() {
             Configuration writerSliceConfig = getPluginJobConf();
-            String address = writerSliceConfig.getString(Key.ENDPOINT);
-            this.conn = new TSDBConnection(address);
-            this.batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE);
-            this.retrySize = writerSliceConfig.getInt(Key.MAX_RETRY_TIME);
+
+            // single field | multi fields
+            this.multiField = writerSliceConfig.getBool(Key.MULTI_FIELD, false);
             this.ignoreWriteError = writerSliceConfig.getBool(Key.IGNORE_WRITE_ERROR);
+
+            // for tsdb
+            if (DB_TYPE == SourceDBType.TSDB) {
+                String address = writerSliceConfig.getString(Key.ENDPOINT);
+                String database = writerSliceConfig.getString(Key.DATABASE);
+                String username = writerSliceConfig.getString(Key.USERNAME);
+                String password = writerSliceConfig.getString(Key.PASSWORD);
+                this.conn = new TSDBConnection(address, database, username, password);
+                this.batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE);
+                this.retrySize = writerSliceConfig.getInt(Key.MAX_RETRY_TIME);
+
+            } else if (DB_TYPE == SourceDBType.RDB) {
+                // for rdb
+                int timeSize = 0;
+                int fieldSize = 0;
+                int tagSize = 0;
+                batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE, 100);
+                List<String> columnName = writerSliceConfig.getList(Key.COLUMN, String.class);
+                List<String> columnType = writerSliceConfig.getList(Key.COLUMN_TYPE, String.class);
+                Set<String> typeSet = new HashSet<String>(columnType);
+                if (columnName.size() != columnType.size()) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
+                            "The parameter [" + Key.COLUMN_TYPE + "] should has same length with [" + Key.COLUMN + "].");
+                }
+
+                for (String type : columnType) {
+                    if (TSDBModel.TSDB_TAG.equals(type)) {
+                        tagSize ++;
+                    } else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type) || TSDBModel.TSDB_FIELD_STRING.equals(type)
+                            || TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
+                        fieldSize++;
+                    } else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
+                        timeSize++;
+                    }
+                }
+
+                if (fieldSize == 0) {
+                    // compatible with previous usage of TSDB_METRIC_NUM and TSDB_METRIC_STRING
+                    if (!typeSet.contains(TSDBModel.TSDB_METRIC_NUM) && !typeSet.contains(TSDBModel.TSDB_METRIC_STRING)) {
+                        throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
+                                "The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set at least one of "
+                                        + TSDBModel.TSDB_FIELD_DOUBLE + ", " + TSDBModel.TSDB_FIELD_STRING + " or " + TSDBModel.TSDB_FIELD_BOOL + ".");
+                    }
+                }
+
+                if (tagSize == 0) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
+                            "The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set " + TSDBModel.TSDB_TAG + ". ");
+                }
+
+                if (timeSize != 1) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
+                            "The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set one and only one "
+                                    + TSDBModel.TSDB_TIMESTAMP + ".");
+                }
+
+                if (multiField) {
+                    // check source db type
+                    tableName = writerSliceConfig.getString(Key.TABLE);
+                    if (StringUtils.isBlank(tableName)) {
+                        throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
+                                "The parameter [" + Key.TABLE + "] h must set when use multi field input.");
+                    }
+                }
+                tsdbConverter = new TSDBConverter(columnName, columnType);
+
+            }
         }
 
         @Override
@@ -111,30 +265,52 @@ public class TSDBWriter extends Writer {
 
         @Override
         public void startWrite(RecordReceiver recordReceiver) {
-            try {
-                Record lastRecord = null;
-                Record record;
-                int count = 0;
-                StringBuilder dps = new StringBuilder();
-                while ((record = recordReceiver.getFromReader()) != null) {
-                    final int recordLength = record.getColumnNumber();
-                    for (int i = 0; i < recordLength; i++) {
-                        dps.append(record.getColumn(i).asString());
-                        dps.append(",");
-                        count++;
-                        if (count == batchSize) {
-                            count = 0;
-                            batchPut(record, "[" + dps.substring(0, dps.length() - 1) + "]");
-                            dps = new StringBuilder();
+            // for tsdb
+            if (DB_TYPE == SourceDBType.TSDB) {
+                try {
+                    Record lastRecord = null;
+                    Record record;
+                    int count = 0;
+                    StringBuilder dps = new StringBuilder();
+                    while ((record = recordReceiver.getFromReader()) != null) {
+                        final int recordLength = record.getColumnNumber();
+                        for (int i = 0; i < recordLength; i++) {
+                            dps.append(record.getColumn(i).asString());
+                            dps.append(",");
+                            count++;
+                            if (count == batchSize) {
+                                count = 0;
+                                batchPut(record, "[" + dps.substring(0, dps.length() - 1) + "]");
+                                dps = new StringBuilder();
+                            }
                         }
+                        lastRecord = record;
                     }
-                    lastRecord = record;
+                    if (StringUtils.isNotBlank(dps.toString())) {
+                        batchPut(lastRecord, "[" + dps.substring(0, dps.length() - 1) + "]");
+                    }
+                } catch (Exception e) {
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, e);
                 }
-                if (StringUtils.isNotBlank(dps.toString())) {
-                    batchPut(lastRecord, "[" + dps.substring(0, dps.length() - 1) + "]");
+            } else if (DB_TYPE == SourceDBType.RDB) {
+                // for rdb
+                List<Record> writerBuffer = new ArrayList<Record>(this.batchSize);
+                Record record;
+                long total = 0;
+                while ((record = recordReceiver.getFromReader()) != null) {
+                    writerBuffer.add(record);
+                    if (writerBuffer.size() >= this.batchSize) {
+                        total += doBatchInsert(writerBuffer);
+                        writerBuffer.clear();
+                    }
                 }
-            } catch (Exception e) {
-                throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, e);
+                if (!writerBuffer.isEmpty()) {
+                    total += doBatchInsert(writerBuffer);
+                    writerBuffer.clear();
+                }
+                getTaskPluginCollector().collectMessage("write size", total + "");
+                LOG.info("Task finished, write size: {}", total);
+
             }
         }
 
@@ -143,12 +319,13 @@ public class TSDBWriter extends Writer {
                 RetryUtil.executeWithRetry(new Callable<Integer>() {
                     @Override
                     public Integer call() {
-                        if (!conn.put(dps)) {
-                            getTaskPluginCollector().collectDirtyRecord(record, "Put data points failed!");
-                            throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION,
-                                    "Put data points failed!");
+                        final boolean success = multiField ? conn.mput(dps) : conn.put(dps);
+                        if (success) {
+                            return 0;
                         }
-                        return 0;
+                        getTaskPluginCollector().collectDirtyRecord(record, "Put data points failed!");
+                        throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION,
+                                "Put data points failed!");
                     }
                 }, retrySize, 60000L, true);
             } catch (Exception e) {
@@ -160,6 +337,47 @@ public class TSDBWriter extends Writer {
             }
         }
 
+        private long doBatchInsert(final List<Record> writerBuffer) {
+            int size;
+            if (ignoreWriteError) {
+                if (multiField) {
+                    List<MultiFieldPoint> points = tsdbConverter.transRecord2MultiFieldPoint(writerBuffer, tableName);
+                    size = points.size();
+                    MultiFieldIgnoreErrorsResult ignoreErrorsResult = tsdb.multiFieldPutSync(points, MultiFieldIgnoreErrorsResult.class);
+                    if (ignoreErrorsResult == null) {
+                        LOG.error("Unexpected inner error for insert");
+                    } else if (ignoreErrorsResult.getFailed() > 0) {
+                        LOG.error("write TSDB failed num:" + ignoreErrorsResult.getFailed());
+                    }
+                } else {
+                    List<Point> points = tsdbConverter.transRecord2Point(writerBuffer);
+                    size = points.size();
+                    IgnoreErrorsResult ignoreErrorsResult = tsdb.putSync(points, IgnoreErrorsResult.class);
+                    if (ignoreErrorsResult == null) {
+                        LOG.error("Unexpected inner error for insert");
+                    } else if (ignoreErrorsResult.getFailed() > 0) {
+                        LOG.error("write TSDB failed num:" + ignoreErrorsResult.getFailed());
+                    }
+                }
+            } else {
+                SummaryResult summaryResult;
+                if (multiField) {
+                    List<MultiFieldPoint> points = tsdbConverter.transRecord2MultiFieldPoint(writerBuffer, tableName);
+                    size = points.size();
+                    summaryResult = tsdb.multiFieldPutSync(points, SummaryResult.class);
+                } else {
+                    List<Point> points = tsdbConverter.transRecord2Point(writerBuffer);
+                    size = points.size();
+                    summaryResult = tsdb.putSync(points, SummaryResult.class);
+                }
+                if (summaryResult.getFailed() > 0) {
+                    LOG.error("write TSDB failed num:" + summaryResult.getFailed());
+                    throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, "Write TSDB failed", new Exception());
+                }
+            }
+            return size;
+        }
+
         @Override
         public void post() {
         }

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBWriterErrorCode.java

@@ -13,6 +13,7 @@ import com.alibaba.datax.common.spi.ErrorCode;
 public enum TSDBWriterErrorCode implements ErrorCode {
 
     REQUIRED_VALUE("TSDBWriter-00", "Missing the necessary value"),
+    ILLEGAL_VALUE("TSDBWriter-01", "Illegal value"),
     RUNTIME_EXCEPTION("TSDBWriter-01", "Runtime exception"),
     RETRY_WRITER_EXCEPTION("TSDBWriter-02", "After repeated attempts, the write still fails");
 

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/util/HttpUtils.java

@@ -1,11 +1,14 @@
 package com.alibaba.datax.plugin.writer.util;
 
 import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.client.fluent.Content;
 import org.apache.http.client.fluent.Request;
 import org.apache.http.entity.ContentType;
 
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -19,43 +22,44 @@ import java.util.concurrent.TimeUnit;
  */
 public final class HttpUtils {
 
-    public final static Charset UTF_8 = Charset.forName("UTF-8");
     public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
     public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
 
+    private static final String CREDENTIALS_FORMAT = "%s:%s";
+    private static final String BASIC_AUTHENTICATION_FORMAT = "Basic %s";
+
     private HttpUtils() {
     }
 
-    public static String get(String url) throws Exception {
-        Content content = Request.Get(url)
+    public static String get(String url, String username, String password) throws Exception {
+        final Request request = Request.Get(url)
                 .connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL)
-                .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL)
+                .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+        addAuth(request, username, password);
+        Content content = request
                 .execute()
                 .returnContent();
         if (content == null) {
             return null;
         }
-        return content.asString(UTF_8);
+        return content.asString(StandardCharsets.UTF_8);
     }
 
-    public static String post(String url, Map<String, Object> params) throws Exception {
-        return post(url, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+    public static String post(String url, String username, String password, Map<String, Object> params) throws Exception {
+        return post(url, username, password, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
     }
 
-    public static String post(String url, String params) throws Exception {
-        return post(url, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
+    public static String post(String url, String username, String password, String params) throws Exception {
+        return post(url, username, password, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
     }
 
-    public static String post(String url, Map<String, Object> params,
+    public static String post(String url, String username, String password, String params,
                               int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
-        return post(url, JSON.toJSONString(params), connectTimeoutInMill, socketTimeoutInMill);
-    }
-
-    public static String post(String url, String params,
-                              int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
-        Content content = Request.Post(url)
+        Request request = Request.Post(url)
                 .connectTimeout(connectTimeoutInMill)
-                .socketTimeout(socketTimeoutInMill)
+                .socketTimeout(socketTimeoutInMill);
+        addAuth(request, username, password);
+        Content content = request
                 .addHeader("Content-Type", "application/json")
                 .bodyString(params, ContentType.APPLICATION_JSON)
                 .execute()
@@ -63,6 +67,22 @@ public final class HttpUtils {
         if (content == null) {
             return null;
         }
-        return content.asString(UTF_8);
+        return content.asString(StandardCharsets.UTF_8);
+    }
+
+    private static void addAuth(Request request, String username, String password) {
+        String authorization = generateHttpAuthorization(username, password);
+        if (authorization != null) {
+            request.setHeader("Authorization", authorization);
+        }
+    }
+
+    private static String generateHttpAuthorization(String username, String password) {
+        if (StringUtils.isBlank(username) || StringUtils.isBlank(password)) {
+            return null;
+        }
+        String credentials = String.format(CREDENTIALS_FORMAT, username, password);
+        credentials = Base64.getEncoder().encodeToString(credentials.getBytes());
+        return String.format(BASIC_AUTHENTICATION_FORMAT, credentials);
     }
 }

tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/util/TSDBUtils.java

@@ -2,6 +2,7 @@ package com.alibaba.datax.plugin.writer.util;
 
 import com.alibaba.datax.plugin.writer.conn.DataPoint4TSDB;
 import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -22,45 +23,56 @@ public final class TSDBUtils {
     private TSDBUtils() {
     }
 
-    public static String version(String address) {
+    public static String version(String address, String username, String password) {
         String url = String.format("%s/api/version", address);
         String rsp;
         try {
-            rsp = HttpUtils.get(url);
+            rsp = HttpUtils.get(url, username, password);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
         return rsp;
     }
 
-    public static String config(String address) {
+    public static String config(String address, String username, String password) {
         String url = String.format("%s/api/config", address);
         String rsp;
         try {
-            rsp = HttpUtils.get(url);
+            rsp = HttpUtils.get(url, username, password);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
         return rsp;
     }
 
-    public static boolean put(String address, List<DataPoint4TSDB> dps) {
-        return put(address, JSON.toJSON(dps));
+    public static boolean put(String address, String database, String username, String password, List<DataPoint4TSDB> dps) {
+        return put(address, database, username, password, JSON.toJSON(dps));
     }
 
-    public static boolean put(String address, DataPoint4TSDB dp) {
-        return put(address, JSON.toJSON(dp));
+    public static boolean put(String address, String database, String username, String password, DataPoint4TSDB dp) {
+        return put(address, database, username, password, JSON.toJSON(dp));
     }
 
-    private static boolean put(String address, Object o) {
-        return put(address, o.toString());
+    private static boolean put(String address, String database, String username, String password, Object o) {
+        return put(address, database, username, password, o.toString());
     }
 
-    public static boolean put(String address, String s) {
-        String url = String.format("%s/api/put", address);
+    public static boolean put(String address, String database, String username, String password, String s) {
+        return put(address, database, username, password, s, false);
+    }
+
+    public static boolean mput(String address, String database, String username, String password, String s) {
+        return put(address, database, username, password, s, true);
+    }
+
+    public static boolean put(String address, String database, String username, String password, String s, boolean multiField) {
+        String url = address + (multiField ? "/api/mput" : "/api/put");
+        if (!StringUtils.isBlank(database)) {
+            url = url.concat("?db=" + database);
+        }
         String rsp;
         try {
-            rsp = HttpUtils.post(url, s);
+            rsp = HttpUtils.post(url, username, password, s);
             // If successful, the returned content should be null.
             assert rsp == null;
         } catch (Exception e) {