[Feature-#1781][connector] [doris] supports doris pre-sql and post-sql

DTStack · Aug 3, 2023 · b1471e3 · b1471e3
1 parent d1608c4
commit b1471e3
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/...r-doris/src/main/java/com/dtstack/chunjun/connector/doris/sink/DorisDynamicTableSink.java b/...r-doris/src/main/java/com/dtstack/chunjun/connector/doris/sink/DorisDynamicTableSink.java
@@ -116,6 +116,8 @@ private JdbcOutputFormatBuilder jdbcBuilder(RowType rowType, DorisConfig dorisCo
                 (CollectionUtil.isNullOrEmpty(jdbcConfig.getUniqueKey()))
                         ? EWriteMode.INSERT.name()
                         : EWriteMode.UPDATE.name());
+        jdbcConfig.setPreSql(dorisConfig.getPreSql());
+        jdbcConfig.setPostSql(dorisConfig.getPostSql());
 
         builder.setColumnNameList(columnNameList);
         builder.setColumnTypeList(columnTypeList);

diff --git a/...ris/src/main/java/com/dtstack/chunjun/connector/doris/table/DorisDynamicTableFactory.java b/...ris/src/main/java/com/dtstack/chunjun/connector/doris/table/DorisDynamicTableFactory.java
@@ -42,6 +42,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -53,6 +54,8 @@
 import static com.dtstack.chunjun.connector.jdbc.options.JdbcLookupOptions.DRUID_PREFIX;
 import static com.dtstack.chunjun.connector.jdbc.options.JdbcLookupOptions.VERTX_PREFIX;
 import static com.dtstack.chunjun.connector.jdbc.options.JdbcLookupOptions.getLibConfMap;
+import static com.dtstack.chunjun.connector.jdbc.options.JdbcSinkOptions.SINK_POST_SQL;
+import static com.dtstack.chunjun.connector.jdbc.options.JdbcSinkOptions.SINK_PRE_SQL;
 
 /** declare doris table factory info. */
 public class DorisDynamicTableFactory extends JdbcDynamicTableFactory
@@ -148,6 +151,12 @@ protected void validateConfigOptions(ReadableConfig config, ResolvedSchema table
 
     private static DorisConfig getConfByOptions(ReadableConfig config) {
         DorisConfig dorisConfig = new DorisConfig();
+        if (StringUtils.isNotEmpty(config.get(SINK_PRE_SQL))) {
+            dorisConfig.setPreSql(Arrays.asList(config.get(SINK_PRE_SQL).split(";")));
+        }
+        if (StringUtils.isNotEmpty(config.get(SINK_POST_SQL))) {
+            dorisConfig.setPostSql(Arrays.asList(config.get(SINK_POST_SQL).split(";")));
+        }
 
         dorisConfig.setFeNodes(config.get(DorisOptions.FENODES));