add updates

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/Scan.java

@@ -74,16 +74,20 @@ public interface Scan {
     LogScanner createLogScanner();
 
     /**
-     * Creates a {@link BatchScanner} to read current data in the given table bucket for this scan.
+     * Creates a {@link BatchScanner} to read data for this scan over the whole table.
      *
-     * <p>Notes:
-     *
-     * <ul>
-     *   <li>Projection configured via {@link #project} is supported client-side.
-     *   <li>For full scans (when {@link #limit(int)} is not set), the provided {@link TableBucket}
-     *       is used only to identify the table (and optional partition); the bucket id is ignored.
-     * </ul>
+     * <p>For full scans (when {@link #limit(int)} is not set), the returned scanner reads a single
+     * snapshot of current values across all buckets. Use {@link BatchScanner#snapshotAll()} for
+     * non-partitioned tables or {@link BatchScanner#snapshotAllPartition(String)} for partitioned
+     * tables.
+     */
+    BatchScanner createBatchScanner();
+
+    /**
+     * [Deprecated] Creates a {@link BatchScanner} to read current data in the given table bucket
+     * for this scan.
      */
+    @Deprecated
     BatchScanner createBatchScanner(TableBucket tableBucket);
 
     /**

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java

@@ -115,16 +115,27 @@ public LogScanner createLogScanner() {
     }
 
     @Override
-    public BatchScanner createBatchScanner(TableBucket tableBucket) {
+    public BatchScanner createBatchScanner() {
         if (limit == null) {
             if (!tableInfo.hasPrimaryKey()) {
                 throw new UnsupportedOperationException(
                         "Full scan BatchScanner is only supported for primary key tables.");
             }
-            // Full scan semantics: one-shot snapshot of current values across all buckets.
             return new DefaultBatchScanner(
                     tableInfo, conn.getMetadataUpdater(), projectedColumns, partitionFilter);
         }
+        // For limited scans, the legacy API requires a TableBucket to indicate which bucket.
+        // Keep behavior via deprecated overloads; here we throw to signal misuse.
+        throw new UnsupportedOperationException(
+                "Limit scan requires bucket specification; use createBatchScanner(TableBucket) instead.");
+    }
+
+    @Override
+    public BatchScanner createBatchScanner(TableBucket tableBucket) {
+        if (limit == null) {
+            // delegate to whole table scanner
+            return createBatchScanner();
+        }
         return new LimitBatchScanner(
                 tableInfo, tableBucket, conn.getMetadataUpdater(), projectedColumns, limit);
     }

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanner.java

@@ -38,6 +38,33 @@
 @PublicEvolving
 public interface BatchScanner extends Closeable {
 
+    /**
+     * Configure the scanner to read a snapshot over the whole table (non-partitioned tables) or the
+     * whole (single) partition must be selected via {@link #snapshotAllPartition(String)} for
+     * partitioned tables.
+     *
+     * <p>For partitioned tables, calling this without specifying a partition name will cause the
+     * implementation to throw an IllegalArgumentException when polling.
+     *
+     * @return a scanner configured for a one-shot snapshot over the current data
+     */
+    default BatchScanner snapshotAll() {
+        return this;
+    }
+
+    /**
+     * Configure the scanner to read a snapshot from the specified partition. Only applicable to
+     * partitioned tables.
+     *
+     * @param partitionName the partition to scan
+     * @return a scanner configured for a one-shot snapshot over the given partition
+     * @throws UnsupportedOperationException if the table is not partitioned
+     */
+    default BatchScanner snapshotAllPartition(String partitionName) {
+        throw new UnsupportedOperationException(
+                "Partition filter is only supported for partitioned tables");
+    }
+
     /**
      * Poll one batch records. The method should return null when reaching the end of the input.
      *

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/DefaultBatchScanner.java

@@ -65,6 +65,24 @@
  */
 public class DefaultBatchScanner implements BatchScanner {
 
+    @Override
+    public BatchScanner snapshotAll() {
+        return this;
+    }
+
+    @Override
+    public BatchScanner snapshotAllPartition(String partitionName) {
+        if (!tableInfo.isPartitioned()) {
+            throw new UnsupportedOperationException(
+                    "Partition filter is only supported for partitioned tables");
+        }
+        return new DefaultBatchScanner(
+                tableInfo,
+                metadataUpdater,
+                projectedFields,
+                PartitionFilter.ofPartitionName(partitionName));
+    }
+
     private final TableInfo tableInfo;
     private final MetadataUpdater metadataUpdater;
     @Nullable private final int[] projectedFields;

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/DefaultBatchScannerITCase.java

@@ -70,7 +70,7 @@ void testFullScanPrimaryKeyTable_returnsAllRows() throws Exception {
             writer.flush();
 
             // Full scan via Scan API (limit == null)
-            BatchScanner scanner = table.newScan().createBatchScanner(new TableBucket(tableId, 0));
+            BatchScanner scanner = table.newScan().createBatchScanner().snapshotAll();
             List<InternalRow> actualRows = collectRows(scanner);
 
             // Assert count
@@ -104,7 +104,8 @@ void testProjectionInFullScan() throws Exception {
             BatchScanner scanner =
                     table.newScan()
                             .project(projected)
-                            .createBatchScanner(new TableBucket(tableId, 0));
+                            .createBatchScanner()
+                            .snapshotAll();
             List<InternalRow> actualRows = collectRows(scanner);
 
             assertThat(actualRows).hasSize(5);
@@ -127,7 +128,7 @@ void testFilterOnNonPartitionedTableThrows() throws Exception {
                             () ->
                                     table.newScan()
                                             .filter(PartitionFilter.ofPartitionName("p=1"))
-                                            .createBatchScanner(new TableBucket(tableId, 0)))
+                                            .createBatchScanner())
                     .isInstanceOf(UnsupportedOperationException.class)
                     .hasMessageContaining(
                             "Partition filter is only supported for partitioned tables");
@@ -170,7 +171,8 @@ void testFullScanPartitionedTable_requiresPartitionFilter() throws Exception {
             assertThatThrownBy(
                             () ->
                                     table.newScan()
-                                            .createBatchScanner(new TableBucket(tableId, 0))
+                                            .createBatchScanner()
+                                            .snapshotAll()
                                             .pollBatch(java.time.Duration.ofMillis(1)))
                     .isInstanceOf(IllegalArgumentException.class)
                     .hasMessageContaining("requires a PartitionFilter with a partition name");
@@ -179,7 +181,8 @@ void testFullScanPartitionedTable_requiresPartitionFilter() throws Exception {
             BatchScanner scannerA =
                     table.newScan()
                             .filter(PartitionFilter.ofPartitionName("a"))
-                            .createBatchScanner(new TableBucket(tableId, 0));
+                            .createBatchScanner()
+                            .snapshotAllPartition("a");
             List<InternalRow> rowsA = collectRows(scannerA);
             assertThat(rowsA).hasSize(4);
             List<Object[]> valuesA = toValues(rowsA, schema.getRowType());
@@ -195,7 +198,8 @@ void testFullScanPartitionedTable_requiresPartitionFilter() throws Exception {
             BatchScanner scannerB =
                     table.newScan()
                             .filter(PartitionFilter.ofPartitionName("b"))
-                            .createBatchScanner(new TableBucket(tableId, 0));
+                            .createBatchScanner()
+                            .snapshotAllPartition("b");
             List<InternalRow> rowsB = collectRows(scannerB);
             assertThat(rowsB).hasSize(4);
             List<Object[]> valuesB = toValues(rowsB, schema.getRowType());
@@ -224,7 +228,7 @@ void testFullScanOnLogTableThrows() throws Exception {
 
         try (Table table = conn.getTable(tablePath)) {
             assertThatThrownBy(
-                            () -> table.newScan().createBatchScanner(new TableBucket(tableId, 0)))
+                            () -> table.newScan().createBatchScanner())
                     .isInstanceOf(UnsupportedOperationException.class)
                     .hasMessageContaining(
                             "Full scan BatchScanner is only supported for primary key tables.");