[rust] Add reusable UnionRead source boundaries

.github/workflows/client-integration.yml

@@ -194,6 +194,11 @@ jobs:
         env:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full
+      - name: UnionRead boundary integration tests
+        run: cargo test --locked -p fluss-lake --features integration_tests --test test_prepare
+        env:
+          RUST_LOG: DEBUG
+          RUST_BACKTRACE: full
 
   gateway-integration:
     needs: [detect-changes, build-server-image]

.github/workflows/rust-build-and-test.yml

@@ -78,6 +78,36 @@ jobs:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full
 
+  lake-msrv:
+    name: "Lake MSRV (1.91)"
+    # The development toolchain floats to stable; check the lake package's
+    # declared MSRV explicitly without raising the core workspace's MSRV.
+    timeout-minutes: 60
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install the lake MSRV toolchain
+        run: rustup toolchain install 1.91.0 --profile minimal
+
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
+        with:
+          workspaces: fluss-rust
+
+      - name: Check default features on the declared MSRV
+        run: cargo +1.91.0 check -p fluss-lake --locked --all-targets
+
+      # Compile integration targets without starting Docker or Java services.
+      - name: Check all features on the declared MSRV
+        run: cargo +1.91.0 check -p fluss-lake --locked --all-features --all-targets
+
+      - name: Test source preparation on the declared MSRV
+        run: cargo +1.91.0 test -p fluss-lake --locked --lib
+        env:
+          RUST_LOG: DEBUG
+          RUST_BACKTRACE: full
+
   proto-up-to-date:
     timeout-minutes: 15
     runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-24.04-arm' || 'ubuntu-24.04' }}

fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlanner.java

@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.metadata;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.admin.OffsetSpec;
+import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
+import org.apache.fluss.metadata.PartitionInfo;
+import org.apache.fluss.metadata.PartitionSpec;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.ExceptionUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Captures structured bounded UnionRead inputs for client or engine planning.
+ *
+ * <p>Calls block the caller and must run on a planning thread, not an Admin RPC callback thread.
+ * They use the supplied Admin without closing it or opening lake files. Request timeouts follow the
+ * client configuration; there is no additional whole-capture timeout. Interrupted waits abort
+ * capture, but do not cancel RPCs already issued. Concurrent calls have independent capture state.
+ *
+ * <p>The result fixes a schema, an optional lake snapshot and per-bucket log ranges; it is neither
+ * a cross-bucket transactional snapshot nor a retention lease. Without a lake baseline, PK reads
+ * reconstruct only keys represented in retained logs. KV snapshot fallback and shared historical
+ * logs are not yet supported. Consumers must enforce the captured scope on both lake and log
+ * inputs.
+ *
+ * @since 1.1
+ */
+@PublicEvolving
+public final class UnionReadBoundaryPlanner {
+
+    private UnionReadBoundaryPlanner() {}
+
+    /**
+     * Captures table-wide boundaries, suitable for sharing across scans with different predicates.
+     *
+     * @param admin caller-owned Admin client
+     * @param tablePath table whose boundaries should be captured
+     * @return captured source metadata and log boundaries; engines own task serialization
+     * @throws InterruptedException if interrupted while waiting for metadata
+     * @throws Exception if metadata capture or boundary validation fails
+     */
+    public static UnionReadContext captureReadContext(Admin admin, TablePath tablePath)
+            throws Exception {
+        return captureReadContext(admin, tablePath, new PartitionSpec(Collections.emptyMap()));
+    }
+
+    /**
+     * Captures boundaries within an exact, possibly partial partition spec before querying offsets.
+     *
+     * <p>The scope is copied on entry. Partitions removed or recreated during capture are excluded
+     * by ID; other lookup failures propagate. Later schema additions do not replace the frozen
+     * schema. A scoped context must not be reused as a table-wide boundary.
+     *
+     * @param admin caller-owned Admin client
+     * @param tablePath table whose boundaries should be captured
+     * @param scope partition keys and metadata-encoded values; an empty spec selects the whole
+     *     table
+     * @return captured source metadata and log boundaries; engines own task serialization
+     * @throws InterruptedException if interrupted while waiting for metadata
+     * @throws Exception if the scope is invalid, metadata capture fails or boundaries are
+     *     inconsistent
+     */
+    public static UnionReadContext captureReadContext(
+            Admin admin, TablePath tablePath, PartitionSpec scope) throws Exception {
+        checkNotNull(admin, "admin");
+        checkNotNull(tablePath, "tablePath");
+        checkNotNull(scope, "scope");
+        PartitionSpec capturedScope = new PartitionSpec(new HashMap<>(scope.getSpecMap()));
+        TableInfo table = admin.getTableInfo(tablePath).get();
+        checkArgument(
+                table.getPartitionKeys().containsAll(capturedScope.getSpecMap().keySet())
+                        && !capturedScope.getSpecMap().containsValue(null),
+                "Scope must reference partition keys with non-null encoded values.");
+        LakeSnapshot snapshot = readableSnapshot(admin, table);
+        List<PartitionInfo> selected =
+                table.isPartitioned()
+                        ? admin.listPartitionInfos(tablePath, capturedScope).get()
+                        : Collections.singletonList(null);
+        List<PartitionCapture> captures = new ArrayList<>();
+        for (PartitionInfo partition : selected) {
+            captures.add(readOffsets(admin, table, partition));
+        }
+        recheckPartitions(admin, table, capturedScope, captures);
+        validateIdentity(table, admin.getTableInfo(tablePath).get());
+        Map<TableBucket, Long> seams =
+                snapshot == null ? Collections.emptyMap() : snapshot.getTableBucketsOffset();
+        if (snapshot != null) {
+            checkArgument(snapshot.getSnapshotId() >= 0, "Invalid lake snapshot ID.");
+            for (Map.Entry<TableBucket, Long> seam : seams.entrySet()) {
+                checkArgument(
+                        seam.getKey().getTableId() == table.getTableId()
+                                && seam.getValue() != null
+                                && seam.getValue() >= 0,
+                        "Snapshot contains an invalid bucket seam: %s",
+                        seam.getKey());
+            }
+        }
+        List<UnionReadContext.LogRange> ranges = new ArrayList<>();
+        for (PartitionCapture capture : captures) {
+            int count =
+                    PartitionInfo.bucketCountOrDefault(capture.partition, table.getNumBuckets());
+            checkArgument(
+                    capture.earliest.size() == count && capture.stops.size() == count,
+                    "Offset maps must cover exactly the captured buckets.");
+            Map<String, String> values =
+                    capture.partition == null
+                            ? Collections.emptyMap()
+                            : capture.partition.getPartitionSpec().getSpecMap();
+            for (int bucketId = 0; bucketId < count; bucketId++) {
+                TableBucket bucket =
+                        new TableBucket(
+                                table.getTableId(),
+                                capture.partition == null
+                                        ? null
+                                        : capture.partition.getPartitionId(),
+                                bucketId);
+                Long begin = capture.earliest.get(bucket);
+                Long end = capture.stops.get(bucket);
+                Long seam = seams.get(bucket);
+                checkArgument(begin != null && end != null, "Missing log offsets for %s", bucket);
+                long start = seam == null ? begin : seam;
+                checkArgument(
+                        start >= 0 && begin >= 0 && end >= start && end >= begin,
+                        "Invalid concrete log offsets for %s",
+                        bucket);
+                ranges.add(new UnionReadContext.LogRange(bucket, values, count, seam, begin, end));
+            }
+        }
+        return new UnionReadContext(table, snapshot, capturedScope, ranges);
+    }
+
+    @Nullable
+    private static LakeSnapshot readableSnapshot(Admin admin, TableInfo table)
+            throws InterruptedException, ExecutionException {
+        if (!table.getTableConfig().isDataLakeEnabled()) {
+            return null;
+        }
+        try {
+            return admin.getReadableLakeSnapshot(table.getTablePath()).get();
+        } catch (ExecutionException e) {
+            if (ExceptionUtils.stripExecutionException(e)
+                    instanceof LakeTableSnapshotNotExistException) {
+                return null;
+            }
+            throw e;
+        }
+    }
+
+    private static PartitionCapture readOffsets(
+            Admin admin, TableInfo table, @Nullable PartitionInfo partition)
+            throws InterruptedException {
+        PartitionCapture capture = new PartitionCapture(partition);
+        try {
+            int count = partition == null ? table.getNumBuckets() : partition.getBucketCount();
+            checkArgument(count > 0, "Invalid bucket count: %s", count);
+            List<Integer> buckets = IntStream.range(0, count).boxed().collect(Collectors.toList());
+            capture.putOffsets(
+                    table.getTableId(),
+                    offsets(
+                            admin,
+                            table.getTablePath(),
+                            partition,
+                            buckets,
+                            new OffsetSpec.EarliestSpec()),
+                    capture.earliest);
+            capture.putOffsets(
+                    table.getTableId(),
+                    offsets(
+                            admin,
+                            table.getTablePath(),
+                            partition,
+                            buckets,
+                            new OffsetSpec.LatestSpec()),
+                    capture.stops);
+        } catch (ExecutionException | RuntimeException e) {
+            // A failed lookup may be caused by deletion. Only the ID recheck can prove that.
+            capture.error = e;
+        }
+        return capture;
+    }
+
+    private static Map<Integer, Long> offsets(
+            Admin admin,
+            TablePath path,
+            @Nullable PartitionInfo partition,
+            List<Integer> buckets,
+            OffsetSpec spec)
+            throws InterruptedException, ExecutionException {
+        return (partition == null
+                        ? admin.listOffsets(path, buckets, spec)
+                        : admin.listOffsets(path, partition.getPartitionName(), buckets, spec))
+                .all()
+                .get();
+    }
+
+    private static void recheckPartitions(
+            Admin admin, TableInfo table, PartitionSpec scope, List<PartitionCapture> captures)
+            throws Exception {
+        Exception recheckError = null;
+        if (table.isPartitioned()) {
+            try {
+                Set<Long> liveIds =
+                        admin.listPartitionInfos(table.getTablePath(), scope).get().stream()
+                                .map(PartitionInfo::getPartitionId)
+                                .collect(Collectors.toSet());
+                // Successful name-based lookups can also cross a same-name recreation.
+                captures.removeIf(capture -> !liveIds.contains(capture.partition.getPartitionId()));
+            } catch (ExecutionException | RuntimeException e) {
+                recheckError = e;
+            }
+        }
+        for (PartitionCapture capture : captures) {
+            if (capture.error != null) {
+                throw capture.error;
+            }
+        }
+        if (recheckError != null) {
+            throw recheckError;
+        }
+    }
+
+    private static void validateIdentity(TableInfo captured, TableInfo live) {
+        checkArgument(
+                captured.getTableId() == live.getTableId()
+                        && captured.getTablePath().equals(live.getTablePath()),
+                "Table identity changed during boundary capture.");
+        if (captured.getTableConfig().isDataLakeEnabled()) {
+            checkArgument(
+                    live.getTableConfig().isDataLakeEnabled()
+                            && captured.getLakeTablePath().equals(live.getLakeTablePath())
+                            && captured.getTableConfig()
+                                    .getDataLakeFormat()
+                                    .equals(live.getTableConfig().getDataLakeFormat()),
+                    "Lake mapping changed during boundary capture.");
+        }
+    }
+
+    private static final class PartitionCapture {
+        @Nullable private final PartitionInfo partition;
+        private final Map<TableBucket, Long> earliest = new HashMap<>();
+        private final Map<TableBucket, Long> stops = new HashMap<>();
+        @Nullable private Exception error;
+
+        private PartitionCapture(@Nullable PartitionInfo partition) {
+            this.partition = partition;
+        }
+
+        private void putOffsets(
+                long tableId, Map<Integer, Long> offsets, Map<TableBucket, Long> target) {
+            offsets.forEach(
+                    (bucket, offset) ->
+                            target.put(
+                                    new TableBucket(
+                                            tableId,
+                                            partition == null ? null : partition.getPartitionId(),
+                                            bucket),
+                                    offset));
+        }
+    }
+}