[feat-#846][hdfs]hdfs sql support partitionColumn

DTStack · May 30, 2022 · a3fc935 · a3fc935
1 parent d6c7f89
commit a3fc935
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 2 deletions.
diff --git a/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/source/HdfsDynamicTableSource.java b/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/source/HdfsDynamicTableSource.java
@@ -48,10 +48,17 @@ public class HdfsDynamicTableSource implements ScanTableSource {
 
     private final HdfsConf hdfsConf;
     private final TableSchema tableSchema;
+    private final List<String> partitionKeyList;
 
     public HdfsDynamicTableSource(HdfsConf hdfsConf, TableSchema tableSchema) {
+        this(hdfsConf, tableSchema, new ArrayList<>());
+    }
+
+    public HdfsDynamicTableSource(
+            HdfsConf hdfsConf, TableSchema tableSchema, List<String> partitionKeyList) {
         this.hdfsConf = hdfsConf;
         this.tableSchema = tableSchema;
+        this.partitionKeyList = partitionKeyList;
     }
 
     @Override
@@ -61,10 +68,14 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
         String[] fieldNames = tableSchema.getFieldNames();
         List<FieldConf> columnList = new ArrayList<>(fieldNames.length);
         for (int i = 0; i < fieldNames.length; i++) {
+            String fieldName = fieldNames[i];
             FieldConf field = new FieldConf();
-            field.setName(fieldNames[i]);
+            field.setName(fieldName);
             field.setType(rowType.getTypeAt(i).asSummaryString());
             field.setIndex(i);
+            if (partitionKeyList.contains(fieldName)) {
+                field.setPart(true);
+            }
             columnList.add(field);
         }
         hdfsConf.setColumn(columnList);

diff --git a/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java b/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java
@@ -36,6 +36,7 @@
 import org.apache.flink.table.utils.TableSchemaUtils;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -94,11 +95,12 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         TableSchema physicalSchema =
                 TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
         HdfsConf hdfsConf = getHdfsConf(config);
+        List<String> partitionKeys = context.getCatalogTable().getPartitionKeys();
         hdfsConf.setParallelism(config.get(SourceOptions.SCAN_PARALLELISM));
         hdfsConf.setHadoopConfig(
                 HdfsOptions.getHadoopConfig(context.getCatalogTable().getOptions()));
 
-        return new HdfsDynamicTableSource(hdfsConf, physicalSchema);
+        return new HdfsDynamicTableSource(hdfsConf, physicalSchema, partitionKeys);
     }
 
     @Override