add bounded param for arrow source

Signed-off-by: chenxu <[email protected]>
lakesoul-io · Nov 13, 2024 · 2f7ad34 · 2f7ad34
1 parent 2071415
commit 2f7ad34
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/...l-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulOneSplitRecordsReader.java b/...l-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulOneSplitRecordsReader.java
@@ -274,6 +274,7 @@ public Set<String> finishedSplits() {
 
     @Override
     public void close() throws Exception {
+        LOG.info("Close reader split {}, read num {}", splitId, totalRead);
         if (this.currentVCR != null) {
             this.currentVCR.close();
             this.currentVCR = null;

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
@@ -31,10 +31,12 @@ public static LakeSoulArrowSource create(
         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(),
+                isBounded,
                 tableRowType,
                 tablePartitionKeys.primaryKeys,
                 tablePartitionKeys.rangeKeys
@@ -56,8 +58,8 @@ public static LakeSoulArrowSource create(
                 tableInfo,
                 tableId,
                 conf.toMap(),
-                tableRowType,
                 isBounded,
+                tableRowType,
                 tablePartitionKeys.primaryKeys,
                 tablePartitionKeys.rangeKeys,
                 remainingPartitions
@@ -68,8 +70,8 @@ public static LakeSoulArrowSource create(
             TableInfo tableInfo,
             TableId tableId,
             Map<String, String> optionParams,
-            RowType tableRowType,
             boolean isBounded,
+            RowType tableRowType,
             List<String> pkColumns,
             List<String> partitionColumns,
             List<Map<String, String>> remainingPartitions
@@ -94,6 +96,7 @@ public static LakeSoulArrowSource create(
             TableInfo tableInfo,
             TableId tableId,
             Map<String, String> optionParams,
+            boolean isBounded,
             RowType tableRowType,
             List<String> pkColumns,
             List<String> partitionColumns
@@ -103,7 +106,7 @@ public static LakeSoulArrowSource create(
                 tableRowType,
                 tableRowType,
                 tableRowType,
-                false,
+                isBounded,
                 pkColumns,
                 partitionColumns,
                 optionParams,