refactor: move dataSource escapeChar from 'CreateTableSqlBuilder' to …

…'DataSourceMeta'
qlangtech · Oct 12, 2022 · 93bc02e · 93bc02e
1 parent 261050f
commit 93bc02e
Show file tree

Hide file tree

Showing 8 changed files with 747 additions and 500 deletions.
diff --git a/...iter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java b/...iter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java
diff --git a/common/src/main/java/com/alibaba/datax/plugin/rdbms/writer/util/EscapeableEntity.java b/common/src/main/java/com/alibaba/datax/plugin/rdbms/writer/util/EscapeableEntity.java
@@ -30,4 +30,12 @@ protected String escapeEntity(String val) {
             return val;
         }
     }
+
+    protected String unescapeEntity(String val) {
+        if (containEscapeChar) {
+            return StringUtils.remove(val, this.escapeChar);
+        } else {
+            return val;
+        }
+    }
 }
diff --git a/common/src/main/java/com/alibaba/datax/plugin/rdbms/writer/util/SelectCols.java b/common/src/main/java/com/alibaba/datax/plugin/rdbms/writer/util/SelectCols.java
@@ -26,8 +26,12 @@ public Iterator<String> iterator() {
     }
 
     public static SelectCols createSelectCols(Configuration conf) {
+        return createSelectCols(conf, null);
+    }
+
+    public static SelectCols createSelectCols(Configuration conf, String escapeChar) {
         return new SelectCols(
-                conf.getList(Key.COLUMN, String.class), conf.get(Key.ESCAPE_CHAR, String.class));
+                conf.getList(Key.COLUMN, String.class), StringUtils.defaultString(escapeChar, conf.get(Key.ESCAPE_CHAR, String.class)));
     }
 
     public static SelectCols createSelectCols(List<String> allColumns) {
@@ -36,8 +40,8 @@ public static SelectCols createSelectCols(List<String> allColumns) {
 
     private SelectCols(List<String> columns, String escapeChar) {
         super(escapeChar);
-        this.columns = columns;
-        if (columns == null || columns.isEmpty()) {
+        this.columns = columns.stream().map((c) -> unescapeEntity(c)).collect(Collectors.toList());
+        if (this.columns == null || this.columns.isEmpty()) {
             throw new IllegalArgumentException("param colums can not be empty ");
         }
     }
@@ -118,4 +122,5 @@ public int size() {
     public boolean containsCol(String name) {
         return columns.contains(name);
     }
+
 }
diff --git a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/CommonRdbmsReader.java b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/CommonRdbmsReader.java
@@ -18,6 +18,7 @@
 import com.google.common.collect.Lists;
 import com.qlangtech.tis.plugin.ds.ColumnMetaData;
 import com.qlangtech.tis.plugin.ds.IDataSourceFactoryGetter;
+import com.qlangtech.tis.plugin.ds.TableNotFoundException;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
@@ -202,8 +203,13 @@ public void startRead(Configuration readerSliceConfig,
 
             Connection conn = DBUtil.getConnection(this.readerDataSourceFactoryGetter, jdbcUrl,
                     username, password);
-            Map<String, ColumnMetaData> tabCols = ColumnMetaData.toMap(this.readerDataSourceFactoryGetter.getDataSourceFactory()
-                    .getTableMetadata(conn, table));
+            Map<String, ColumnMetaData> tabCols = null;
+            try {
+                tabCols = ColumnMetaData.toMap(this.readerDataSourceFactoryGetter.getDataSourceFactory()
+                        .getTableMetadata(conn, table));
+            } catch (TableNotFoundException e) {
+                throw new RuntimeException(e);
+            }
             if (MapUtils.isEmpty(tabCols)) {
                 throw new IllegalStateException("table:" + table + " relevant tabCols can not be empty");
             }

diff --git a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DBUtil.java b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DBUtil.java
@@ -8,13 +8,15 @@
 import com.alibaba.datax.plugin.rdbms.writer.util.SelectTable;
 import com.alibaba.druid.sql.parser.SQLParserUtils;
 import com.alibaba.druid.sql.parser.SQLStatementParser;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.qlangtech.tis.datax.impl.DataxReader;
 import com.qlangtech.tis.datax.impl.DataxWriter;
 import com.qlangtech.tis.extension.Describable;
 import com.qlangtech.tis.offline.DataxUtils;
 import com.qlangtech.tis.plugin.ds.ColumnMetaData;
 import com.qlangtech.tis.plugin.ds.IDataSourceFactoryGetter;
+import com.qlangtech.tis.plugin.ds.TableNotFoundException;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -522,8 +524,12 @@ public static List<String> getTableColumns(DataBaseType dataBaseType, IDataSourc
     }
 
     private static List<String> getTableColums(IDataSourceFactoryGetter dataSourceFactoryGetter, SelectTable tableName) {
-        List<ColumnMetaData> tabMeta = dataSourceFactoryGetter.getDataSourceFactory().getTableMetadata(tableName.getUnescapeTabName());
-        return tabMeta.stream().map((c) -> c.getName()).collect(Collectors.toList());
+        try {
+            List<ColumnMetaData> tabMeta = dataSourceFactoryGetter.getDataSourceFactory().getTableMetadata(tableName.getUnescapeTabName());
+            return tabMeta.stream().map((c) -> c.getName()).collect(Collectors.toList());
+        } catch (TableNotFoundException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     public static List<String> getTableColumnsByConn(DataBaseType dataBaseType, IDataSourceFactoryGetter conn, SelectTable tableName, String basicMsg) {
@@ -580,13 +586,24 @@ public static List<ColumnMetaData> getColumnMetaData(
      */
     public static List<ColumnMetaData> getColumnMetaData(
             Optional<Connection> connection, IDataSourceFactoryGetter dsGetter, SelectTable tableName, SelectCols userConfiguredColumns) {
-//        if (userConfiguredColumns.) {
-//            throw new IllegalArgumentException("param userConfiguredColumns can not be empty");
-//        }
-        List<ColumnMetaData> tabCols = connection.isPresent()
-                ? dsGetter.getDataSourceFactory().getTableMetadata(connection.get(), tableName.getUnescapeTabName())
-                : dsGetter.getDataSourceFactory().getTableMetadata(tableName.getUnescapeTabName());
-        return tabCols.stream().filter((c) -> userConfiguredColumns.containsCol(c.getName())).collect(Collectors.toList());
+
+        Map<String, ColumnMetaData> colMapper = null;
+        try {
+            List<ColumnMetaData> tabCols = connection.isPresent()
+                    ? dsGetter.getDataSourceFactory().getTableMetadata(connection.get(), tableName.getUnescapeTabName())
+                    : dsGetter.getDataSourceFactory().getTableMetadata(tableName.getUnescapeTabName());
+            colMapper = tabCols.stream().collect(Collectors.toMap((c) -> c.getName(), (c) -> c));
+        } catch (TableNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+
+        List<ColumnMetaData> result = Lists.newArrayList();
+        for (String col : userConfiguredColumns) {
+            result.add(Objects.requireNonNull(
+                    colMapper.get(col), "col:" + col + " relevant meta can not be null"));
+        }
+        return result;
+        // return tabCols.stream().filter((c) -> userConfiguredColumns.containsCol(c.getName())).collect(Collectors.toList());
         // return tabCols;
 //        Statement statement = null;
 //        ResultSet rs = null;