remove hudi component,and add processing for fill dataType for hive col

datavane/tis#391
qlangtech · Nov 6, 2024 · 5b9b8fa · 5b9b8fa
1 parent bacf0d4
commit 5b9b8fa
Show file tree

Hide file tree

Showing 7 changed files with 39 additions and 7 deletions.
diff --git a/tis-datax/pom.xml b/tis-datax/pom.xml
@@ -70,12 +70,12 @@
         <module>tis-datax-mariadb-plugin</module>
 <!--        <module>tis-ds-mysql-mariadb-plugin</module>-->
         <module>tis-datax-local-executor</module>
-        <!-- 为测试通过暂时注释掉-->
+        <!-- 为测试通过暂时注释掉
         <module>tis-datax-hudi-dependency</module>
         <module>tis-datax-hudi-test</module>
         <module>tis-datax-hudi-plugin</module>
         <module>tis-datax-hudi-common</module>
-
+-->
         <module>tis-datax-hdfs-plugin</module>
         <module>tis-datax-hdfs-reader-writer-plugin</module>
 

diff --git a/.../tis-datax-dfs-plugin/src/main/java/com/qlangtech/tis/plugin/datax/AbstractDFSReader.java b/.../tis-datax-dfs-plugin/src/main/java/com/qlangtech/tis/plugin/datax/AbstractDFSReader.java
@@ -74,7 +74,7 @@ public void startScanDependency() {
      * ================================================================================
      */
     @SubForm(desClazz = SelectedTab.class, idListGetScript = "return com.qlangtech.tis.plugin.datax.DataXDFSReaderWithMeta.getDFSFiles(filter);", atLeastOne = true)
-    public transient List<SelectedTab> selectedTabs;
+    public transient List<ISelectedTab> selectedTabs;
 
     public abstract List<DataXDFSReaderWithMeta.TargetResMeta> getSelectedEntities();
 

diff --git a/...datax-dfs-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataXDFSReaderWithMeta.java b/...datax-dfs-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataXDFSReaderWithMeta.java
@@ -137,7 +137,7 @@ public List<ColumnMetaData> getTableMetadata(boolean inSink, EntityName table) t
     @Override
     public IGroupChildTaskIterator getSubTasks(Predicate<ISelectedTab> filter) {
 
-        final List<SelectedTab> tabs = selectedTabs;
+        final List<ISelectedTab> tabs = selectedTabs;
         final int tabsLength = tabs.size();
         AtomicInteger selectedTabIndex = new AtomicInteger(0);
         ConcurrentHashMap<String, List<DBDataXChildTask>> groupedInfo = new ConcurrentHashMap();
@@ -156,7 +156,7 @@ public boolean hasNext() {
 
             @Override
             public IDataxReaderContext next() {
-                SelectedTab tab = tabs.get(currentIndex);
+                ISelectedTab tab = tabs.get(currentIndex);
 
                 ColumnMetaData.fillSelectedTabMeta(tab, (t) -> {
                     List<ColumnMetaData> colsMeta = getFTPFileMetaData(EntityName.parse(t.getName()), dfs);

diff --git a/...lat-table-builder-plugin/src/main/java/com/qlangtech/tis/hive/reader/DataXHiveReader.java b/...lat-table-builder-plugin/src/main/java/com/qlangtech/tis/hive/reader/DataXHiveReader.java
@@ -35,6 +35,8 @@
 import com.qlangtech.tis.plugin.datax.AbstractDFSReader;
 import com.qlangtech.tis.plugin.datax.DataXDFSReaderWithMeta;
 import com.qlangtech.tis.plugin.datax.SelectedTab;
+import com.qlangtech.tis.plugin.datax.common.BasicDataXRdbmsReader;
+import com.qlangtech.tis.plugin.datax.common.TableColsMeta;
 import com.qlangtech.tis.plugin.datax.format.FileFormat;
 import com.qlangtech.tis.plugin.ds.ColumnMetaData;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
@@ -69,9 +71,29 @@ public static String getDftTemplate() {
         return IOUtils.loadResourceFromClasspath(DataXHiveReader.class, "DataXHiveReader-tpl.json");
     }
 
+    private transient int preSelectedTabsHash;
+
     @Override
     public List<ISelectedTab> getSelectedTabs() {
-        return Objects.requireNonNull(this.selectedTabs, "selectedTabs can not be null").stream().collect(Collectors.toList());
+        //BasicDataXRdbmsReader
+        Objects.requireNonNull(this.selectedTabs, "selectedTabs can not be null");
+        if (this.preSelectedTabsHash == selectedTabs.hashCode()) {
+            return selectedTabs;
+        }
+
+        try (TableColsMeta colMeta = this.getDfsLinker().getTabsMeta()) {
+            selectedTabs = this.selectedTabs.stream().map((tab) -> {
+                ColumnMetaData.fillSelectedTabMeta(tab, (t) -> {
+                    return colMeta.get(t.getName());
+                });
+                return tab;
+            }).collect(Collectors.toList());
+            this.preSelectedTabsHash = selectedTabs.hashCode();
+            return selectedTabs;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+
     }
 
     @Override

diff --git a/...-flat-table-builder-plugin/src/main/java/com/qlangtech/tis/hive/reader/HiveDFSLinker.java b/...-flat-table-builder-plugin/src/main/java/com/qlangtech/tis/hive/reader/HiveDFSLinker.java
@@ -35,6 +35,7 @@
 import com.qlangtech.tis.plugin.annotation.FormFieldType;
 import com.qlangtech.tis.plugin.annotation.Validator;
 import com.qlangtech.tis.plugin.datax.common.BasicDataXRdbmsWriter;
+import com.qlangtech.tis.plugin.datax.common.TableColsMeta;
 import com.qlangtech.tis.plugin.datax.format.FileFormat;
 import com.qlangtech.tis.plugin.datax.format.TextFormat;
 import com.qlangtech.tis.plugin.tdfs.IExclusiveTDFSType;
@@ -86,6 +87,11 @@ public FileSystemFactory getFs() {
         return fileSystem;
     }
 
+    public TableColsMeta getTabsMeta() {
+        Hiveserver2DataSourceFactory dsFactory = getDataSourceFactory();
+        return new TableColsMeta(dsFactory, dsFactory.dbName);
+    }
+
 
     public Hiveserver2DataSourceFactory getDataSourceFactory() {
         if (StringUtils.isBlank(this.linker)) {

diff --git a/tis-datax/tis-hive-plugin/src/main/java/com/qlangtech/tis/hive/DefaultHiveMetaStore.java b/tis-datax/tis-hive-plugin/src/main/java/com/qlangtech/tis/hive/DefaultHiveMetaStore.java
@@ -87,6 +87,10 @@ public HiveTable getTable(String database, String tableName) {
             StorageDescriptor storageDesc = table.getSd();
 
 
+//            for (FieldSchema hiveCol : storageDesc.getCols()) {
+//
+//            }
+
             return new HiveTable(table.getTableName()) {
                 @Override
                 public List<String> getPartitionKeys() {

diff --git a/tis-incr/pom.xml b/tis-incr/pom.xml
@@ -67,7 +67,7 @@
         <module>tis-flink-cdc-mongdb-plugin</module>
 <!--        <module>tis-sink-clickhouse-plugin</module>-->
 
-<!--    为了编译通过暂时先注释掉 -->   <module>tis-sink-hudi-plugin</module>
+<!--    为了编译通过暂时先注释掉    <module>tis-sink-hudi-plugin</module>-->
 
         <!--        <module>tis-chunjun-dependency</module>-->
         <module>tis-flink-chunjun-mysql-plugin</module>