add tests

Author: polyzos

.idea/vcs.xml

@@ -20,16 +20,16 @@
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.row.InternalRow;
 
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 /**
- * The lookuper is used to fetch rows from a table either by primary key or by a key prefix,
- * depending on how it is created.
+ * Lookuper performs point lookups by primary key or key prefix.
  *
  * <p>Instances are created via {@link Lookup} builders (e.g., {@code table.newLookup()}) and can
- * target primary-key lookups (exact match) or prefix-key lookups. Some operations are only
- * supported for primary-key tables and will throw {@link UnsupportedOperationException} otherwise.
+ * target primary-key lookups (exact match) or prefix-key lookups.
+ *
+ * <p>Note: Full scans are not part of the Lookuper API. Use the Scan API instead: {@code
+ * table.newScan().createBatchScanner(...)}.
  *
  * @since 0.6
  */
@@ -53,36 +53,4 @@ public interface Lookuper {
      *     empty if not found; the future may complete exceptionally on RPC or server errors
      */
     CompletableFuture<LookupResult> lookup(InternalRow lookupKey);
-
-    /**
-     * Returns all current values of a primary-key KV table using a point-in-time snapshot. This is
-     * not a streaming/cursor operation; it materializes the full result as a list.
-     *
-     * <p>For partitioned tables, prefer {@link #snapshotAllPartition(String)}.
-     *
-     * @return a future completing with an unordered list of {@link InternalRow} representing the
-     *     current values; the future may complete exceptionally if unsupported by the lookuper or
-     *     on RPC/server errors
-     * @throws UnsupportedOperationException if the implementation does not support snapshot reads
-     */
-    default CompletableFuture<List<InternalRow>> snapshotAll() {
-        throw new UnsupportedOperationException(
-                "snapshotAll() is only supported for primary key lookuper.");
-    }
-
-    /**
-     * Returns all current values for the specified partition of a partitioned primary-key table,
-     * using a point-in-time snapshot.
-     *
-     * @param partitionName the partition identifier (e.g., a date-based partition value)
-     * @return a future completing with an unordered list of {@link InternalRow} for that partition;
-     *     the future may complete exceptionally if unsupported by the lookuper or on RPC/server
-     *     errors
-     * @throws UnsupportedOperationException if the implementation does not support partition
-     *     snapshot reads
-     */
-    default CompletableFuture<List<InternalRow>> snapshotAllPartition(String partitionName) {
-        throw new UnsupportedOperationException(
-                "snapshotAllPartition() is not supported by this lookuper.");
-    }
 }

fluss-client/src/main/java/org/apache/fluss/client/lookup/Lookuper.java

@@ -20,36 +20,20 @@
 import org.apache.fluss.bucketing.BucketingFunction;
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.table.getter.PartitionGetter;
-import org.apache.fluss.exception.LeaderNotAvailableException;
 import org.apache.fluss.exception.PartitionNotExistException;
-import org.apache.fluss.exception.TableNotPartitionedException;
 import org.apache.fluss.metadata.DataLakeFormat;
-import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
-import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.record.DefaultValueRecordBatch;
-import org.apache.fluss.record.ValueRecord;
-import org.apache.fluss.record.ValueRecordReadContext;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.decode.RowDecoder;
 import org.apache.fluss.row.encode.KeyEncoder;
 import org.apache.fluss.row.encode.ValueDecoder;
-import org.apache.fluss.rpc.gateway.TabletServerGateway;
-import org.apache.fluss.rpc.messages.FullScanRequest;
-import org.apache.fluss.rpc.messages.FullScanResponse;
-import org.apache.fluss.rpc.protocol.Errors;
 import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
 
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 import static org.apache.fluss.client.utils.ClientUtils.getPartitionId;
@@ -58,10 +42,8 @@
 /**
  * Client-side lookuper implementation for primary-key tables.
  *
- * <p>This class supports single-key lookups as well as bounded, point-in-time snapshot reads that
- * return all current values of a KV table (and per-partition for partitioned tables). Snapshot
- * reads are executed via the FULL_SCAN RPC and aggregate results across the leaders of all buckets
- * on each target server.
+ * <p>This class supports single-key lookups. For any scan semantics (full scan, limited scan, or
+ * snapshot-by-id), use the Scan API instead.
  */
 class PrimaryKeyLookuper implements Lookuper {
 
@@ -185,124 +167,4 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
      *
      * @return a future with an unordered list of current rows
      */
-    @Override
-    public CompletableFuture<List<InternalRow>> snapshotAll() {
-        if (tableInfo.isPartitioned()) {
-            CompletableFuture<List<InternalRow>> result = new CompletableFuture<>();
-            result.completeExceptionally(
-                    new TableNotPartitionedException(
-                            "Table is partitioned. Please use snapshotAllPartition(partitionName)."));
-
-            return result;
-        }
-
-        return executeFullScan(Optional.empty());
-    }
-
-    /**
-     * Returns all current values for the specified partition of a partitioned primary-key table
-     * using a point-in-time snapshot.
-     *
-     * @param partitionName the partition identifier (for example, a date string)
-     * @return a future with an unordered list of current rows in the partition
-     */
-    @Override
-    public CompletableFuture<List<InternalRow>> snapshotAllPartition(String partitionName) {
-        if (!tableInfo.isPartitioned()) {
-            CompletableFuture<List<InternalRow>> result = new CompletableFuture<>();
-            result.completeExceptionally(
-                    new TableNotPartitionedException(
-                            "Table is not partitioned. Please use snapshotAll()."));
-            return result;
-        }
-
-        // Resolve partition id from name
-        TablePath tablePath = tableInfo.getTablePath();
-        PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath, partitionName);
-
-        try {
-            metadataUpdater.checkAndUpdatePartitionMetadata(physicalTablePath);
-        } catch (PartitionNotExistException e) {
-            CompletableFuture<List<InternalRow>> result = new CompletableFuture<>();
-            result.completeExceptionally(e);
-            return result;
-        }
-
-        Optional<Long> partitionId = metadataUpdater.getPartitionId(physicalTablePath);
-        return executeFullScan(partitionId);
-    }
-
-    /**
-     * Decodes and aggregates FULL_SCAN RPC responses into a list of rows.
-     *
-     * <p>Throws a mapped client exception if any response contains an error.
-     *
-     * @param responseFutures futures of server responses per target leader node
-     * @return aggregated, unordered rows
-     */
-    private List<InternalRow> decodeFullScanResponses(
-            List<CompletableFuture<FullScanResponse>> responseFutures) {
-        List<InternalRow> out = new ArrayList<>();
-        for (CompletableFuture<FullScanResponse> responseFuture : responseFutures) {
-            FullScanResponse response = responseFuture.join();
-
-            if (response.hasErrorCode() && response.getErrorCode() != Errors.NONE.code()) {
-                Errors err = Errors.forCode(response.getErrorCode());
-                throw err.exception(
-                        response.hasErrorMessage() ? response.getErrorMessage() : err.message());
-            }
-
-            if (response.hasRecords()) {
-                ByteBuffer buffer = ByteBuffer.wrap(response.getRecords());
-                DefaultValueRecordBatch values = DefaultValueRecordBatch.pointToByteBuffer(buffer);
-
-                ValueRecordReadContext context =
-                        new ValueRecordReadContext(kvValueDecoder.getRowDecoder());
-                for (ValueRecord record : values.records(context)) {
-                    out.add(record.getRow());
-                }
-            }
-        }
-        return out;
-    }
-
-    private CompletableFuture<List<InternalRow>> executeFullScan(Optional<Long> partitionIdOpt) {
-        Long partitionId = partitionIdOpt.orElse(null);
-        long tableId = tableInfo.getTableId();
-        int numBuckets = tableInfo.getNumBuckets();
-
-        // Find leader tablet/servers for this table/partition
-        HashSet<Integer> leaderServers = new HashSet<>();
-        for (int bucketId = 0; bucketId < numBuckets; bucketId++) {
-            TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
-
-            // ensure metadata for this bucket is present
-            metadataUpdater.checkAndUpdateMetadata(tableInfo.getTablePath(), tableBucket);
-            int leader = metadataUpdater.leaderFor(tableBucket);
-            leaderServers.add(leader);
-        }
-
-        List<CompletableFuture<FullScanResponse>> responseFutures = new ArrayList<>();
-        for (int leader : leaderServers) {
-            TabletServerGateway gateway = metadataUpdater.newTabletServerClientForNode(leader);
-            if (gateway == null) {
-                CompletableFuture<List<InternalRow>> result = new CompletableFuture<>();
-                result.completeExceptionally(
-                        new LeaderNotAvailableException(
-                                "Leader server " + leader + " is not found in metadata cache."));
-                return result;
-            }
-
-            FullScanRequest request = new FullScanRequest();
-            request.setTableId(tableId);
-
-            if (partitionId != null) {
-                request.setPartitionId(partitionId);
-            }
-            responseFutures.add(gateway.fullScan(request));
-        }
-        // Decode all responses and aggregate rows
-        return CompletableFuture.allOf(responseFutures.toArray(new CompletableFuture[0]))
-                .thenApply(v -> decodeFullScanResponses(responseFutures));
-    }
 }

fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java

@@ -76,16 +76,27 @@ public interface Scan {
     /**
      * Creates a {@link BatchScanner} to read current data in the given table bucket for this scan.
      *
-     * <p>Note: this API doesn't support pre-configured with {@link #project}.
+     * <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>
      */
     BatchScanner createBatchScanner(TableBucket tableBucket);
 
     /**
      * Creates a {@link BatchScanner} to read given snapshot data in the given table bucket for this
      * scan.
      *
-     * <p>Note: this API doesn't support pre-configured with {@link #project} and {@link
-     * #limit(int)} and only support for Primary Key Tables.
+     * <p>Notes:
+     *
+     * <ul>
+     *   <li>Projection configured via {@link #project} is supported client-side.
+     *   <li>{@link #limit(int)} is not supported on snapshot scans.
+     *   <li>Only supported for primary-key tables.
+     * </ul>
      */
     BatchScanner createBatchScanner(TableBucket tableBucket, long snapshotId);
 }

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

@@ -20,7 +20,6 @@
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.table.scanner.PartitionFilter;
 import org.apache.fluss.exception.LeaderNotAvailableException;
-import org.apache.fluss.exception.TableNotPartitionedException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
@@ -58,8 +57,8 @@
  * Default implementation of {@link BatchScanner} that performs a full scan against tablet servers.
  *
  * <p>This scanner issues FULL_SCAN RPCs to the leaders of all buckets and aggregates the results.
- * It returns all current values at a point in time for primary-key tables. The first call to
- * {@link #pollBatch(Duration)} returns the complete snapshot; subsequent calls return {@code null}.
+ * It returns all current values at a point in time for primary-key tables. The first call to {@link
+ * #pollBatch(Duration)} returns the complete snapshot; subsequent calls return {@code null}.
  *
  * <p>Note: For partitioned tables, callers may provide a {@link PartitionFilter} with a partition
  * name to restrict the scan to a single partition.
@@ -116,7 +115,11 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
         } catch (TimeoutException e) {
             // try again in next poll
             return CloseableIterator.emptyIterator();
+        } catch (RuntimeException re) {
+            // propagate runtime exceptions (e.g., IllegalArgumentException) without wrapping
+            throw re;
         } catch (Exception e) {
+            // wrap checked exceptions into IOException as the API declares
             throw new IOException(e);
         }
     }
@@ -125,8 +128,8 @@ private List<CompletableFuture<FullScanResponse>> issueFullScanRequests() {
         Long partitionId = null;
         if (tableInfo.isPartitioned()) {
             if (partitionFilter == null || partitionFilter.getPartitionName() == null) {
-                throw new TableNotPartitionedException(
-                        "Partition filter is required for partitioned table full scan.");
+                throw new IllegalArgumentException(
+                        "Full scan on a partitioned table requires a PartitionFilter with a partition name.");
             }
             TablePath tablePath = tableInfo.getTablePath();
             PhysicalTablePath physicalTablePath =