support batch merge source

Signed-off-by: fphantam <[email protected]>
lakesoul-io · Nov 13, 2024 · 2071415 · 2071415
1 parent 54e7198
commit 2071415
Showing 1 changed file with 49 additions and 0 deletions.
diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/source/arrow/LakeSoulArrowSource.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/source/arrow/LakeSoulArrowSource.java
@@ -41,6 +41,55 @@ public static LakeSoulArrowSource create(
         );
     }
 
+    public static LakeSoulArrowSource create(
+            String tableNamespace,
+            String tableName,
+            Configuration conf,
+            List<Map<String, String>> remainingPartitions
+    ) throws IOException {
+        TableId tableId = new TableId(LakeSoulCatalog.CATALOG_NAME, tableNamespace, tableName);
+        TableInfo tableInfo = DataOperation.dbManager().getTableInfoByNameAndNamespace(tableName, tableNamespace);
+        RowType tableRowType = ArrowUtils.fromArrowSchema(Schema.fromJSON(tableInfo.getTableSchema()));
+        DBUtil.TablePartitionKeys tablePartitionKeys = DBUtil.parseTableInfoPartitions(tableInfo.getPartitions());
+        boolean isBounded = conf.getBoolean("IS_BOUNDED", false);
+        return new LakeSoulArrowSource(
+                tableInfo,
+                tableId,
+                conf.toMap(),
+                tableRowType,
+                isBounded,
+                tablePartitionKeys.primaryKeys,
+                tablePartitionKeys.rangeKeys,
+                remainingPartitions
+        );
+    }
+
+    LakeSoulArrowSource(
+            TableInfo tableInfo,
+            TableId tableId,
+            Map<String, String> optionParams,
+            RowType tableRowType,
+            boolean isBounded,
+            List<String> pkColumns,
+            List<String> partitionColumns,
+            List<Map<String, String>> remainingPartitions
+    ) {
+        super(
+                tableId,
+                tableRowType,
+                tableRowType,
+                tableRowType,
+                isBounded,
+                pkColumns,
+                partitionColumns,
+                optionParams,
+                remainingPartitions,
+                null,
+                null
+        );
+        this.encodedTableInfo = tableInfo.toByteArray();
+    }
+
     LakeSoulArrowSource(
             TableInfo tableInfo,
             TableId tableId,