doris同步优化

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/DorisFactoryProvider.java

@@ -9,16 +9,14 @@
 /////////////////////////////////////////////////////////////
 package org.dromara.dbswitch.product.doris;
 
-import org.dromara.dbswitch.core.annotation.Product;
+import javax.sql.DataSource;
 import org.dromara.dbswitch.common.type.ProductTypeEnum;
+import org.dromara.dbswitch.core.annotation.Product;
 import org.dromara.dbswitch.core.features.ProductFeatures;
 import org.dromara.dbswitch.core.provider.AbstractFactoryProvider;
 import org.dromara.dbswitch.core.provider.meta.MetadataProvider;
-import org.dromara.dbswitch.core.provider.sync.AutoCastTableDataSynchronizeProvider;
 import org.dromara.dbswitch.core.provider.sync.TableDataSynchronizeProvider;
-import org.dromara.dbswitch.core.provider.write.AutoCastTableDataWriteProvider;
 import org.dromara.dbswitch.core.provider.write.TableDataWriteProvider;
-import javax.sql.DataSource;
 
 @Product(ProductTypeEnum.DORIS)
 public class DorisFactoryProvider extends AbstractFactoryProvider {
@@ -39,12 +37,12 @@ public class DorisFactoryProvider extends AbstractFactoryProvider {
 
   @Override
   public TableDataWriteProvider createTableDataWriteProvider(boolean useInsert) {
-    return new AutoCastTableDataWriteProvider(this);
+    return new DorisTableDataWriteProvider(this);
   }
 
   @Override
   public TableDataSynchronizeProvider createTableDataSynchronizeProvider() {
-    return new AutoCastTableDataSynchronizeProvider(this);
+    return new DorisTableDataSynchronizer(this);
   }
 
 }

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/DorisMetadataQueryProvider.java

@@ -354,7 +354,8 @@ public class DorisMetadataQueryProvider extends AbstractMetadataProvider {
     if (CollectionUtils.isNotEmpty(primaryKeys)) {
       String primaryKeyAsString = getPrimaryKeyAsString(primaryKeys);
       // 自动分桶（BUCKETS AUTO）功能要求 Apache Doris 1.2.2 及以上版本
-      builder.append(" DISTRIBUTED BY HASH(").append(primaryKeyAsString).append(") BUCKETS AUTO ");
+      builder.append(" DISTRIBUTED BY HASH(").append(primaryKeyAsString).append(") ");
+      builder.append(" PROPERTIES ('replication_num' = '1') ");
     }
   }
 

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/DorisTableDataSynchronizer.java

@@ -0,0 +1,62 @@
+// Copyright tang.  All rights reserved.
+// https://gitee.com/inrgihc/dbswitch
+//
+// Use of this source code is governed by a BSD-style license
+//
+// Author: tang (inrgihc@126.com)
+// Date : 2020/1/2
+// Location: beijing , china
+/////////////////////////////////////////////////////////////
+package org.dromara.dbswitch.product.doris;
+
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.dbswitch.common.entity.CloseableDataSource;
+import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
+import org.dromara.dbswitch.core.provider.sync.AutoCastTableDataSynchronizeProvider;
+
+@Slf4j
+public class DorisTableDataSynchronizer extends AutoCastTableDataSynchronizeProvider {
+
+  private List<String> fieldNames;
+  private final CloseableDataSource dataSource;
+
+  private final DorisUtils dorisUtils = new DorisUtils();
+
+  public DorisTableDataSynchronizer(ProductFactoryProvider factoryProvider) {
+    super(factoryProvider);
+    dataSource = (CloseableDataSource) factoryProvider.getDataSource();
+  }
+
+  @Override
+  public void prepare(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
+    this.fieldNames = fieldNames;
+    super.prepare(schemaName, tableName, fieldNames, pks);
+    try {
+      dorisUtils.init(schemaName, tableName, dataSource);
+    } catch (Exception e) {
+      log.warn("Failed to init by DorisUtils#init(),information:: {}", e.getMessage());
+    }
+  }
+
+  @Override
+  public long executeInsert(List<Object[]> recordValues) {
+    try {
+      return dorisUtils.addOrUpdateData(fieldNames, recordValues);
+    } catch (Exception e) {
+      log.warn("Failed to addOrUpdateData by DorisUtils#addOrUpdateData(),information:: {}", e.getMessage());
+      return super.executeInsert(recordValues);
+    }
+  }
+
+  @Override
+  public long executeUpdate(List<Object[]> recordValues) {
+    try {
+      return dorisUtils.addOrUpdateData(fieldNames, recordValues);
+    } catch (Exception e) {
+      log.warn("Failed to addOrUpdateData by DorisUtils#addOrUpdateData(),information:: {}", e.getMessage());
+      return super.executeUpdate(recordValues);
+    }
+  }
+
+}

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/DorisTableDataWriteProvider.java

@@ -0,0 +1,49 @@
+// Copyright tang.  All rights reserved.
+// https://gitee.com/inrgihc/dbswitch
+//
+// Use of this source code is governed by a BSD-style license
+//
+// Author: tang (inrgihc@126.com)
+// Date : 2020/1/2
+// Location: beijing , china
+/////////////////////////////////////////////////////////////
+package org.dromara.dbswitch.product.doris;
+
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.dbswitch.common.entity.CloseableDataSource;
+import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
+import org.dromara.dbswitch.core.provider.write.AutoCastTableDataWriteProvider;
+
+@Slf4j
+public class DorisTableDataWriteProvider extends AutoCastTableDataWriteProvider {
+
+  private final CloseableDataSource dataSource;
+  private final DorisUtils dorisUtils = new DorisUtils();
+
+  public DorisTableDataWriteProvider(ProductFactoryProvider factoryProvider) {
+    super(factoryProvider);
+    dataSource = (CloseableDataSource) factoryProvider.getDataSource();
+  }
+
+  @Override
+  public void prepareWrite(String schemaName, String tableName, List<String> fieldNames) {
+    super.prepareWrite(schemaName, tableName, fieldNames);
+    try {
+      dorisUtils.init(schemaName, tableName, dataSource);
+    } catch (Exception e) {
+      log.warn("Failed to init by DorisUtils#init(),information: {}", e.getMessage());
+    }
+  }
+
+  @Override
+  public long write(List<String> fieldNames, List<Object[]> recordValues) {
+    try {
+      return dorisUtils.addOrUpdateData(fieldNames, recordValues);
+    } catch (Exception e) {
+      log.warn("Failed to insertOrUpdate data by DorisUtils#addOrUpdateData(),information: {}", e.getMessage());
+      return super.write(fieldNames, recordValues);
+    }
+  }
+
+}

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/DorisUtils.java

@@ -0,0 +1,162 @@
+// Copyright tang.  All rights reserved.
+// https://gitee.com/inrgihc/dbswitch
+//
+// Use of this source code is governed by a BSD-style license
+//
+// Author: tang (inrgihc@126.com)
+// Date : 2020/1/2
+// Location: beijing , china
+/////////////////////////////////////////////////////////////
+package org.dromara.dbswitch.product.doris;
+
+import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.util.ReUtil;
+import cn.hutool.db.Db;
+import cn.hutool.db.Entity;
+import cn.hutool.json.JSONArray;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+import org.dromara.dbswitch.common.entity.CloseableDataSource;
+
+@Slf4j
+public final class DorisUtils {
+
+  private String dbName;
+  private String tbName;
+  private String host;
+  private String username;
+  private String password;
+  private CloseableDataSource dataSource;
+  private String httpPort;
+
+  public void init(String schemaName, String tableName, DataSource dataSource) {
+    this.dataSource = (CloseableDataSource) dataSource;
+    this.host = ReUtil.extractMulti("jdbc:mysql://(.*):[0-9]{2,8}/", this.dataSource.getJdbcUrl(), "$1");
+    this.username = this.dataSource.getUserName();
+    this.password = this.dataSource.getPassword();
+    this.tbName = tableName;
+    this.dbName = schemaName;
+    this.getHttpPort(dataSource);
+  }
+
+  public void getHttpPort(DataSource dataSource) {
+    Db use = Db.use(dataSource);
+    try {
+      List<Entity> frontends = use.query("SHOW FRONTENDS");
+      if (CollectionUtils.isNotEmpty(frontends)) {
+        List<FrontendEntity> frontendEntities = BeanUtil.copyToList(frontends, FrontendEntity.class);
+        if (frontendEntities.size() > 1) {
+          List<FrontendEntity> leader = frontendEntities.stream().filter(i -> i.getRole().equals("LEADER"))
+              .collect(Collectors.toList());
+          FrontendEntity frontendEntity = leader.get(0);
+          this.httpPort = frontendEntity.getHttpport();
+        } else {
+          this.httpPort = frontendEntities.get(0).getHttpport();
+        }
+      }
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  public long addOrUpdateData(List<String> fieldNames, List<Object[]> recordValues) {
+    if (CollectionUtils.isEmpty(fieldNames) || CollectionUtils.isEmpty(recordValues)) {
+      return 0L;
+    }
+    List<Object> objectList = asObjectList(fieldNames, recordValues);
+    JSONArray array = JSONUtil.parseArray(objectList);
+    JSONObject jsonObject = JSONUtil.createObj().set("data", array);
+    try {
+      sendData(jsonObject.toString());
+      return recordValues.size();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void sendData(String content) throws Exception {
+    final String loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load",
+        this.host,
+        this.httpPort,
+        this.dbName,
+        this.tbName);
+
+    final HttpClientBuilder httpClientBuilder = HttpClients
+        .custom()
+        .setRedirectStrategy(new DefaultRedirectStrategy() {
+          @Override
+          protected boolean isRedirectable(String method) {
+            return true;
+          }
+        });
+    try (CloseableHttpClient client = httpClientBuilder.build()) {
+      HttpPut put = new HttpPut(loadUrl);
+      StringEntity entity = new StringEntity(content, "UTF-8");
+      put.setHeader(HttpHeaders.EXPECT, "100-continue");
+      put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(this.username, this.password));
+      put.setHeader("strip_outer_array", "true");
+      put.setHeader("format", "JSON");
+      put.setHeader("json_root", "$.data");
+      put.setHeader("ignore_json_size", "true");
+      put.setHeader("Content-Type", "application/json");
+      put.setEntity(entity);
+      try (CloseableHttpResponse response = client.execute(put)) {
+        String loadResult = "";
+        if (response.getEntity() != null) {
+          loadResult = EntityUtils.toString(response.getEntity());
+        }
+        final int statusCode = response.getStatusLine().getStatusCode();
+        // statusCode 200 just indicates that doris be service is ok, not stream load
+        // you should see the output content to find whether stream load is success
+        if (statusCode != 200) {
+          throw new IOException(
+              String.format("Stream load failed, statusCode=%s load result=%s", statusCode, loadResult));
+        }
+      }
+    }
+  }
+
+  private String basicAuthHeader(String username, String password) {
+    final String tobeEncode = username + ":" + password;
+    byte[] encoded = Base64.getEncoder().encode(tobeEncode.getBytes(StandardCharsets.UTF_8));
+    return "Basic " + new String(encoded);
+  }
+
+  private List<Object> asObjectList(List<String> fieldNames, List<Object[]> recordValues) {
+    int fieldCount = Math.min(fieldNames.size(), recordValues.get(0).length);
+    List<Object> rows = new ArrayList<>(recordValues.size());
+    for (Object[] row : recordValues) {
+      Map<String, Object> columns = new LinkedHashMap<>(fieldCount);
+      for (int i = 0; i < fieldCount; ++i) {
+        Object rowValue = row[i];
+        if (row[i] instanceof Timestamp) {
+          rowValue = String.valueOf(rowValue);
+        }
+        columns.put(fieldNames.get(i), rowValue);
+      }
+      rows.add(columns);
+    }
+    return rows;
+  }
+}

dbswitch-product/dbswitch-product-doris/src/main/java/org/dromara/dbswitch/product/doris/FrontendEntity.java

@@ -0,0 +1,22 @@
+// Copyright tang.  All rights reserved.
+// https://gitee.com/inrgihc/dbswitch
+//
+// Use of this source code is governed by a BSD-style license
+//
+// Author: tang (inrgihc@126.com)
+// Date : 2020/1/2
+// Location: beijing , china
+/////////////////////////////////////////////////////////////
+package org.dromara.dbswitch.product.doris;
+
+import lombok.Data;
+
+@Data
+public class FrontendEntity {
+
+  private String ip;
+  private String httpport;
+  private Boolean alive;
+  private Boolean join;
+  private String role;
+}