[feat-855][doris] doris stream load change to json format.

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/options/DorisConf.java

@@ -33,10 +33,6 @@
  */
 public class DorisConf extends ChunJunCommonConf {
 
- private String fieldDelimiter;
-
- private String lineDelimiter;
-
  private String database;
 
  private String table;
@@ -58,22 +54,6 @@ public class DorisConf extends ChunJunCommonConf {
 
  private Properties loadProperties;
 
- public String getFieldDelimiter() {
- return fieldDelimiter;
- }
-
- public void setFieldDelimiter(String fieldDelimiter) {
- this.fieldDelimiter = fieldDelimiter;
- }
-
- public String getLineDelimiter() {
- return lineDelimiter;
- }
-
- public void setLineDelimiter(String lineDelimiter) {
- this.lineDelimiter = lineDelimiter;
- }
-
  public String getDatabase() {
  return database;
  }

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/options/DorisConfBuilder.java

@@ -48,16 +48,6 @@ public DorisConfBuilder setFeNodes(List<String> feNodes) {
  return this;
  }
 
- public DorisConfBuilder setFieldDelimiter(String fieldDelimiter) {
- this.dorisConf.setFieldDelimiter(fieldDelimiter);
- return this;
- }
-
- public DorisConfBuilder setLineDelimiter(String lineDelimiter) {
- this.dorisConf.setLineDelimiter(lineDelimiter);
- return this;
- }
-
  public DorisConfBuilder setUsername(String username) {
  this.dorisConf.setUsername(username);
  return this;

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/rest/Carrier.java

@@ -4,10 +4,13 @@
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.StringJoiner;
+import java.util.stream.IntStream;
 
 /**
  * Company：www.dtstack.com.
@@ -17,18 +20,16 @@
  */
 public class Carrier implements Serializable {
  private static final long serialVersionUID = 1L;
- private final StringJoiner insertContent;
+ private final List<Map<String, Object>> insertContent;
  private final StringJoiner deleteContent;
- private final String fieldDelimiter;
  private int batch = 0;
  private String database;
  private String table;
  private List<String> columns;
  private final Set<Integer> rowDataIndexes = new HashSet<>();
 
- public Carrier(String fieldDelimiter, String lineDelimiter) {
- this.fieldDelimiter = fieldDelimiter;
- insertContent = new StringJoiner(lineDelimiter);
+ public Carrier() {
+ insertContent = new ArrayList<>();
  deleteContent = new StringJoiner(" OR ");
  }
 
@@ -48,8 +49,8 @@ public void setTable(String table) {
  this.table = table;
  }
 
- public String getInsertContent() {
- return insertContent.toString();
+ public List<Map<String, Object>> getInsertContent() {
+ return insertContent;
  }
 
  public String getDeleteContent() {
@@ -78,14 +79,20 @@ public void addInsertContent(List<String> insertV) {
  // It is certain that in this case, the size
  // of insertV is twice the size of column
  List<String> forward = insertV.subList(0, columns.size());
- String forwardV = StringUtils.join(forward, fieldDelimiter);
+ final Map<String, Object> forwardV = new HashMap<>(columns.size());
+ IntStream.range(0, columns.size())
+ .forEach(i -> forwardV.put(columns.get(i), forward.get(i)));
  insertContent.add(forwardV);
  List<String> behind = insertV.subList(columns.size(), insertV.size());
- String behindV = StringUtils.join(behind, fieldDelimiter);
+ final Map<String, Object> behindV = new HashMap<>(columns.size());
+ IntStream.range(0, columns.size())
+ .forEach(i -> behindV.put(columns.get(i), behind.get(i)));
  insertContent.add(behindV);
  } else {
- String s = StringUtils.join(insertV, fieldDelimiter);
- insertContent.add(s);
+ final Map<String, Object> values = new HashMap<>(columns.size());
+ IntStream.range(0, columns.size())
+ .forEach(i -> values.put(columns.get(i), insertV.get(i)));
+ insertContent.add(values);
  }
  }
  }
@@ -110,7 +117,9 @@ private String buildMergeOnConditions(List<String> columns, List<String> values)
  List<String> deleteOnStr = new ArrayList<>();
  for (int i = 0, size = columns.size(); i < size; i++) {
  String s =
- columns.get(i)
+ "`"
+ + columns.get(i)
+ + "`"
  + "<=>"
  + "'"
  + ((values.get(i)) == null ? "" : values.get(i))

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/rest/DorisLoadClient.java

@@ -61,16 +61,13 @@ public class DorisLoadClient implements Serializable {
  .collect(Collectors.toCollection(HashSet::new));
 
  private static final String LOAD_URL_PATTERN = "http://%s/api/%s/%s/_stream_load?";
- private static final String NULL_VALUE = "\\N";
  private static final String KEY_SCHEMA = "schema";
  private static final String KEY_TABLE = "table";
  private static final String KEY_POINT = ".";
  public static final String KEY_BEFORE = "before_";
  public static final String KEY_AFTER = "after_";
 
  private final DorisStreamLoad dorisStreamLoad;
- private final String fieldDelimiter;
- private final String lineDelimiter;
  private final boolean nameMapped;
  private String hostPort;
  private final DorisConf conf;
@@ -80,8 +77,6 @@ public DorisLoadClient(DorisStreamLoad dorisStreamLoad, DorisConf conf, String h
  this.hostPort = hostPort;
  this.conf = conf;
  this.nameMapped = conf.isNameMapped();
- this.fieldDelimiter = conf.getFieldDelimiter();
- this.lineDelimiter = conf.getLineDelimiter();
  }
 
  public void setHostPort(String hostPort) {
@@ -271,15 +266,15 @@ private void wrapValuesFromRowData(
  if (column.equalsIgnoreCase(trueCol)) {
  insertV.add(convert(value, i));
  deleteV.add(convert(value, i));
- break;
+ continue;
  }
  }
  // case 2, need to insert.
  if (headers[i].startsWith(KEY_AFTER)) {
  String trueCol = headers[i].substring(6);
  if (column.equalsIgnoreCase(trueCol)) {
  insertV.add(convert(value, i));
- break;
+ continue;
  }
  }
  // case 3. column name is obvious.
@@ -288,7 +283,6 @@ private void wrapValuesFromRowData(
  if (delete) {
  deleteV.add(convert(value, i));
  }
- break;
  }
  }
  }
@@ -301,7 +295,7 @@ private Carrier initCarrier(
  List<String> deleteV,
  String schema,
  String table) {
- Carrier carrier = new Carrier(fieldDelimiter, lineDelimiter);
+ Carrier carrier = new Carrier();
  carrier.setColumns(columns);
  carrier.setDatabase(schema);
  carrier.setTable(table);
@@ -347,6 +341,6 @@ private List<String> getColumnName(List<FieldConf> fields) {
 
  private String convert(@Nonnull ColumnRowData rowData, int index) {
  Object value = rowData.getField(index);
- return (value == null || "".equals(value.toString())) ? NULL_VALUE : value.toString();
+ return (value == null || "".equals(value.toString())) ? null : value.toString();
  }
 }

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/rest/DorisStreamLoad.java

@@ -46,10 +46,12 @@
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Date;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 /**
  * @author tiezhu@dtstack.com
@@ -63,8 +65,6 @@ public class DorisStreamLoad implements Serializable {
  new ArrayList<>(Arrays.asList("Success", "Publish Timeout"));
  private final String authEncoding;
  private final Properties streamLoadProp;
- private final String fieldDelimiter;
- private final String lineDelimiter;
 
  public DorisStreamLoad(DorisConf options) {
  this.authEncoding =
@@ -73,8 +73,6 @@ public DorisStreamLoad(DorisConf options) {
  String.format("%s:%s", options.getUsername(), options.getPassword())
  .getBytes(StandardCharsets.UTF_8));
  this.streamLoadProp = options.getLoadProperties();
- this.fieldDelimiter = options.getFieldDelimiter();
- this.lineDelimiter = options.getLineDelimiter();
  }
 
  /**
@@ -94,23 +92,34 @@ private HttpPut generatePut(
  httpPut.setHeader("Expect", "100-continue");
  httpPut.setHeader("Content-Type", "text/plain; charset=UTF-8");
  httpPut.setHeader("label", label);
- httpPut.setHeader("columns", StringUtils.join(columnNames, ","));
+ httpPut.setHeader("format", "json");
+ // if body is list type ,strip_outer_array should be true
+ httpPut.setHeader("strip_outer_array", "true");
+ List<String> columns =
+ columnNames.stream()
+ .map(this::quoteColumn)
+ .collect(Collectors.toCollection(LinkedList::new));
+ httpPut.setHeader("columns", StringUtils.join(columns, ","));
  if (StringUtils.isNotBlank(mergeConditions)) {
  httpPut.setHeader("merge_type", "MERGE");
  httpPut.setHeader("delete", mergeConditions);
  } else {
  httpPut.setHeader("merge_type", "APPEND");
  }
- httpPut.setHeader("column_separator", fieldDelimiter);
- if (!"\n".equals(lineDelimiter)) {
- httpPut.setHeader("line_delimiter", lineDelimiter);
- }
+ // httpPut.setHeader("column_separator", fieldDelimiter);
+ // if (!"\n".equals(lineDelimiter)) {
+ // httpPut.setHeader("line_delimiter", lineDelimiter);
+ // }
  for (Map.Entry<Object, Object> entry : streamLoadProp.entrySet()) {
  httpPut.setHeader(String.valueOf(entry.getKey()), String.valueOf(entry.getValue()));
  }
  return httpPut;
  }
 
+ private String quoteColumn(String column) {
+ return "`" + column + "`";
+ }
+
  public static class LoadResponse {
  public int status;
  public String respContent;
@@ -138,9 +147,9 @@ public String toString() {
  */
  public void load(Carrier carrier, String loadUrlStr) throws IOException {
  List<String> columnNames = carrier.getColumns();
- String value = carrier.getInsertContent();
+ String json = OM.writeValueAsString(carrier.getInsertContent());
  String mergeConditions = carrier.getDeleteContent();
- LoadResponse loadResponse = loadBatch(columnNames, value, mergeConditions, loadUrlStr);
+ LoadResponse loadResponse = loadBatch(columnNames, json, mergeConditions, loadUrlStr);
  LOG.debug("StreamLoad Response:{}", loadResponse);
  if (loadResponse.status != 200) {
  throw new ConnectException("stream load error, detail : " + loadResponse);
@@ -187,7 +196,7 @@ private String generateLabel() {
  String formatDate = sdf.format(new Date());
  label =
  String.format(
- "chunjun_connector_%s_%s",
+ "flinkx_connector_%s_%s",
  formatDate, UUID.randomUUID().toString().replaceAll("-", ""));
  }
  return label;

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/sink/DorisSinkFactory.java

@@ -49,11 +49,7 @@
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.DORIS_WRITE_MODE_DEFAULT;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.EXEC_MEM_LIMIT_KEY;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.FE_NODES_KEY;
-import static com.dtstack.chunjun.connector.doris.options.DorisKeys.FIELD_DELIMITER;
-import static com.dtstack.chunjun.connector.doris.options.DorisKeys.FIELD_DELIMITER_KEY;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.FLUSH_INTERNAL_MS_KEY;
-import static com.dtstack.chunjun.connector.doris.options.DorisKeys.LINE_DELIMITER;
-import static com.dtstack.chunjun.connector.doris.options.DorisKeys.LINE_DELIMITER_KEY;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.LOAD_OPTIONS_KEY;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.LOAD_PROPERTIES_KEY;
 import static com.dtstack.chunjun.connector.doris.options.DorisKeys.PASSWORD_KEY;
@@ -136,10 +132,6 @@ public DorisSinkFactory(SyncConf syncConf) {
  .setDatabase(parameter.getStringVal(DATABASE_KEY))
  .setTable(parameter.getStringVal(TABLE_KEY))
  .setFeNodes((List<String>) parameter.getVal(FE_NODES_KEY))
- .setFieldDelimiter(
- parameter.getStringVal(FIELD_DELIMITER_KEY, FIELD_DELIMITER))
- .setLineDelimiter(
- parameter.getStringVal(LINE_DELIMITER_KEY, LINE_DELIMITER))
  .setLoadOptions(loadConf)
  .setLoadProperties(
  parameter.getProperties(LOAD_PROPERTIES_KEY, new Properties()))

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/table/DorisDynamicTableFactory.java

@@ -66,8 +66,6 @@ public class DorisDynamicTableFactory implements DynamicTableSinkFactory {
  DorisOptions.EXEC_MEM_LIMIT,
  DorisOptions.DESERIALIZE_QUEUE_SIZE,
  DorisOptions.DESERIALIZE_ARROW_ASYNC,
- DorisOptions.FIELD_DELIMITER,
- DorisOptions.LINE_DELIMITER,
  DorisOptions.MAX_RETRIES,
  DorisOptions.WRITE_MODE,
  DorisOptions.BATCH_SIZE)
@@ -112,8 +110,6 @@ private static DorisConf getConfByOptions(ReadableConfig config) {
 
  LoadConf loadConf = getLoadConf(config);
  dorisConf.setLoadConf(loadConf);
- dorisConf.setFieldDelimiter(config.get(DorisOptions.FIELD_DELIMITER));
- dorisConf.setLineDelimiter(config.get(DorisOptions.LINE_DELIMITER));
  dorisConf.setLoadProperties(new Properties());
  dorisConf.setMaxRetries(config.get(DorisOptions.MAX_RETRIES));
  dorisConf.setWriteMode(config.get(DorisOptions.WRITE_MODE));