diff --git a/.github/workflows/client-integration.yml b/.github/workflows/client-integration.yml index 062ae7ec5dd..658e2e440a8 100644 --- a/.github/workflows/client-integration.yml +++ b/.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] diff --git a/.github/workflows/rust-build-and-test.yml b/.github/workflows/rust-build-and-test.yml index f6021bef054..be5779f3145 100644 --- a/.github/workflows/rust-build-and-test.yml +++ b/.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' }} diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlanner.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlanner.java new file mode 100644 index 00000000000..8df22732832 --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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 selected = + table.isPartitioned() + ? admin.listPartitionInfos(tablePath, capturedScope).get() + : Collections.singletonList(null); + List 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 seams = + snapshot == null ? Collections.emptyMap() : snapshot.getTableBucketsOffset(); + if (snapshot != null) { + checkArgument(snapshot.getSnapshotId() >= 0, "Invalid lake snapshot ID."); + for (Map.Entry 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 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 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 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 offsets( + Admin admin, + TablePath path, + @Nullable PartitionInfo partition, + List 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 captures) + throws Exception { + Exception recheckError = null; + if (table.isPartitioned()) { + try { + Set 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 earliest = new HashMap<>(); + private final Map stops = new HashMap<>(); + @Nullable private Exception error; + + private PartitionCapture(@Nullable PartitionInfo partition) { + this.partition = partition; + } + + private void putOffsets( + long tableId, Map offsets, Map target) { + offsets.forEach( + (bucket, offset) -> + target.put( + new TableBucket( + tableId, + partition == null ? null : partition.getPartitionId(), + bucket), + offset)); + } + } +} diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadContext.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadContext.java new file mode 100644 index 00000000000..b20dd0c69ca --- /dev/null +++ b/fluss-client/src/main/java/org/apache/fluss/client/metadata/UnionReadContext.java @@ -0,0 +1,213 @@ +/* + * 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.metadata.DataLakeFormat; +import org.apache.fluss.metadata.MergeEngineType; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; + +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.Optional; +import java.util.OptionalLong; + +/** + * Captured source boundaries, independent of an engine's task layout or transport format. + * + *

The context fixes source metadata and log offsets but provides no global transaction snapshot + * or retention lease. Engines must preserve the captured schema, lake snapshot, partition scope and + * log ranges when constructing distributed tasks. Credentials and query configuration are supplied + * separately. The selected lake snapshot can include partitions absent from the log ranges. + * + * @since 1.1 + */ +@PublicEvolving +public final class UnionReadContext { + private final TablePath tablePath; + private final long tableId; + private final long tableModifiedTime; + private final int schemaId; + private final Schema schema; + private final List partitionKeys; + private final List bucketKeys; + private final int numBuckets; + private final Map partitionFilter; + @Nullable private final TablePath lakeTablePath; + private final Optional lakeFormat; + private final Optional mergeEngine; + private final OptionalLong lakeSnapshotId; + private final List logRanges; + + UnionReadContext( + TableInfo table, + @Nullable LakeSnapshot snapshot, + PartitionSpec scope, + List ranges) { + tablePath = table.getTablePath(); + tableId = table.getTableId(); + tableModifiedTime = table.getModifiedTime(); + schemaId = table.getSchemaId(); + schema = table.getSchema(); + partitionKeys = Collections.unmodifiableList(new ArrayList<>(table.getPartitionKeys())); + bucketKeys = Collections.unmodifiableList(new ArrayList<>(table.getBucketKeys())); + numBuckets = table.getNumBuckets(); + partitionFilter = Collections.unmodifiableMap(new HashMap<>(scope.getSpecMap())); + boolean lakeEnabled = table.getTableConfig().isDataLakeEnabled(); + lakeTablePath = lakeEnabled ? table.getLakeTablePath() : null; + lakeFormat = lakeEnabled ? table.getTableConfig().getDataLakeFormat() : Optional.empty(); + mergeEngine = table.getTableConfig().getMergeEngineType(); + lakeSnapshotId = + snapshot == null ? OptionalLong.empty() : OptionalLong.of(snapshot.getSnapshotId()); + logRanges = Collections.unmodifiableList(new ArrayList<>(ranges)); + } + + public TablePath getTablePath() { + return tablePath; + } + + public long getTableId() { + return tableId; + } + + /** Source metadata modification time, not a query snapshot timestamp. */ + public long getTableModifiedTime() { + return tableModifiedTime; + } + + public int getSchemaId() { + return schemaId; + } + + /** Frozen, unprojected schema. */ + public Schema getSchema() { + return schema; + } + + public List getPartitionKeys() { + return partitionKeys; + } + + public List getBucketKeys() { + return bucketKeys; + } + + /** Table default; each log range carries its partition's actual bucket count. */ + public int getNumBuckets() { + return numBuckets; + } + + /** Exact scope for both lake and log inputs; empty means table-wide. */ + public Map getPartitionFilter() { + return partitionFilter; + } + + public Optional getLakeTablePath() { + return Optional.ofNullable(lakeTablePath); + } + + public Optional getLakeFormat() { + return lakeFormat; + } + + public Optional getMergeEngine() { + return mergeEngine; + } + + /** Readable lake snapshot, absent before initial tiering or for a non-lake table. */ + public OptionalLong getLakeSnapshotId() { + return lakeSnapshotId; + } + + /** Live log boundaries, not a complete inventory of lake partitions or engine tasks. */ + public List getLogRanges() { + return logRanges; + } + + /** + * One physical bucket's half-open log interval. Offsets are comparable only within that bucket. + * A missing lake seam means reading starts at the captured earliest retained offset; for PK + * tables this covers only keys reconstructible from retained logs. + */ + @PublicEvolving + public static final class LogRange { + private final TableBucket tableBucket; + private final Map partitionValues; + private final int bucketCount; + private final OptionalLong lakeSnapshotOffset; + private final long earliestOffset; + private final long stopOffset; + + LogRange( + TableBucket bucket, + Map partitionValues, + int bucketCount, + @Nullable Long seam, + long earliest, + long stop) { + this.tableBucket = bucket; + this.partitionValues = Collections.unmodifiableMap(new HashMap<>(partitionValues)); + this.bucketCount = bucketCount; + this.lakeSnapshotOffset = seam == null ? OptionalLong.empty() : OptionalLong.of(seam); + this.earliestOffset = earliest; + this.stopOffset = stop; + } + + public TableBucket getTableBucket() { + return tableBucket; + } + + /** + * Business partition values in Fluss metadata encoding; empty for an unpartitioned table. + */ + public Map getPartitionValues() { + return partitionValues; + } + + public int getBucketCount() { + return bucketCount; + } + + /** A present zero is a lake seam, not an absent baseline. */ + public OptionalLong getLakeSnapshotOffset() { + return lakeSnapshotOffset; + } + + public long getStartOffset() { + return lakeSnapshotOffset.orElse(earliestOffset); + } + + public long getStopOffset() { + return stopOffset; + } + + /** Retention observation, not a lease; execution must detect later data loss. */ + public long getEarliestOffset() { + return earliestOffset; + } + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerITCase.java new file mode 100644 index 00000000000..3718707ca02 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerITCase.java @@ -0,0 +1,90 @@ +/* + * 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.client.admin.ClientToServerITCaseBase; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Exercises the public capture entry against the ordinary Java client test cluster. */ +class UnionReadBoundaryPlannerITCase extends ClientToServerITCaseBase { + + @Test + void testCapturePartitionScope() throws Exception { + TablePath path = TablePath.of("union_read_boundary", "orders"); + long tableId = + createTable( + path, + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("region", DataTypes.STRING()) + .primaryKey("id", "region") + .build()) + .distributedBy(2, "id") + .partitionedBy("region") + .build(), + false); + try { + for (String region : Arrays.asList("US", "EU")) { + admin.createPartition( + path, + new PartitionSpec(Collections.singletonMap("region", region)), + false) + .get(); + } + for (PartitionInfo partition : admin.listPartitionInfos(path).get()) { + for (int bucket = 0; bucket < partition.getBucketCount(); bucket++) { + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady( + new TableBucket(tableId, partition.getPartitionId(), bucket)); + } + } + UnionReadContext all = UnionReadBoundaryPlanner.captureReadContext(admin, path); + UnionReadContext scoped = + UnionReadBoundaryPlanner.captureReadContext( + admin, + path, + new PartitionSpec(Collections.singletonMap("region", "US"))); + assertThat(all.getLogRanges()).hasSize(4); + assertThat(scoped.getLakeFormat()).isEmpty(); + assertThat(scoped.getTableId()).isEqualTo(tableId); + assertThat(scoped.getLogRanges()).hasSize(2); + assertThat(scoped.getPartitionFilter().get("region")).isEqualTo("US"); + for (UnionReadContext.LogRange range : scoped.getLogRanges()) { + assertThat(range.getPartitionValues().get("region")).isEqualTo("US"); + assertThat(range.getStartOffset()).isEqualTo(range.getEarliestOffset()); + assertThat(range.getStopOffset()).isZero(); + } + } finally { + admin.dropTable(path, false).get(); + } + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerTest.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerTest.java new file mode 100644 index 00000000000..8e975c08585 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/metadata/UnionReadBoundaryPlannerTest.java @@ -0,0 +1,300 @@ +/* + * 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.client.admin.Admin; +import org.apache.fluss.client.admin.ListOffsetsResult; +import org.apache.fluss.client.admin.OffsetSpec; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.LakeTableSnapshotNotExistException; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.ResolvedPartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isA; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Tests capture sequencing and metadata races, through the structured boundary API. */ +class UnionReadBoundaryPlannerTest { + + @Test + void testScopedCaptureUsesSnapshotBeforeOffsetsAndFrozenSchema() throws Exception { + Admin admin = mock(Admin.class); + TableInfo table = table(true); + TableInfo evolved = table(true, 7, 3); + when(admin.getTableInfo(table.getTablePath())) + .thenReturn( + CompletableFuture.completedFuture(table), + CompletableFuture.completedFuture(table), + CompletableFuture.completedFuture(evolved)); + PartitionInfo partition = partition(9, "US", 1); + PartitionSpec scope = new PartitionSpec(Collections.singletonMap("region", "US")); + when(admin.listPartitionInfos(table.getTablePath(), scope)) + .thenReturn( + CompletableFuture.completedFuture(Collections.singletonList(partition))); + CompletableFuture snapshot = new CompletableFuture<>(); + when(admin.getReadableLakeSnapshot(table.getTablePath())).thenReturn(snapshot); + when(admin.listOffsets( + eq(table.getTablePath()), + eq(partition.getPartitionName()), + eq(Collections.singletonList(0)), + isA(OffsetSpec.EarliestSpec.class))) + .thenReturn(offsets(1, 8)); + when(admin.listOffsets( + eq(table.getTablePath()), + eq(partition.getPartitionName()), + eq(Collections.singletonList(0)), + isA(OffsetSpec.LatestSpec.class))) + .thenReturn(offsets(1, 20)); + + InOrder order = inOrder(admin); + Thread.currentThread().interrupt(); + try { + assertThatThrownBy( + () -> + UnionReadBoundaryPlanner.captureReadContext( + admin, table.getTablePath(), scope)) + .isInstanceOf(InterruptedException.class); + } finally { + Thread.interrupted(); + } + order.verify(admin).getTableInfo(table.getTablePath()); + order.verify(admin).getReadableLakeSnapshot(table.getTablePath()); + + snapshot.complete( + new LakeSnapshot(42, Collections.singletonMap(new TableBucket(7, 9L, 0), 12L))); + UnionReadContext context = + UnionReadBoundaryPlanner.captureReadContext(admin, table.getTablePath(), scope); + assertThat(context.getSchemaId()).isEqualTo(2); + assertThat(context.getPartitionFilter().get("region")).isEqualTo("US"); + assertThat(context.getLogRanges()).hasSize(1); + assertThat(context.getLogRanges().get(0).getStartOffset()).isEqualTo(12); + assertThat(context.getLogRanges().get(0).getStopOffset()).isEqualTo(20); + table.getProperties().setString("table.datalake.table-name", "changed"); + assertThat(context.getPartitionFilter()).containsEntry("region", "US"); + assertThat(context.getLakeTablePath()).contains(TablePath.of("lake_db", "orders_lake")); + order.verify(admin).getTableInfo(table.getTablePath()); + order.verify(admin).getReadableLakeSnapshot(table.getTablePath()); + order.verify(admin).listPartitionInfos(table.getTablePath(), scope); + order.verify(admin) + .listOffsets( + eq(table.getTablePath()), + eq(partition.getPartitionName()), + anyList(), + isA(OffsetSpec.EarliestSpec.class)); + order.verify(admin) + .listOffsets( + eq(table.getTablePath()), + eq(partition.getPartitionName()), + anyList(), + isA(OffsetSpec.LatestSpec.class)); + order.verify(admin).listPartitionInfos(table.getTablePath(), scope); + order.verify(admin).getTableInfo(table.getTablePath()); + + assertThatThrownBy( + () -> + UnionReadBoundaryPlanner.captureReadContext( + admin, + table.getTablePath(), + new PartitionSpec(Collections.singletonMap("amount", "1")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("partition keys"); + } + + @Test + void testNoLakeBaselineAndRecreatedTable() throws Exception { + for (boolean lakeEnabled : new boolean[] {false, true}) { + Admin admin = mock(Admin.class); + TableInfo table = table(false); + table.getProperties() + .setString("table.datalake.enabled", Boolean.toString(lakeEnabled)); + when(admin.getTableInfo(table.getTablePath())) + .thenReturn(CompletableFuture.completedFuture(table)); + when(admin.getReadableLakeSnapshot(table.getTablePath())) + .thenReturn(failed(new LakeTableSnapshotNotExistException("not tiered yet"))); + when(admin.listOffsets( + eq(table.getTablePath()), + anyList(), + isA(OffsetSpec.EarliestSpec.class))) + .thenReturn(offsets(2, 8)); + when(admin.listOffsets( + eq(table.getTablePath()), anyList(), isA(OffsetSpec.LatestSpec.class))) + .thenReturn(offsets(2, 20)); + UnionReadContext context = + UnionReadBoundaryPlanner.captureReadContext(admin, table.getTablePath()); + assertThat(context.getLogRanges().get(0).getStartOffset()).isEqualTo(8); + assertThat(context.getLogRanges().get(0).getLakeSnapshotOffset()).isEmpty(); + if (lakeEnabled) { + assertThat(context.getLakeSnapshotId()).isEmpty(); + } else { + assertThat(context.getLakeFormat()).isEmpty(); + verify(admin, never()).getReadableLakeSnapshot(any()); + } + TableInfo recreated = table(false, 8, 2); + when(admin.getTableInfo(table.getTablePath())) + .thenReturn( + CompletableFuture.completedFuture(table), + CompletableFuture.completedFuture(recreated)); + assertThatThrownBy( + () -> + UnionReadBoundaryPlanner.captureReadContext( + admin, table.getTablePath())) + .hasMessage("Table identity changed during boundary capture."); + } + } + + @Test + void testPartitionRecheckSkipsOnlyConfirmedRemoval() throws Exception { + // Both successful and failed lookups may have crossed a same-name recreation. + for (boolean lookupFails : new boolean[] {false, true}) { + for (int recheck : new int[] {0, 1, 2}) { + Admin admin = mock(Admin.class); + TableInfo table = table(true); + table.getProperties().setString("table.datalake.enabled", "false"); + when(admin.getTableInfo(table.getTablePath())) + .thenReturn(CompletableFuture.completedFuture(table)); + PartitionInfo original = partition(9, "US", 1); + CompletableFuture> live = + recheck == 2 + ? failed(new IllegalStateException("recheck failed")) + : CompletableFuture.completedFuture( + Collections.singletonList( + recheck == 0 ? partition(10, "US", 1) : original)); + when(admin.listPartitionInfos(eq(table.getTablePath()), any(PartitionSpec.class))) + .thenReturn( + CompletableFuture.completedFuture( + Collections.singletonList(original)), + live); + when(admin.listOffsets( + eq(table.getTablePath()), + eq(original.getPartitionName()), + anyList(), + isA(OffsetSpec.EarliestSpec.class))) + .thenAnswer( + call -> { + if (lookupFails) { + throw new IllegalStateException("lookup failed"); + } + return offsets(1, 8); + }); + when(admin.listOffsets( + eq(table.getTablePath()), + eq(original.getPartitionName()), + anyList(), + isA(OffsetSpec.LatestSpec.class))) + .thenReturn(offsets(1, 20)); + if (recheck == 0 || (!lookupFails && recheck == 1)) { + UnionReadContext context = + UnionReadBoundaryPlanner.captureReadContext( + admin, table.getTablePath()); + assertThat(context.getLogRanges()).hasSize(recheck == 0 ? 0 : 1); + } else { + assertThatThrownBy( + () -> + UnionReadBoundaryPlanner.captureReadContext( + admin, table.getTablePath())) + .hasStackTraceContaining( + lookupFails ? "lookup failed" : "recheck failed"); + } + } + } + } + + static TableInfo table(boolean partitioned) { + return table(partitioned, 7, 2); + } + + static TableInfo table(boolean partitioned, long tableId, int schemaId) { + Map properties = new HashMap<>(); + properties.put("table.datalake.enabled", "true"); + properties.put("table.datalake.format", "paimon"); + properties.put("table.datalake.database-name", "lake_db"); + properties.put("table.datalake.table-name", "orders_lake"); + properties.put("table.datalake.paimon.s3.secret-key", "SECRET"); + properties.put("table.datalake.paimon.warehouse", "s3://private/warehouse"); + Schema.Builder schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("region", DataTypes.STRING()) + .column("amount", DataTypes.BIGINT()); + if (schemaId > 2) { + schema.column("added", DataTypes.STRING()); + } + if (partitioned) { + schema.primaryKey("id", "region"); + } + return new TableInfo( + TablePath.of("fluss", "orders"), + tableId, + schemaId, + schema.build(), + partitioned ? new ArrayList<>(Collections.singletonList("id")) : new ArrayList<>(), + partitioned + ? new ArrayList<>(Collections.singletonList("region")) + : new ArrayList<>(), + 2, + Configuration.fromMap(properties), + new Configuration(), + null, + null, + 0, + 0); + } + + static PartitionInfo partition(long id, String value, int bucketCount) { + return new PartitionInfo( + id, ResolvedPartitionSpec.fromPartitionValue("region", value), null, bucketCount); + } + + private static ListOffsetsResult offsets(int buckets, long offset) { + Map> offsets = new HashMap<>(); + for (int bucket = 0; bucket < buckets; bucket++) { + offsets.put(bucket, CompletableFuture.completedFuture(offset)); + } + return new ListOffsetsResult(offsets); + } + + private static CompletableFuture failed(Throwable error) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(error); + return future; + } +} diff --git a/fluss-rust/Cargo.lock b/fluss-rust/Cargo.lock index 7e716e90e56..a353cd931c3 100644 --- a/fluss-rust/Cargo.lock +++ b/fluss-rust/Cargo.lock @@ -1103,6 +1103,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "fluss-lake" +version = "1.0.0" +dependencies = [ + "arrow", + "fluss-rs", + "fluss-test-cluster", + "futures", + "serde_json", + "thiserror 1.0.69", + "tokio", +] + [[package]] name = "fluss-rs" version = "1.0.0" diff --git a/fluss-rust/Cargo.toml b/fluss-rust/Cargo.toml index 3a82f8fca51..8e3f297e3cb 100644 --- a/fluss-rust/Cargo.toml +++ b/fluss-rust/Cargo.toml @@ -28,7 +28,7 @@ keywords = ["fluss", "streaming-storage", "datalake"] [workspace] resolver = "2" -members = ["crates/fluss", "crates/fluss/gen", "crates/fluss-test-cluster", "crates/examples", "bindings/python", "bindings/cpp", "bindings/elixir/native/fluss_nif"] +members = ["crates/fluss", "crates/fluss/gen", "crates/fluss-lake", "crates/fluss-test-cluster", "crates/examples", "bindings/python", "bindings/cpp", "bindings/elixir/native/fluss_nif"] [workspace.dependencies] fluss = { package = "fluss-rs", version = "1.0.0", path = "crates/fluss", features = ["storage-all"] } diff --git a/fluss-rust/DEPENDENCIES.rust.tsv b/fluss-rust/DEPENDENCIES.rust.tsv index c8f4dfdac8b..1c0de6191e7 100644 --- a/fluss-rust/DEPENDENCIES.rust.tsv +++ b/fluss-rust/DEPENDENCIES.rust.tsv @@ -94,6 +94,7 @@ fixedbitset@0.5.7 X X flatbuffers@25.12.19 X fluss-cpp@1.0.0 X fluss-examples@1.0.0 X +fluss-lake@1.0.0 X fluss-rs@1.0.0 X fluss-test-cluster@1.0.0 X fluss_nif@1.0.0 X diff --git a/fluss-rust/crates/fluss-lake/Cargo.toml b/fluss-rust/crates/fluss-lake/Cargo.toml new file mode 100644 index 00000000000..014565f3228 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/Cargo.toml @@ -0,0 +1,44 @@ +# 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] +name = "fluss-lake" +version = { workspace = true } +edition = { workspace = true } +# Paimon 0.3 requires Rust 1.91; this package MSRV applies to all features. +# Keep in sync with the `lake-msrv` CI job. +rust-version = "1.91" +license = { workspace = true } +authors = { workspace = true } +description = "Engine-neutral bounded lake and log read kernel for Apache Fluss" +homepage = { workspace = true } +repository = { workspace = true } +keywords = { workspace = true } +categories = { workspace = true } + +[features] +integration_tests = ["fluss/integration_tests"] + +[dependencies] +fluss = { package = "fluss-rs", path = "../fluss" } +futures = "0.3" +thiserror = "1.0" + +[dev-dependencies] +serde_json = { workspace = true } +arrow = { workspace = true } +fluss-test-cluster = { path = "../fluss-test-cluster" } +tokio = { workspace = true } diff --git a/fluss-rust/crates/fluss-lake/DEPENDENCIES.rust.tsv b/fluss-rust/crates/fluss-lake/DEPENDENCIES.rust.tsv new file mode 100644 index 00000000000..4152cb232ea --- /dev/null +++ b/fluss-rust/crates/fluss-lake/DEPENDENCIES.rust.tsv @@ -0,0 +1,359 @@ +crate Apache-2.0 Apache-2.0 WITH LLVM-exception BSD-2-Clause BSD-3-Clause BSL-1.0 CC0-1.0 CDLA-Permissive-2.0 ISC LGPL-2.1-or-later MIT Unicode-3.0 Unlicense Zlib +ahash@0.8.12 X X +aho-corasick@1.1.4 X X +android_system_properties@0.1.5 X X +anstream@1.0.0 X X +anstyle@1.0.14 X X +anstyle-parse@1.0.0 X X +anstyle-query@1.1.5 X X +anstyle-wincon@3.0.11 X X +anyhow@1.0.102 X X +arrow@59.0.0 X +arrow-arith@59.0.0 X +arrow-array@59.0.0 X X +arrow-buffer@59.0.0 X +arrow-cast@59.0.0 X +arrow-csv@59.0.0 X +arrow-data@59.0.0 X +arrow-ipc@59.0.0 X +arrow-json@59.0.0 X +arrow-ord@59.0.0 X +arrow-row@59.0.0 X +arrow-schema@59.0.0 X +arrow-select@59.0.0 X +arrow-string@59.0.0 X +astral-tokio-tar@0.6.0 X X +async-stream@0.3.6 X +async-stream-impl@0.3.6 X +async-trait@0.1.89 X X +atoi@2.0.0 X +atomic-waker@1.1.2 X X +autocfg@1.5.0 X X +axum@0.8.8 X +axum-core@0.5.6 X +backon@1.6.0 X +base64@0.21.7 X X +base64@0.22.1 X X +bigdecimal@0.4.10 X X +bitflags@2.11.0 X X +bitvec@1.0.1 X +block-buffer@0.10.4 X X +bollard@0.20.2 X +bollard-buildkit-proto@0.7.0 X +bollard-stubs@1.52.1-rc.29.1.3 X +bumpalo@3.20.2 X X +byteorder@1.5.0 X X +bytes@1.11.1 X +cc@1.2.59 X X +cfg-if@1.0.4 X X +chrono@0.4.44 X X +clap@4.6.0 X X +clap_builder@4.6.0 X X +clap_derive@4.6.0 X X +clap_lex@1.1.0 X X +colorchoice@1.0.5 X X +const-oid@0.9.6 X X +const-random@0.1.18 X X +const-random-macro@0.1.16 X X +core-foundation@0.10.1 X X +core-foundation-sys@0.8.7 X X +cpufeatures@0.2.17 X X +crc32c@0.6.8 X X +crossbeam-utils@0.8.21 X X +crunchy@0.2.4 X +crypto-common@0.1.7 X X +csv@1.4.0 X X +csv-core@0.1.13 X X +darling@0.23.0 X +darling_core@0.23.0 X +darling_macro@0.23.0 X +dashmap@6.1.0 X +delegate@0.13.5 X X +deranged@0.5.8 X X +digest@0.10.7 X X +displaydoc@0.2.5 X X +dlv-list@0.5.2 X X +docker_credential@1.3.2 X X +either@1.15.0 X X +equivalent@1.0.2 X X +errno@0.3.14 X X +etcetera@0.11.0 X X +fastrand@2.4.0 X X +ferroid@0.8.9 X X +filetime@0.2.27 X X +find-msvc-tools@0.1.9 X X +flatbuffers@25.12.19 X +fluss-lake@1.0.0 X +fluss-rs@1.0.0 X +fluss-test-cluster@1.0.0 X +fnv@1.0.7 X X +foldhash@0.2.0 X +form_urlencoded@1.2.2 X X +funty@2.0.0 X +futures@0.3.32 X X +futures-channel@0.3.32 X X +futures-core@0.3.32 X X +futures-executor@0.3.32 X X +futures-io@0.3.32 X X +futures-macro@0.3.32 X X +futures-sink@0.3.32 X X +futures-task@0.3.32 X X +futures-util@0.3.32 X X +generic-array@0.14.7 X +getrandom@0.2.17 X X +getrandom@0.3.4 X X +getrandom@0.4.2 X X +gloo-timers@0.3.0 X X +h2@0.4.13 X +half@2.7.1 X X +hashbrown@0.14.5 X X +hashbrown@0.16.1 X X +hashbrown@0.17.1 X X +heck@0.5.0 X X +hex@0.4.3 X X +hmac@0.12.1 X X +home@0.5.12 X X +http@1.4.0 X X +http-body@1.0.1 X +http-body-util@0.1.3 X +httparse@1.10.1 X X +httpdate@1.0.3 X X +hyper@1.9.0 X +hyper-named-pipe@0.1.0 X +hyper-rustls@0.27.7 X X X +hyper-timeout@0.5.2 X X +hyper-util@0.1.20 X +hyperlocal@0.9.1 X +iana-time-zone@0.1.65 X X +iana-time-zone-haiku@0.1.2 X X +icu_collections@2.2.0 X +icu_locale_core@2.2.0 X +icu_normalizer@2.2.0 X +icu_normalizer_data@2.2.0 X +icu_properties@2.2.0 X +icu_properties_data@2.2.0 X +icu_provider@2.2.0 X +ident_case@1.0.1 X X +idna@1.1.0 X X +idna_adapter@1.2.1 X X +indexmap@2.13.1 X X +ipnet@2.12.0 X X +iri-string@0.7.12 X X +is_terminal_polyfill@1.70.2 X X +itertools@0.14.0 X X +itoa@1.0.18 X X +jiff@0.2.23 X X +jiff-tzdb@0.1.6 X X +jiff-tzdb-platform@0.1.3 X X +jobserver@0.1.34 X X +js-sys@0.3.94 X X +lexical-core@1.0.6 X X +lexical-parse-float@1.0.6 X X +lexical-parse-integer@1.0.6 X X +lexical-util@1.0.7 X X +lexical-write-float@1.0.6 X X +lexical-write-integer@1.0.6 X X +libc@0.2.184 X X +libm@0.2.16 X +libredox@0.1.15 X +linked-hash-map@0.5.6 X X +linux-raw-sys@0.12.1 X X X +litemap@0.8.2 X +lock_api@0.4.14 X X +log@0.4.29 X X +lz4_flex@0.13.1 X +matchit@0.8.4 X X +md-5@0.10.6 X X +memchr@2.8.0 X X +metrics@0.24.5 X +mime@0.3.17 X X +mio@1.2.0 X +num@0.4.3 X X +num-bigint@0.4.6 X X +num-complex@0.4.6 X X +num-conv@0.2.1 X X +num-integer@0.1.46 X X +num-iter@0.1.45 X X +num-rational@0.4.2 X X +num-traits@0.2.19 X X +once_cell@1.21.4 X X +once_cell_polyfill@1.70.2 X X +opendal@0.55.0 X +openssl-probe@0.2.1 X X +ordered-float@5.3.0 X +ordered-multimap@0.7.3 X +parking_lot@0.12.5 X X +parking_lot_core@0.9.12 X X +parse-display@0.10.0 X X +parse-display@0.9.1 X X +parse-display-derive@0.10.0 X X +parse-display-derive@0.9.1 X X +percent-encoding@2.3.2 X X +pin-project@1.1.11 X X +pin-project-internal@1.1.11 X X +pin-project-lite@0.2.17 X X +pkg-config@0.3.32 X X +plain@0.2.3 X X +portable-atomic@1.13.1 X X +portable-atomic-util@0.2.6 X X +potential_utf@0.1.5 X +powerfmt@0.2.0 X X +ppv-lite86@0.2.21 X X +proc-macro2@1.0.106 X X +prost@0.14.3 X +prost-derive@0.14.3 X +prost-types@0.14.3 X +quick-xml@0.37.5 X +quick-xml@0.38.4 X +quote@1.0.45 X X +r-efi@5.3.0 X X X +r-efi@6.0.0 X X X +radium@0.7.0 X +rand@0.8.5 X X +rand@0.9.3 X X +rand_chacha@0.3.1 X X +rand_chacha@0.9.0 X X +rand_core@0.6.4 X X +rand_core@0.9.5 X X +rapidhash@4.4.1 X X +redox_syscall@0.5.18 X +redox_syscall@0.7.3 X +regex@1.12.3 X X +regex-automata@0.4.14 X X +regex-syntax@0.8.10 X X +reqsign@0.16.5 X +reqwest@0.12.28 X X +ring@0.17.14 X X +rust-ini@0.21.3 X +rustc-hash@2.1.2 X X +rustc_version@0.4.1 X X +rustix@1.1.4 X X X +rustls@0.23.37 X X X +rustls-native-certs@0.8.3 X X X +rustls-pki-types@1.14.0 X X +rustls-webpki@0.103.13 X +rustversion@1.0.22 X X +ryu@1.0.23 X X +schannel@0.1.29 X +scopeguard@1.2.0 X X +security-framework@3.7.0 X X +security-framework-sys@2.17.0 X X +semver@1.0.28 X X +serde@1.0.228 X X +serde_core@1.0.228 X X +serde_derive@1.0.228 X X +serde_json@1.0.149 X X +serde_repr@0.1.20 X X +serde_urlencoded@0.7.1 X X +serde_with@3.18.0 X X +serde_with_macros@3.18.0 X X +sha1@0.10.6 X X +sha2@0.10.9 X X +shlex@1.3.0 X X +signal-hook-registry@1.4.8 X X +simdutf8@0.1.5 X X +slab@0.4.12 X +smallvec@1.15.1 X X +snafu@0.8.9 X X +snafu-derive@0.8.9 X X +socket2@0.6.3 X X +stable_deref_trait@1.2.1 X X +strsim@0.11.1 X +structmeta@0.3.0 X X +structmeta-derive@0.3.0 X X +strum@0.26.3 X +strum_macros@0.26.4 X +subtle@2.6.1 X +syn@2.0.117 X X +sync_wrapper@1.0.2 X +synstructure@0.13.2 X +tap@1.0.1 X +tempfile@3.27.0 X X +testcontainers@0.27.2 X X +thiserror@1.0.69 X X +thiserror@2.0.18 X X +thiserror-impl@1.0.69 X X +thiserror-impl@2.0.18 X X +time@0.3.47 X X +time-core@0.1.8 X X +tiny-keccak@2.0.2 X +tinystr@0.8.3 X +tokio@1.52.3 X +tokio-macros@2.7.0 X +tokio-rustls@0.26.4 X X +tokio-stream@0.1.18 X +tokio-util@0.7.18 X +tonic@0.14.5 X +tonic-prost@0.14.5 X +tower@0.5.3 X +tower-http@0.6.8 X +tower-layer@0.3.3 X +tower-service@0.3.3 X +tracing@0.1.44 X +tracing-attributes@0.1.31 X +tracing-core@0.1.36 X +try-lock@0.2.5 X +twox-hash@2.1.2 X +typenum@1.19.0 X X +unicode-ident@1.0.24 X X X +untrusted@0.9.0 X +ureq@3.3.0 X X +ureq-proto@0.6.0 X X +url@2.5.8 X X +utf8-zero@0.8.1 X X +utf8_iter@1.0.4 X X +utf8parse@0.2.2 X X +uuid@1.23.0 X X +value-bag@1.12.0 X X +version_check@0.9.5 X X +want@0.3.1 X +wasi@0.11.1+wasi-snapshot-preview1 X X X +wasip2@1.0.2+wasi-0.2.9 X X X +wasip3@0.4.0+wasi-0.3.0-rc-2026-01-06 X X X +wasm-bindgen@0.2.117 X X +wasm-bindgen-futures@0.4.67 X X +wasm-bindgen-macro@0.2.117 X X +wasm-bindgen-macro-support@0.2.117 X X +wasm-bindgen-shared@0.2.117 X X +wasm-streams@0.4.2 X X +web-sys@0.3.94 X X +web-time@1.1.0 X X +webpki-roots@1.0.6 X +winapi@0.3.9 X X +winapi-i686-pc-windows-gnu@0.4.0 X X +winapi-x86_64-pc-windows-gnu@0.4.0 X X +windows-core@0.62.2 X X +windows-implement@0.60.2 X X +windows-interface@0.59.3 X X +windows-link@0.2.1 X X +windows-result@0.4.1 X X +windows-strings@0.5.1 X X +windows-sys@0.52.0 X X +windows-sys@0.61.2 X X +windows-targets@0.52.6 X X +windows_aarch64_gnullvm@0.52.6 X X +windows_aarch64_msvc@0.52.6 X X +windows_i686_gnu@0.52.6 X X +windows_i686_gnullvm@0.52.6 X X +windows_i686_msvc@0.52.6 X X +windows_x86_64_gnu@0.52.6 X X +windows_x86_64_gnullvm@0.52.6 X X +windows_x86_64_msvc@0.52.6 X X +wit-bindgen@0.51.0 X X X +writeable@0.6.3 X +wyz@0.5.1 X +xattr@1.6.1 X X +yoke@0.8.2 X +yoke-derive@0.8.2 X +zerocopy@0.8.48 X X X +zerocopy-derive@0.8.48 X X X +zerofrom@0.1.7 X +zerofrom-derive@0.1.7 X +zeroize@1.8.2 X X +zerotrie@0.2.4 X +zerovec@0.11.6 X +zerovec-derive@0.11.3 X +zmij@1.0.21 X +zstd@0.13.3 X +zstd-safe@7.2.4 X X +zstd-sys@2.0.16+zstd.1.5.7 X X diff --git a/fluss-rust/crates/fluss-lake/README.md b/fluss-rust/crates/fluss-lake/README.md new file mode 100644 index 00000000000..83f5274618d --- /dev/null +++ b/fluss-rust/crates/fluss-lake/README.md @@ -0,0 +1,118 @@ + + +# UnionRead source preparation + +`fluss-lake` captures reusable source boundaries for Fluss-only and lake-plus-log reads without opening a lake catalog or reading lake files. This layer provides boundary capture and structured access; the default lake planner and reader are delivered separately. + +```rust,no_run +# use std::sync::Arc; +# use fluss::client::FlussConnection; +# use fluss::metadata::TablePath; +# use fluss_lake::{FlussLakeTable}; +# async fn capture_read_context(connection: Arc, path: &TablePath) -> fluss_lake::Result<()> { +let table = FlussLakeTable::open(connection, path).await?; +let context = table.capture_read_context().await?; +// Reuse this object across scans; engines own cross-process task transport. +# Ok(()) +# } +``` + +The context freezes table/schema identity and half-open log ranges `[start, stop)`. Lake-enabled tables additionally carry the physical lake mapping, format and optional readable snapshot. Non-lake tables skip the lake snapshot lookup. Projection and row predicates belong to the subsequent planning layer. `capture_read_context()` captures table-wide boundaries; `capture_read_context_with_partition_scope()` captures a restricted partition scope before querying offsets. Actual per-partition bucket counts are preserved. The context contains no credentials or physical lake-file tasks, is not a global transactional snapshot and does not acquire a retention lease. + +PK tables are accepted by this boundary API. Capturing boundaries does not validate an executor's merge-engine support or supply a KV snapshot: without a lake baseline, retained logs alone may omit existing keys. The subsequent default UnionRead executor targets deduplicate semantics; other merge engines and KV snapshot fallback remain separate work. + +## Partition scope and schema + +```rust,no_run +# use std::collections::HashMap; +# use fluss::metadata::PartitionSpec; +# use fluss_lake::{FlussLakeTable, Result}; +# async fn capture_read_context(table: &FlussLakeTable) -> Result<()> { +let filter = PartitionSpec::new(HashMap::from([("region", "US")])); +let context = table.capture_read_context_with_partition_scope(&filter).await?; +assert_eq!(context.partition_filter().get("region").map(String::as_str), Some("US")); +# Ok(()) +# } +``` + +The spec is an exact conjunction of partition-key/value pairs, reusing the core Admin API. It may contain a subset of the partition keys; an empty spec selects the whole table. Values must use Fluss partition metadata encoding. Automatic predicate-to-scope extraction belongs to the subsequent scan/planning PR. This boundary-only entry is not a general SQL expression API: engines must derive a safe spec or keep the default full scope. Only matching live partitions have offsets queried. A single ID recheck after capture excludes partitions that disappeared or were recreated under the same name. Lookup errors are suppressed only when the original partition ID is confirmed absent; failed rechecks do not prove deletion. The table identity is checked again after capture. + +The `partition_filter` map records the captured scope. Lake planning must apply it too, including to lake-only expired partitions. An empty log-range list does not imply an empty lake selection. A scoped context may be shared only by scans within its scope; use a table-wide context when sharing across different partition selections. This layer cannot enforce scope on an engine-native reader: the adapter must honor it, and the subsequent default planner must apply it before generating tasks. + +`validate_table()` checks the source table ID and logical path, plus the lake mapping and format when captured. Enabling tiering later does not change an already frozen Fluss-only read. It does not replace the captured schema or reject later column additions. Consumers must bind projection and predicates to `table_schema()` and map batch columns by stable column ID. The core log scanner's fixed-schema mode supports this mapping, but its target is the schema used to create the scanner: constructing a scanner from fresh metadata alone does not pin it to the context. Lake adapters likewise need to map the fixed snapshot's schema to the context's output schema. Unsupported conversions must fail rather than reinterpret columns by live positions. The context stores the typed Fluss schema captured during preparation. + +## Boundaries from a Java coordinator + +Java coordinators use `org.apache.fluss.client.metadata.UnionReadBoundaryPlanner` in `fluss-client`. It captures structured boundaries using the supplied Admin client, without opening lake files or depending on Flink. + +```java +UnionReadContext context = UnionReadBoundaryPlanner.captureReadContext(admin, tablePath); +UnionReadContext scoped = UnionReadBoundaryPlanner.captureReadContext(admin, tablePath, scope); +for (UnionReadContext.LogRange range : scoped.getLogRanges()) { + // Build engine-owned tasks using the bucket identity and [start, stop). +} +``` + +The result exposes the frozen schema, lake mapping/snapshot, partition scope and per-bucket ranges. It contains no arbitrary table properties or credentials. Capture applies the scope before querying offsets, uses each partition's bucket count, and rechecks partition IDs and table identity. Only confirmed removed partitions are skipped. The caller owns Admin. Capture blocks the planning thread and queries partitions sequentially; do not invoke it from Admin RPC callbacks. Interrupted waits abort capture without cancelling issued RPCs. Request timeouts follow client configuration. + +## Engine-owned transport + +Fluss defines structured read boundaries, not a universal cross-language context format. In-process callers share the context directly. A distributed connector maps these fields into its own tasks and owns serialization, version compatibility, distribution and retry. Its worker must preserve source IDs, captured schema, lake mapping/snapshot, partition scope, per-partition bucket counts, baseline presence and concrete offsets. No worker should recapture metadata to reconstruct a missing field. Runtime credentials remain local. + +A Rust worker can import engine-decoded fields without another metadata lookup: + +```rust,no_run +# use fluss::metadata::{TableInfo, PartitionSpec}; +# use fluss_lake::{FlussLakeLogRange, FlussLakeReadContext}; +# fn import(captured_table: &TableInfo, snapshot_id: Option, ranges: Vec, scope: &PartitionSpec) -> fluss_lake::Result<()> { +let context = FlussLakeReadContext::from_boundary_with_partition_filter( + captured_table, snapshot_id, ranges, scope, +)?; +# Ok(()) +# } +``` + +`FlussLakeLogRange::new()` accepts the physical bucket identity, business partition, bucket count, optional lake seam, stop and earliest offset. `from_boundary()` imports a table-wide capture; `from_boundary_with_partition_filter()` imports a scoped one. These constructors validate table/partition identities, concrete offsets and complete bucket coverage. They do not prove that independently supplied values were captured consistently: the coordinator must keep the readable snapshot and matching seams together, then capture stops. + +Each range associates its physical bucket with `FlussLakeLogPartition::Ordinary(identity)`, including an unpartitioned table. Consumers must handle the partition kind or reject execution. Bucket pruning additionally requires the routing rules for that kind. No C ABI or Java-to-Rust metadata conversion adapter is provided in this PR. + +Shared historical logs (`__historical__`) are not supported yet. They can contain changes for several business partitions and require a distinct log partition kind, snapshot-aligned offsets and record-level partition filtering. Reading lake-only expired partitions does not cover these pending changes. This remains follow-up work; no historical kind is emitted or accepted by this version. + +Each Rust log range exposes its baseline: + +| `baseline` | Meaning | +| --- | --- | +| `FlussLakeBaseline::Lake` | Read the context's pinned lake snapshot, then apply logs from `start_offset`, the snapshot's seam for this bucket. A zero seam is valid. | +| `FlussLakeBaseline::None` | No baseline; `start_offset` equals the captured earliest retained offset. | + +The baseline is determined by the presence of the bucket's seam, not by whether its offset is positive or lake file planning returns any files. The Rust constructor takes `snapshot_offset: Option` to preserve this distinction. A PK bucket without a baseline can only reconstruct state represented in the retained log. It cannot recover keys absent from that history. This matches the earliest-log fallback, not a guarantee of a complete historical PK view. `validate_available()` checks whether the frozen range was already missing at capture; execution must also detect later retention. + +KV snapshot baselines and SST readers remain future work. They need a snapshot ID and matching log seam captured before the stop offset, plus snapshot lifecycle management. Rust's non-exhaustive baseline and partition enums leave room for additional kinds; adapters must reject unsupported semantics rather than interpreting them as no baseline. + +Execution must include selected lake-only partitions from the pinned snapshot, even when they no longer appear among the live log ranges. For required log ranges, check `validate_available()` after safe pruning and still detect retention or truncation while reading. A captured earliest offset does not protect data from later expiry. Runtime credentials and engine-specific scheduling remain separate. + +## Execution and future modes + +Preparation fixes the captured bucket set; partitions created later are not added by workers. The final identity recheck excludes removed or replaced partitions without refreshing the frozen schema or offsets. Capture is not atomic across buckets or partition metadata. + +This context describes a bounded read, not streaming progress or a checkpoint. Streaming UnionRead can reuse source identities and baseline seams through an explicit future API; it must define changelog output, discovery and recovery separately. Runtime connections and credentials remain local. Engines own scheduling, retry and result publication. A missing frozen input must fail rather than switch snapshots or advance offsets; earlier batches are only a partial result of a failed attempt. + +Row-filter negotiation belongs to the subsequent adapter layer. Ordinary client filters remain strict; adapters may negotiate accepted/remaining conditions. A condition accepted only for pruning must remain for exact evaluation. The boundary API neither serializes query expressions nor silently drops unsupported filters. + +Shared historical-log support is tracked in [#5001](https://github.com/apache/fluss/issues/5001); it is separate from reading expired partitions already present in the lake snapshot. diff --git a/fluss-rust/crates/fluss-lake/src/boundary.rs b/fluss-rust/crates/fluss-lake/src/boundary.rs new file mode 100644 index 00000000000..8c065dfdf44 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/boundary.rs @@ -0,0 +1,476 @@ +// 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. + +//! Freezes the lake snapshot and server-issued log ranges within a partition scope. +//! +//! This module does not open a lake catalog, generate splits or schedule reads. + +use crate::error::planning_client_error; +use crate::{ + FlussLakeError, FlussLakeLogRange, FlussLakePartitionIdentity, FlussLakeReadContext, Result, +}; +use fluss::client::FlussAdmin; +use fluss::metadata::{ + LakeSnapshotInfo, PartitionInfo, PartitionSpec, TableBucket, TableInfo, TablePath, +}; +use fluss::rpc::message::OffsetSpec; +use futures::StreamExt; +use std::collections::{HashMap, HashSet}; + +/// Freezes live boundaries after applying the requested partition spec. +pub(crate) async fn freeze_read_context( + admin: &FlussAdmin, + table_info: &TableInfo, + partition_filter: &PartitionSpec, +) -> Result { + crate::partition::validate_partition_filter( + partition_filter.get_spec_map().keys(), + &table_info.partition_keys, + )?; + let table_path = &table_info.table_path; + if table_info.num_buckets <= 0 { + return Err(FlussLakeError::PlanningFailed(format!( + "table {table_path} has invalid bucket count {}", + table_info.num_buckets + ))); + } + + let readable_snapshot = if table_info + .table_config + .is_datalake_enabled() + .map_err(|error| FlussLakeError::PlanningFailed(error.to_string()))? + { + match admin.get_lake_snapshot(table_path, None, Some(true)).await { + Ok(snapshot) => Some(snapshot), + Err(error) + if error.api_error() == Some(fluss::error::FlussError::LakeSnapshotNotExist) => + { + None + } + Err(error) => { + return Err(planning_client_error("get readable lake snapshot", error)); + } + } + } else { + None + }; + + let snapshot_offsets = + collect_snapshot_offsets(table_path, table_info.table_id, readable_snapshot.as_ref())?; + let partitions = if table_info.partition_keys.is_empty() { + vec![( + None, + None, + FlussLakePartitionIdentity::Unpartitioned, + table_info.num_buckets, + )] + } else { + let mut partition_infos = admin + .list_partition_infos_with_spec(table_path, Some(partition_filter)) + .await + .map_err(|error| planning_client_error("list table partitions", error))?; + partition_infos.sort_by(|left, right| { + left.get_partition_name() + .cmp(&right.get_partition_name()) + .then_with(|| left.get_partition_id().cmp(&right.get_partition_id())) + }); + partition_infos + .into_iter() + .map(|partition| { + let count = partition_bucket_count(&partition)?; + let (id, name, identity) = partition_identity(partition); + Ok((id, name, identity, count)) + }) + .collect::>>()? + }; + + // Bound metadata fan-out and preserve the canonical partition order. + let captured = + futures::stream::iter(partitions.into_iter().map(|(id, name, identity, count)| { + let snapshot_offsets = &snapshot_offsets; + async move { + let result = async { + let bucket_ids: Vec = (0..count).collect(); + let (earliest_offsets, latest_offsets) = + load_server_offsets(admin, table_path, name.as_deref(), &bucket_ids) + .await?; + (0..count) + .map(|bucket_id| { + let bucket = + TableBucket::new_with_partition(table_info.table_id, id, bucket_id); + let earliest = required_offset(&earliest_offsets, &bucket, "earliest")?; + let stop = required_offset(&latest_offsets, &bucket, "latest")?; + let seam = snapshot_offsets.get(&bucket).copied(); + freeze_bucket_range( + bucket, + identity.clone(), + count, + seam, + earliest, + stop, + ) + }) + .collect::>>() + } + .await; + (id, result) + } + })) + .buffered(8) + .collect::>() + .await; + + let bucket_ranges = if table_info.partition_keys.is_empty() { + captured + .into_iter() + .map(|(_, result)| result) + .collect::>>()? + .into_iter() + .flatten() + .collect() + } else { + // Offset lookup resolves partition names internally. Recheck IDs even + // after successful lookups: a name may now refer to a new partition. + // One recheck covers all captures, including failures of any kind. + let live_ids = admin + .list_partition_infos_with_spec(table_path, Some(partition_filter)) + .await + .map(|partitions| { + partitions + .into_iter() + .map(|p| p.get_partition_id()) + .collect() + }) + .map_err(|error| planning_client_error("recheck table partitions", error)); + retain_live_ranges(captured, live_ids)? + }; + + let context = FlussLakeReadContext::from_boundary_with_partition_filter( + table_info, + readable_snapshot.map(|snapshot| snapshot.snapshot_id), + bucket_ranges, + partition_filter, + )?; + // Name-based offset calls may also cross a table drop/recreate. Validate + // identity once after capture, without replacing the frozen schema. + let live_table = admin + .get_table_info(table_path) + .await + .map_err(|error| planning_client_error("recheck source table identity", error))?; + context.validate_table(&live_table)?; + Ok(context) +} + +fn retain_live_ranges( + captured: Vec<(Option, Result>)>, + live_ids: Result>, +) -> Result> { + let live_ids = match live_ids { + Ok(ids) => ids, + Err(recheck_error) => { + // A failed recheck does not prove deletion. Preserve the lookup + // error when there was one, rather than hiding it with a new error. + return Err(captured + .into_iter() + .find_map(|(_, result)| result.err()) + .unwrap_or(recheck_error)); + } + }; + captured + .into_iter() + .filter(|(id, _)| id.is_some_and(|id| live_ids.contains(&id))) + .map(|(_, result)| result) + .collect::>>() + .map(|ranges| ranges.into_iter().flatten().collect()) +} + +/// Admin resolves legacy counts; never substitute the table default here. +fn partition_bucket_count(partition: &PartitionInfo) -> Result { + let count = partition.get_bucket_count(); + if count <= 0 { + return Err(FlussLakeError::PlanningFailed(format!( + "partition {} has invalid bucket count {count}", + partition.get_partition_id() + ))); + } + Ok(count) +} + +fn partition_identity( + partition_info: PartitionInfo, +) -> (Option, Option, FlussLakePartitionIdentity) { + let resolved = partition_info.get_resolved_partition_spec(); + let key_values = resolved + .get_partition_keys() + .iter() + .cloned() + .zip(resolved.get_partition_values().iter().cloned()) + .collect(); + ( + Some(partition_info.get_partition_id()), + Some(partition_info.get_partition_name()), + FlussLakePartitionIdentity::KeyValues(key_values), + ) +} + +fn collect_snapshot_offsets( + table_path: &TablePath, + table_id: i64, + readable_snapshot: Option<&LakeSnapshotInfo>, +) -> Result> { + let Some(snapshot) = readable_snapshot else { + return Ok(HashMap::new()); + }; + if snapshot.table_id != table_id { + return Err(FlussLakeError::PlanningFailed(format!( + "readable snapshot {} belongs to table id {}, but {table_path} resolved to table id {table_id}", + snapshot.snapshot_id, snapshot.table_id + ))); + } + + let mut offsets = HashMap::new(); + for bucket_snapshot in &snapshot.bucket_snapshots { + let Some(offset) = bucket_snapshot.log_offset else { + continue; + }; + if offset < 0 { + return Err(FlussLakeError::PlanningFailed(format!( + "readable snapshot {} returned negative log offset {offset} for bucket {}", + snapshot.snapshot_id, bucket_snapshot.bucket_id + ))); + } + let table_bucket = TableBucket::new_with_partition( + table_id, + bucket_snapshot.partition_id, + bucket_snapshot.bucket_id, + ); + if offsets.insert(table_bucket.clone(), offset).is_some() { + return Err(FlussLakeError::PlanningFailed(format!( + "readable snapshot {} contains duplicate boundary for {table_bucket}", + snapshot.snapshot_id + ))); + } + } + Ok(offsets) +} + +async fn load_server_offsets( + admin: &FlussAdmin, + table_path: &TablePath, + partition_name: Option<&str>, + bucket_ids: &[i32], +) -> Result<(HashMap, HashMap)> { + let earliest = list_server_offsets( + admin, + table_path, + partition_name, + bucket_ids, + OffsetSpec::Earliest, + ); + let latest = list_server_offsets( + admin, + table_path, + partition_name, + bucket_ids, + OffsetSpec::Latest, + ); + futures::try_join!(earliest, latest) +} + +async fn list_server_offsets( + admin: &FlussAdmin, + table_path: &TablePath, + partition_name: Option<&str>, + bucket_ids: &[i32], + offset_spec: OffsetSpec, +) -> Result> { + let offset_name = match offset_spec { + OffsetSpec::Earliest => "earliest", + OffsetSpec::Latest => "latest", + OffsetSpec::Timestamp(_) => "timestamp", + }; + let result = match partition_name { + Some(partition_name) => { + admin + .list_partition_offsets(table_path, partition_name, bucket_ids, offset_spec) + .await + } + None => { + admin + .list_offsets(table_path, bucket_ids, offset_spec) + .await + } + }; + result.map_err(|error| { + planning_client_error(&format!("get server-issued {offset_name} offsets"), error) + }) +} + +fn required_offset( + offsets: &HashMap, + table_bucket: &TableBucket, + offset_name: &str, +) -> Result { + offsets + .get(&table_bucket.bucket_id()) + .copied() + .ok_or_else(|| { + FlussLakeError::PlanningFailed(format!( + "server did not return the {offset_name} offset for {table_bucket}" + )) + }) +} + +fn freeze_bucket_range( + table_bucket: TableBucket, + partition_identity: FlussLakePartitionIdentity, + bucket_count: i32, + snapshot_offset: Option, + earliest_offset: i64, + stop_offset: i64, +) -> Result { + if earliest_offset < 0 || stop_offset < 0 { + return Err(FlussLakeError::PlanningFailed(format!( + "server returned a negative log boundary [{earliest_offset}, {stop_offset}) for {table_bucket}" + ))); + } + + if earliest_offset > stop_offset { + return Err(FlussLakeError::PlanningFailed(format!( + "server earliest offset {earliest_offset} exceeds latest offset {stop_offset} for {table_bucket}" + ))); + } + // Check retention availability after pruning: lake-only and pruned reads + // do not require the log tail. + let start_offset = snapshot_offset.unwrap_or(earliest_offset); + if start_offset > stop_offset { + return Err(FlussLakeError::PlanningFailed(format!( + "read start offset {start_offset} exceeds server latest offset {stop_offset} for {table_bucket}" + ))); + } + + Ok(FlussLakeLogRange::new( + table_bucket, + crate::FlussLakeLogPartition::Ordinary(partition_identity), + bucket_count, + snapshot_offset, + stop_offset, + earliest_offset, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn partition_recheck_discards_only_confirmed_deleted_ids() { + let error = || FlussLakeError::PlanningFailed("original offset failure".into()); + let captured = || { + vec![ + (Some(9), Err(error())), + (Some(10), Ok(vec![root_range(Some(12), 8, 20).unwrap()])), + ] + }; + // ID 11 may be a replacement with the same name as ID 9. + let ranges = retain_live_ranges(captured(), Ok(HashSet::from([10, 11]))).unwrap(); + assert_eq!(ranges.len(), 1); + assert_eq!( + retain_live_ranges(captured(), Ok(HashSet::from([9, 10]))) + .unwrap_err() + .to_string(), + error().to_string() + ); + assert_eq!( + retain_live_ranges( + captured(), + Err(FlussLakeError::PlanningFailed("recheck failed".into())) + ) + .unwrap_err() + .to_string(), + error().to_string() + ); + // Successful name-based offsets are unsafe too when the original ID disappeared. + let successful = vec![(Some(9), Ok(vec![root_range(Some(12), 8, 20).unwrap()]))]; + assert!( + retain_live_ranges(successful, Ok(HashSet::from([11]))) + .unwrap() + .is_empty() + ); + assert!(retain_live_ranges(vec![], Err(error())).is_err()); + } + + #[test] + fn uses_resolved_partition_bucket_counts_and_rejects_missing_counts() { + use fluss::metadata::ResolvedPartitionSpec; + use std::sync::Arc; + + let spec = + ResolvedPartitionSpec::new(Arc::from(["region".to_string()]), vec!["US".to_string()]) + .unwrap(); + let partition = PartitionInfo::new(42, spec); + assert!(partition_bucket_count(&partition).is_err()); + for count in [4, 2, 8, 0, -1] { + let result = partition_bucket_count(&partition.clone().with_bucket_count(count)); + assert_eq!(result.is_ok(), count > 0); + if count > 0 { + assert_eq!(result.unwrap(), count); + } + } + } + + fn root_range( + snapshot_offset: Option, + earliest_offset: i64, + stop_offset: i64, + ) -> Result { + freeze_bucket_range( + TableBucket::new(5, 2), + FlussLakePartitionIdentity::Unpartitioned, + 4, + snapshot_offset, + earliest_offset, + stop_offset, + ) + } + + #[test] + fn freezes_lake_seam_or_earliest_with_server_stop() { + use crate::FlussLakeBaseline::{Lake, None as NoBaseline}; + + for (seam, start, baseline) in [ + (Some(12), 12, Lake), + (Some(7), 7, Lake), + (None, 8, NoBaseline), + (Some(20), 20, Lake), + ] { + let range = root_range(seam, 8, 20).unwrap(); + assert_eq!(range.baseline(), baseline); + assert_eq!(range.start_offset(), start); + assert_eq!(range.stop_offset(), 20); + assert_eq!(range.is_empty(), start == 20); + assert_eq!(range.validate_available().is_ok(), start >= 8); + } + } + + #[test] + fn rejects_start_after_server_latest() { + let result = root_range(Some(21), 8, 20); + + assert!(matches!(result, Err(FlussLakeError::PlanningFailed(_)))); + } +} diff --git a/fluss-rust/crates/fluss-lake/src/error.rs b/fluss-rust/crates/fluss-lake/src/error.rs new file mode 100644 index 00000000000..9b298acbcb6 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/error.rs @@ -0,0 +1,90 @@ +// 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. + +//! Errors returned by UnionRead source preparation and context validation. + +use fluss::error::Error as ClientError; +use thiserror::Error; + +/// Result type returned by UnionRead APIs. +pub type Result = std::result::Result; + +/// Errors surfaced by source preparation and context validation. +/// +/// Match variants to classify failures; diagnostic messages are not a stable +/// protocol. Client failures retain their original error as the source. +#[derive(Debug, Error)] +#[non_exhaustive] +pub enum FlussLakeError { + /// Requested table is not configured for lake reads. + #[error("table is not lake-readable: {0}")] + NotLakeReadable(String), + + /// Source metadata cannot form a valid read boundary. + #[error("UnionRead planning failed: {0}")] + PlanningFailed(String), + + /// A frozen snapshot, partition or required log range is missing or unreadable. + #[error("UnionRead data unavailable: {0}")] + DataUnavailable(String), + + /// Table identity, schema or layout is incompatible with the frozen read inputs. + #[error("UnionRead schema incompatible: {0}")] + SchemaIncompatible(String), + + /// A core client operation failed while preparing source boundaries. + #[error("failed to {action}: {source}")] + Client { + action: String, + #[source] + source: Box, + }, + + /// Public engine-integration context is malformed or incomplete. + #[error("invalid UnionRead context: {0}")] + InvalidReadContext(String), +} + +pub(crate) fn planning_client_error(action: &str, error: ClientError) -> FlussLakeError { + FlussLakeError::Client { + action: action.to_string(), + source: Box::new(error), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::error::Error as _; + + #[test] + fn client_failure_preserves_action_and_typed_source() { + let error = planning_client_error( + "get table metadata", + ClientError::invalid_table("invalid layout"), + ); + assert!(error.to_string().contains("get table metadata")); + assert!(error.source().is_some()); + let FlussLakeError::Client { source, .. } = error else { + panic!("expected a client failure"); + }; + assert_eq!( + source.api_error(), + Some(fluss::error::FlussError::InvalidTableException) + ); + } +} diff --git a/fluss-rust/crates/fluss-lake/src/lib.rs b/fluss-rust/crates/fluss-lake/src/lib.rs new file mode 100644 index 00000000000..7c3ba0a63da --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/lib.rs @@ -0,0 +1,30 @@ +// 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. + +//! Reusable source boundaries for bounded lake and log reads. + +#![doc = include_str!("../README.md")] + +mod boundary; +mod error; +mod partition; +mod read_context; +mod table; + +pub use error::{FlussLakeError, Result}; +pub use partition::{FlussLakeLogPartition, FlussLakePartitionIdentity}; +pub use read_context::{FlussLakeBaseline, FlussLakeLogRange, FlussLakeReadContext}; +pub use table::FlussLakeTable; diff --git a/fluss-rust/crates/fluss-lake/src/partition.rs b/fluss-rust/crates/fluss-lake/src/partition.rs new file mode 100644 index 00000000000..5a967a82702 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/partition.rs @@ -0,0 +1,108 @@ +// 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. + +//! Source partition identity shared by preparation and default planning. + +use crate::{FlussLakeError, Result}; +use std::collections::BTreeMap; + +pub(crate) fn validate_partition_filter<'a>( + keys: impl IntoIterator, + partition_keys: &[String], +) -> Result<()> { + if keys.into_iter().any(|key| !partition_keys.contains(key)) { + return Err(FlussLakeError::InvalidReadContext( + "partition filter must reference only partition keys".into(), + )); + } + Ok(()) +} + +/// Association between a physical log partition and business partition values. +/// +/// Consumers must reject unsupported variants, not skip their log ranges. +/// Shared historical logs may cover multiple business partitions; support for +/// them will require a separate variant rather than a synthetic business value. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum FlussLakeLogPartition { + /// A regular Fluss partition, including the root of an unpartitioned table. + Ordinary(FlussLakePartitionIdentity), +} + +/// Stable partition identity exposed for scheduling and diagnostics. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum FlussLakePartitionIdentity { + /// Root identity for an unpartitioned table. + Unpartitioned, + /// Logical partition-key/value pairs in table partition-key order. + KeyValues(Vec<(String, String)>), +} + +impl FlussLakePartitionIdentity { + /// Whether this partition matches an exact, possibly partial partition spec. + /// Values use Fluss partition metadata encoding, not SQL literal coercion. + pub fn matches_filter(&self, filter: &BTreeMap) -> bool { + match self { + Self::Unpartitioned => filter.is_empty(), + Self::KeyValues(values) => filter.iter().all(|pair| { + values + .iter() + .any(|(key, value)| key == pair.0 && value == pair.1) + }), + } + } + + /// Checks the identity's shape and key order against a resolved table layout. + pub(crate) fn matches_keys(&self, partition_keys: &[String]) -> bool { + match self { + Self::Unpartitioned => partition_keys.is_empty(), + Self::KeyValues(values) => { + !partition_keys.is_empty() + && values.iter().map(|(key, _)| key).eq(partition_keys.iter()) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn partition_identity_requires_the_complete_ordered_key_layout() { + let keys = vec!["region".to_string(), "day".to_string()]; + assert!(FlussLakePartitionIdentity::Unpartitioned.matches_keys(&[])); + assert!(!FlussLakePartitionIdentity::Unpartitioned.matches_keys(&keys)); + for names in [ + vec!["region", "day"], + vec!["day", "region"], + vec!["region"], + vec!["region", "region"], + vec![], + ] { + let identity = FlussLakePartitionIdentity::KeyValues( + names + .iter() + .map(|key| ((*key).into(), String::new())) + .collect(), + ); + assert_eq!(identity.matches_keys(&keys), names == ["region", "day"]); + assert!(!identity.matches_keys(&[])); + } + } +} diff --git a/fluss-rust/crates/fluss-lake/src/read_context.rs b/fluss-rust/crates/fluss-lake/src/read_context.rs new file mode 100644 index 00000000000..e4e8c586493 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/read_context.rs @@ -0,0 +1,823 @@ +// 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. + +//! Source boundaries shared by the default reader and engine-native readers. + +use crate::FlussLakeLogPartition; +use crate::{FlussLakeError, FlussLakePartitionIdentity, Result}; +use fluss::metadata::{PartitionSpec, Schema, TableBucket, TableInfo, TablePath}; +use std::collections::{BTreeMap, HashMap, HashSet}; + +/// Baseline represented by a bucket's frozen log start. +/// +/// KV snapshots are not supported yet. Unknown baseline kinds are rejected +/// by consumers rather than interpreted as an empty baseline. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum FlussLakeBaseline { + /// No baseline: read the log retained at preparation time. + /// For PK tables this cannot recover keys absent from the retained history. + None, + /// The context's pinned lake snapshot, with `start_offset` as its seam. + /// A seam of zero is valid and does not mean the baseline is absent. + Lake, +} + +/// One frozen, half-open Fluss log range, not an engine scheduling unit. +/// +/// Offsets are comparable only within `table_bucket`. Engines must preserve +/// change types and offset order when reconciling primary-key tails. Do not +/// infer the lake file layout from the bucket id. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FlussLakeLogRange { + pub(crate) table_bucket: TableBucket, + pub(crate) partition: FlussLakeLogPartition, + pub(crate) bucket_count: i32, + pub(crate) baseline: FlussLakeBaseline, + pub(crate) start_offset: i64, + pub(crate) stop_offset: i64, + pub(crate) earliest_offset: i64, +} + +impl FlussLakeLogRange { + /// Creates an externally captured log range without querying Fluss. + /// + /// `snapshot_offset` is the readable lake snapshot's seam for this bucket, + /// or `None` when there is no lake baseline. `Some(0)` still denotes a + /// lake baseline. With no seam, the start is the captured earliest offset. + /// `stop_offset` and `earliest_offset` must + /// already be resolved to concrete server offsets. Table identity, + /// partition layout and complete bucket coverage are checked by + /// [`FlussLakeReadContext::from_boundary`]. + pub fn new( + table_bucket: TableBucket, + partition: FlussLakeLogPartition, + bucket_count: i32, + snapshot_offset: Option, + stop_offset: i64, + earliest_offset: i64, + ) -> Self { + Self { + table_bucket, + partition, + bucket_count, + baseline: if snapshot_offset.is_some() { + FlussLakeBaseline::Lake + } else { + FlussLakeBaseline::None + }, + start_offset: snapshot_offset.unwrap_or(earliest_offset), + stop_offset, + earliest_offset, + } + } + + /// Server table, partition, and bucket identity used to subscribe. + pub fn table_bucket(&self) -> &TableBucket { + &self.table_bucket + } + + /// Association with business partitions, distinct from the physical bucket. + /// Consumers must handle this kind or reject the read before emitting results. + pub fn partition(&self) -> &FlussLakeLogPartition { + &self.partition + } + + /// Actual bucket count of this physical log partition. + /// + /// This may differ from the table default. All ranges of the same partition + /// carry the same count and cover exactly `0..bucket_count`. Bucket pruning + /// also requires the routing rules for this log partition kind. + pub fn bucket_count(&self) -> i32 { + self.bucket_count + } + + /// Baseline to read before applying this range's changelog. + pub fn baseline(&self) -> FlussLakeBaseline { + self.baseline + } + + /// Inclusive seam offset, or the captured earliest offset without a baseline. + pub fn start_offset(&self) -> i64 { + self.start_offset + } + + /// Exclusive, server-issued end offset captured during preparation. + pub fn stop_offset(&self) -> i64 { + self.stop_offset + } + + /// Earliest retained offset observed during preparation. + /// + /// Retention may advance after preparation. This is not a lease. + pub fn earliest_offset(&self) -> i64 { + self.earliest_offset + } + + /// Whether this range contains no changelog records. + pub fn is_empty(&self) -> bool { + self.start_offset == self.stop_offset + } + + /// Checks whether the required range was already unavailable at preparation. + /// + /// Call this for every range an engine-native union read needs, after safe + /// partition/bucket pruning. Lake-only reads do not need this check. A + /// reader must still detect retention or truncation during execution. + pub fn validate_available(&self) -> Result<()> { + if self.start_offset < self.earliest_offset { + return Err(FlussLakeError::DataUnavailable(format!( + "required log range [{}, {}) for {} starts before retained offset {}; the frozen view cannot be reconstructed", + self.start_offset, self.stop_offset, self.table_bucket, self.earliest_offset + ))); + } + Ok(()) + } +} + +/// Immutable source state for one table, independent of physical execution. +/// +/// A context fixes the readable lake snapshot and live Fluss bucket boundaries +/// within its partition scope. It is not a global transactional +/// snapshot, a retention lease, or a file/split plan. It carries neither +/// credentials nor projection or row-filter configuration. Scans can share +/// this boundary only within its captured partition scope. +/// +/// The lake snapshot may also contain partitions no longer present in Fluss. +/// Engines must plan the entire selected lake baseline, including those +/// partitions; `log_ranges()` is not a complete lake partition inventory. +/// Engine-native readers own physical planning and reconciliation. Without a +/// baseline, PK results cover only keys reconstructible from the retained log. +/// Workers must not refresh boundaries or silently skip missing inputs. +/// Engines own task serialization and distribution; this object defines no +/// cross-language wire format. Import engine-decoded fields with +/// [`Self::from_boundary`], using the captured metadata rather than live metadata. +#[derive(Debug, Clone)] +pub struct FlussLakeReadContext { + table_path: TablePath, + table_id: i64, + table_modified_time: i64, + schema_id: i32, + schema: Schema, + partition_keys: Vec, + partition_filter: BTreeMap, + bucket_keys: Vec, + num_buckets: i32, + merge_engine: Option, + lake: Option, + log_ranges: Vec, +} + +/// Lake configuration is present only for lake-enabled tables. A configured +/// lake without a readable snapshot is distinct from a Fluss-only table. +#[derive(Debug, Clone)] +struct LakeDescriptor { + table_path: TablePath, + format: String, + snapshot_id: Option, +} + +impl LakeDescriptor { + fn capture(table: &TableInfo, snapshot_id: Option) -> Result> { + if !table + .table_config + .is_datalake_enabled() + .map_err(invalid_context)? + { + if snapshot_id.is_some() { + return Err(invalid_context( + "a non-lake table cannot have a lake snapshot", + )); + } + return Ok(None); + } + Ok(Some(Self { + table_path: crate::table::resolve_lake_table_path(table)?, + format: table + .table_config + .get_datalake_format() + .map_err(invalid_context)? + .ok_or_else(|| invalid_context("missing lake format"))? + .to_string(), + snapshot_id, + })) + } +} + +impl FlussLakeReadContext { + /// Imports boundaries already captured by a trusted query coordinator. + /// + /// Validates identity, layout and complete bucket coverage without + /// querying metadata or refreshing any offsets. The caller must supply + /// the readable snapshot id and its matching seam offsets in `log_ranges`, + /// including every bucket of each captured live partition. Lake-only + /// partitions remain discoverable through the pinned lake snapshot. + /// + /// Runtime credentials and scan configuration are supplied separately. + /// Retention availability must still be checked for the required ranges. + pub fn from_boundary( + table: &TableInfo, + lake_snapshot_id: Option, + log_ranges: Vec, + ) -> Result { + Self::from_boundary_with_partition_filter( + table, + lake_snapshot_id, + log_ranges, + &PartitionSpec::new(HashMap::::new()), + ) + } + + /// Imports a boundary restricted by an exact, possibly partial partition spec. + /// + /// Ordinary log partitions must match the filter. Consumers must apply the + /// same business-partition filter to the pinned lake snapshot, including lake-only partitions, + /// and must not reuse this context for a scan outside that scope. + pub fn from_boundary_with_partition_filter( + table: &TableInfo, + lake_snapshot_id: Option, + log_ranges: Vec, + partition_filter: &PartitionSpec, + ) -> Result { + let context = Self { + table_path: table.table_path.clone(), + table_id: table.table_id, + table_modified_time: table.modified_time, + schema_id: table.schema_id, + schema: table.schema.clone(), + partition_keys: table.partition_keys.to_vec(), + partition_filter: partition_filter + .get_spec_map() + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + bucket_keys: table.bucket_keys.clone(), + num_buckets: table.num_buckets, + merge_engine: table + .table_config + .get_merge_engine_type() + .map_err(invalid_context)? + .map(|engine| engine.to_string()), + lake: LakeDescriptor::capture(table, lake_snapshot_id)?, + log_ranges, + }; + context.validate()?; + Ok(context) + } + + /// Logical Fluss table path. Catalog credentials are supplied separately. + pub fn table_path(&self) -> &TablePath { + &self.table_path + } + + /// Physical lake database/table name, or `None` for a non-lake table. + /// Catalog credentials and runtime catalog mapping are supplied separately. + pub fn lake_table_path(&self) -> Option<&TablePath> { + self.lake.as_ref().map(|lake| &lake.table_path) + } + + /// Server-assigned identity, used to reject a dropped/recreated table. + pub fn table_id(&self) -> i64 { + self.table_id + } + + /// Source metadata modification time, not a query snapshot timestamp. + /// + /// Informational only: unrelated property or comment changes do not + /// invalidate the frozen read identity. + pub fn table_modified_time(&self) -> i64 { + self.table_modified_time + } + + /// Frozen Fluss schema identity. + pub fn schema_id(&self) -> i32 { + self.schema_id + } + + /// Full, unprojected source schema, including stable field identities. + pub fn table_schema(&self) -> &Schema { + &self.schema + } + + /// Table partition-key names in canonical order. + pub fn partition_keys(&self) -> &[String] { + &self.partition_keys + } + + /// Table bucket-key names. Pruning also requires the routing rules for each + /// log partition kind; these keys alone do not establish those rules. + pub fn bucket_keys(&self) -> &[String] { + &self.bucket_keys + } + + /// Exact partition scope shared by lake planning and log capture. + /// An empty map means the whole table. Values use Fluss metadata encoding. + pub fn partition_filter(&self) -> &BTreeMap { + &self.partition_filter + } + + /// Frozen table default for new partitions, not execution parallelism. + /// + /// Use each log range's `bucket_count()` for live partitions and the lake + /// source's layout for lake-only partitions, not this default. + pub fn num_buckets(&self) -> i32 { + self.num_buckets + } + + /// Lake format identifier, such as `paimon`, or `None` for a non-lake table. + pub fn lake_format(&self) -> Option<&str> { + self.lake.as_ref().map(|lake| lake.format.as_str()) + } + + /// Explicit Fluss merge engine, or `None` for the default semantics. + /// + /// Default semantics are append for log tables and deduplicate for PK + /// tables. The lake adapter must separately validate its baseline merge + /// engine and layout before executing a union read. + pub fn merge_engine(&self) -> Option<&str> { + self.merge_engine.as_deref() + } + + /// Readable lake snapshot to pass to the lake planner without refreshing it. + pub fn lake_snapshot_id(&self) -> Option { + self.lake.as_ref().and_then(|lake| lake.snapshot_id) + } + + /// Physical log ranges required within the business `partition_filter()` scope. + /// Consumers must handle every log partition kind or reject the read; ranges + /// must not be skipped merely because their kind is unsupported. + /// Lake-only partitions must be planned separately within the same scope. + pub fn log_ranges(&self) -> &[FlussLakeLogRange] { + &self.log_ranges + } + + /// Checks that live metadata still identifies the same source table. + /// + /// Schema and read semantics come from this context, not live metadata. + /// Readers must map data by column ID into `table_schema()`, even when + /// columns were added during or after preparation. This identity check + /// does not certify that a reader supports every schema evolution. + /// + /// This does not refresh the snapshot, add newly created partitions, or + /// verify that data is still retained. An engine-native adapter must also + /// validate its lake layout and enforce bounded log availability. + pub fn validate_table(&self, table: &TableInfo) -> Result<()> { + let lake_changed = if let Some(lake) = &self.lake { + let live = LakeDescriptor::capture(table, None)?; + live.as_ref() + .is_none_or(|live| live.table_path != lake.table_path || live.format != lake.format) + } else { + // Enabling tiering later does not change a frozen Fluss-only read. + false + }; + if self.table_path() != &table.table_path + || self.table_id() != table.table_id + || lake_changed + { + return Err(FlussLakeError::SchemaIncompatible(format!( + "table metadata no longer matches the frozen read context for {} (table id {}, schema id {})", + self.table_path(), + self.table_id(), + self.schema_id() + ))); + } + Ok(()) + } +} + +impl FlussLakeReadContext { + fn validate(&self) -> Result<()> { + if self.table_path.database().is_empty() + || self.table_path.table().is_empty() + || self.table_id < 0 + || self.table_modified_time < 0 + || self.schema_id < 0 + || self.num_buckets <= 0 + { + return Err(invalid_context( + "invalid table, schema, snapshot, or bucket identity", + )); + } + if let Some(lake) = &self.lake + && (lake.table_path.database().is_empty() + || lake.table_path.table().is_empty() + || lake.snapshot_id.is_some_and(|id| id < 0)) + { + return Err(invalid_context("invalid lake identity")); + } + let schema = &self.schema; + crate::partition::validate_partition_filter( + self.partition_filter.keys(), + &self.partition_keys, + )?; + let columns: HashSet<_> = schema.column_names().into_iter().collect(); + for keys in [&self.partition_keys, &self.bucket_keys] { + let mut seen = HashSet::new(); + if keys + .iter() + .any(|key| !columns.contains(key.as_str()) || !seen.insert(key)) + { + return Err(invalid_context("unknown or duplicate partition/bucket key")); + } + } + if self + .bucket_keys + .iter() + .any(|key| self.partition_keys.contains(key)) + { + return Err(invalid_context( + "bucket keys must not contain partition keys", + )); + } + if schema.primary_key().is_some() { + let primary_keys = schema.primary_key_column_names(); + if self.bucket_keys.is_empty() + || self + .partition_keys + .iter() + .chain(&self.bucket_keys) + .any(|key| !primary_keys.contains(&key.as_str())) + { + return Err(invalid_context( + "PK partition and bucket keys must belong to the primary key", + )); + } + } + let mut groups = HashMap::new(); + let mut partition_ids = HashMap::new(); + for range in &self.log_ranges { + let bucket = range.table_bucket(); + let FlussLakeLogPartition::Ordinary(identity) = range.partition(); + if bucket.table_id() != self.table_id + || bucket.bucket_id() < 0 + || range.bucket_count <= 0 + || bucket.bucket_id() >= range.bucket_count + || bucket.partition_id().is_some_and(|id| id < 0) + || !identity.matches_filter(&self.partition_filter) + || range.start_offset < 0 + || range.earliest_offset < 0 + || range.start_offset > range.stop_offset + || range.earliest_offset > range.stop_offset + || match range.baseline { + FlussLakeBaseline::Lake => self + .lake + .as_ref() + .and_then(|lake| lake.snapshot_id) + .is_none(), + FlussLakeBaseline::None => range.start_offset != range.earliest_offset, + } + { + return Err(invalid_context("invalid bucket identity or log boundary")); + } + match identity { + FlussLakePartitionIdentity::Unpartitioned => { + if !self.partition_keys.is_empty() + || bucket.partition_id().is_some() + || range.bucket_count != self.num_buckets + { + return Err(invalid_context( + "unpartitioned range in a partitioned context", + )); + } + } + FlussLakePartitionIdentity::KeyValues(_) => { + if bucket.partition_id().is_none() + || !identity.matches_keys(&self.partition_keys) + { + return Err(invalid_context( + "partition values do not match the frozen schema", + )); + } + } + } + if let Some(existing) = partition_ids.insert(bucket.partition_id(), identity) + && existing != identity + { + return Err(invalid_context( + "partition id has conflicting logical identities", + )); + } + let (partition_id, bucket_count, buckets) = groups + .entry(identity) + .or_insert_with(|| (bucket.partition_id(), range.bucket_count, HashSet::new())); + if *partition_id != bucket.partition_id() + || *bucket_count != range.bucket_count + || !buckets.insert(bucket.bucket_id()) + { + return Err(invalid_context( + "duplicate or conflicting partition/bucket range", + )); + } + } + if (self.partition_keys.is_empty() && groups.len() != 1) + || groups + .values() + .any(|(_, count, buckets)| buckets.len() != *count as usize) + { + return Err(invalid_context( + "context must cover every bucket of each captured live partition", + )); + } + Ok(()) + } +} + +fn invalid_context(error: impl std::fmt::Display) -> FlussLakeError { + FlussLakeError::InvalidReadContext(error.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use fluss::metadata::DataTypes; + + fn table(partitioned: bool) -> TableInfo { + TableInfo::new( + TablePath::new("fluss", "orders"), + 7, + 2, + Schema::builder() + .column("id", DataTypes::int()) + .column("region", DataTypes::string()) + .column("amount", DataTypes::bigint()) + .primary_key_named("orders_pk", vec!["id", "region"]) + .unwrap() + .build() + .unwrap(), + vec!["id".to_string()], + if partitioned { + vec!["region".to_string()] + } else { + vec![] + } + .into(), + 2, + HashMap::from([ + ("table.datalake.enabled".to_string(), "true".to_string()), + ("table.datalake.format".to_string(), "paimon".to_string()), + ( + "table.datalake.paimon.s3.secret-key".to_string(), + "SECRET".to_string(), + ), + ( + "table.datalake.paimon.warehouse".to_string(), + "s3://private/warehouse".to_string(), + ), + ]), + HashMap::new(), + None, + 0, + 0, + ) + } + + fn context(partitioned: bool) -> FlussLakeReadContext { + let partition = if partitioned { + FlussLakePartitionIdentity::KeyValues(vec![("region".to_string(), "US".to_string())]) + } else { + FlussLakePartitionIdentity::Unpartitioned + }; + FlussLakeReadContext::from_boundary( + &table(partitioned), + Some(42), + (0..2) + .map(|bucket| { + FlussLakeLogRange::new( + TableBucket::new_with_partition(7, partitioned.then_some(9), bucket), + FlussLakeLogPartition::Ordinary(partition.clone()), + 2, + Some(12), + 20, + 8, + ) + }) + .collect(), + ) + .unwrap() + } + + #[test] + fn typed_boundaries_preserve_schema_and_offsets() { + for partitioned in [false, true] { + let original = context(partitioned); + let imported = FlussLakeReadContext::from_boundary( + &table(partitioned), + Some(42), + original.log_ranges().to_vec(), + ) + .unwrap(); + assert_eq!(imported.log_ranges(), original.log_ranges()); + assert_eq!(imported.table_schema(), &table(partitioned).schema); + imported.validate_table(&table(partitioned)).unwrap(); + } + } + + #[test] + fn validates_imported_bucket_coverage_and_bounds() { + let original = context(true).log_ranges().to_vec(); + let mut duplicate = original.clone(); + duplicate[1] = duplicate[0].clone(); + let mut wrong_table = original.clone(); + wrong_table[0].table_bucket = TableBucket::new_with_partition(8, Some(9), 0); + let mut wrong_bounds = original.clone(); + wrong_bounds[0].stop_offset = 11; + let mut wrong_count = original.clone(); + wrong_count[0].bucket_count = 4; + for ranges in [ + vec![original[0].clone()], + duplicate, + wrong_table, + wrong_bounds, + wrong_count, + ] { + assert!(matches!( + FlussLakeReadContext::from_boundary(&table(true), Some(42), ranges), + Err(FlussLakeError::InvalidReadContext(_)) + )); + } + assert!(FlussLakeReadContext::from_boundary(&table(true), None, original).is_err()); + } + + #[test] + fn no_baseline_starts_at_earliest_and_zero_seam_is_still_a_baseline() { + let mut info = table(false); + info.table_config + .properties + .insert("table.datalake.enabled".into(), "false".into()); + let ranges = (0..2) + .map(|bucket| { + FlussLakeLogRange::new( + TableBucket::new(7, bucket), + FlussLakeLogPartition::Ordinary(FlussLakePartitionIdentity::Unpartitioned), + 2, + None, + 20, + 8, + ) + }) + .collect(); + let plain = FlussLakeReadContext::from_boundary(&info, None, ranges).unwrap(); + assert_eq!(plain.lake_format(), None); + assert!( + plain + .log_ranges() + .iter() + .all(|r| r.start_offset() == 8 && r.baseline() == FlussLakeBaseline::None) + ); + plain.validate_table(&table(false)).unwrap(); + assert!( + FlussLakeReadContext::from_boundary(&info, Some(42), plain.log_ranges().to_vec()) + .is_err() + ); + let ranges = (0..2) + .map(|bucket| { + FlussLakeLogRange::new( + TableBucket::new(7, bucket), + FlussLakeLogPartition::Ordinary(FlussLakePartitionIdentity::Unpartitioned), + 2, + (bucket == 0).then_some(0), + 20, + 0, + ) + }) + .collect(); + let mixed = FlussLakeReadContext::from_boundary(&table(false), Some(42), ranges).unwrap(); + assert_eq!(mixed.log_ranges()[0].baseline(), FlussLakeBaseline::Lake); + assert_eq!(mixed.log_ranges()[1].baseline(), FlussLakeBaseline::None); + assert!(mixed.log_ranges().iter().all(|r| r.start_offset() == 0)); + } + + #[test] + fn retained_gap_is_rejected_only_when_range_is_required() { + let mut ranges = context(false).log_ranges().to_vec(); + ranges[0].earliest_offset = 15; + let context = FlussLakeReadContext::from_boundary(&table(false), Some(42), ranges).unwrap(); + assert!(matches!( + context.log_ranges()[0].validate_available(), + Err(FlussLakeError::DataUnavailable(_)) + )); + } + + #[test] + fn scope_covers_lake_only_partitions_and_rejects_unrelated_ranges() { + let filter = PartitionSpec::new(HashMap::from([("region", "US")])); + let context = FlussLakeReadContext::from_boundary_with_partition_filter( + &table(true), + Some(42), + vec![], + &filter, + ) + .unwrap(); + assert!(context.log_ranges().is_empty()); + assert_eq!(context.lake_snapshot_id(), Some(42)); + assert_eq!( + context.partition_filter().get("region").map(String::as_str), + Some("US") + ); + for filter in [ + PartitionSpec::new(HashMap::from([("region", "EU")])), + PartitionSpec::new(HashMap::from([("amount", "1")])), + ] { + assert!( + FlussLakeReadContext::from_boundary_with_partition_filter( + &table(true), + Some(42), + self::context(true).log_ranges().to_vec(), + &filter, + ) + .is_err() + ); + } + } + #[test] + fn validates_source_identity_without_refreshing_schema_or_bounds() { + let context = context(true); + let mut recreated = table(true); + recreated.table_id += 1; + let mut evolved = table(true); + evolved.schema_id += 1; + evolved.num_buckets = 4; + evolved.modified_time += 1; + evolved.comment = Some("new description".into()); + let mut rebucketed = table(false); + rebucketed.num_buckets += 1; + let mut moved = table(true); + moved.table_path = TablePath::new("fluss", "another"); + for table in [recreated, moved] { + assert!(matches!( + context.validate_table(&table), + Err(FlussLakeError::SchemaIncompatible(_)) + )); + } + evolved.schema = Schema::builder() + .column("id", DataTypes::int()) + .column("region", DataTypes::string()) + .column("amount", DataTypes::bigint()) + .column("added", DataTypes::string()) + .primary_key_named("orders_pk", vec!["id", "region"]) + .unwrap() + .build() + .unwrap(); + context.validate_table(&evolved).unwrap(); + assert_eq!(context.table_schema().columns().len(), 3); + assert_eq!(context.schema_id(), 2); + assert_eq!(context.num_buckets(), 2); + assert!( + context + .log_ranges() + .iter() + .all(|range| range.bucket_count() == 2) + ); + self::context(false).validate_table(&rebucketed).unwrap(); + assert_eq!(context.lake_snapshot_id(), Some(42)); + assert_eq!(context.log_ranges()[0].stop_offset(), 20); + } + + #[test] + fn merge_engine_is_frozen_independently_of_live_metadata() { + let mut table = table(true); + table + .properties + .insert("table.merge-engine".into(), "versioned".into()); + table.table_config = + fluss::metadata::TableConfig::from_properties(table.properties.clone()); + let context = FlussLakeReadContext::from_boundary( + &table, + Some(42), + context(true).log_ranges().to_vec(), + ) + .unwrap(); + assert_eq!(context.merge_engine(), Some("versioned")); + context.validate_table(&table).unwrap(); + table.table_config.properties.remove("table.merge-engine"); + context.validate_table(&table).unwrap(); + assert_eq!(context.merge_engine(), Some("versioned")); + } + + #[test] + fn rejects_changed_lake_mapping() { + let context = context(true); + let mut changed = table(true); + changed + .properties + .insert("table.datalake.table-name".into(), "other".into()); + assert!(matches!( + context.validate_table(&changed), + Err(FlussLakeError::SchemaIncompatible(_)) + )); + } +} diff --git a/fluss-rust/crates/fluss-lake/src/table.rs b/fluss-rust/crates/fluss-lake/src/table.rs new file mode 100644 index 00000000000..384ec3c1719 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/src/table.rs @@ -0,0 +1,148 @@ +// 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. + +//! Table entry point for capturing reusable UnionRead source boundaries. + +use crate::error::planning_client_error; +use crate::{FlussLakeError, FlussLakeReadContext, Result}; +use fluss::client::FlussConnection; +use fluss::metadata::{PartitionSpec, TableInfo, TablePath}; +use std::collections::HashMap; +use std::fmt::{Debug, Formatter}; +use std::sync::Arc; + +/// Entry point for source preparation over a Fluss table. +#[derive(Clone)] +pub struct FlussLakeTable { + connection: Arc, + table_path: TablePath, +} + +impl FlussLakeTable { + /// Opens a table for bounded Fluss-only or lake and log reads. + pub async fn open(connection: Arc, table_path: &TablePath) -> Result { + let admin = connection + .get_admin() + .map_err(|error| planning_client_error("create Fluss admin client", error))?; + let table_info = admin + .get_table_info(table_path) + .await + .map_err(|error| planning_client_error("get table metadata", error))?; + Self::try_from_table_info(connection, &table_info) + } + + /// Creates a table from metadata already resolved by the caller. + /// Preparation fetches fresh metadata before capturing source boundaries. + pub fn try_from_table_info( + connection: Arc, + table_info: &TableInfo, + ) -> Result { + validate_readable_table(table_info)?; + Ok(Self { + connection, + table_path: table_info.table_path.clone(), + }) + } + + /// Captures table-wide source boundaries without planning lake files. + /// + /// The result may be shared by scans with different projections and + /// predicates, including separate references to the table in one query. + /// This does not open a lake catalog or acquire a retention lease. + pub async fn capture_read_context(&self) -> Result { + self.capture_read_context_with_partition_scope(&PartitionSpec::new( + HashMap::::new(), + )) + .await + } + + /// Captures only partitions matching an exact, possibly partial spec. + /// + /// Scope selection happens before offset queries and also constrains lake + /// planning. This is an input restriction, not an optional pruning hint. An empty spec selects + /// the whole table. Non-partition keys are rejected; values must use the + /// same encoding as Fluss partition metadata. General SQL predicates + /// remain the engine's responsibility. + pub async fn capture_read_context_with_partition_scope( + &self, + scope: &PartitionSpec, + ) -> Result { + let admin = self + .connection + .get_admin() + .map_err(|error| planning_client_error("create Fluss admin client", error))?; + let table_info = admin + .get_table_info(&self.table_path) + .await + .map_err(|error| planning_client_error("get table metadata", error))?; + validate_readable_table(&table_info)?; + crate::boundary::freeze_read_context(&admin, &table_info, scope).await + } +} + +impl Debug for FlussLakeTable { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussLakeTable") + .field("table_path", &self.table_path) + .finish_non_exhaustive() + } +} + +/// Resolve the physical lake name without changing the logical Fluss identity. +pub(crate) fn resolve_lake_table_path(table_info: &TableInfo) -> Result { + let database = table_info + .properties + .get("table.datalake.database-name") + .map(String::as_str) + .unwrap_or(table_info.table_path.database()); + let table = table_info + .properties + .get("table.datalake.table-name") + .map(String::as_str) + .unwrap_or(table_info.table_path.table()); + if database.is_empty() || table.is_empty() { + return Err(FlussLakeError::PlanningFailed( + "lake database and table names must not be empty".to_string(), + )); + } + Ok(TablePath::new(database, table)) +} + +pub(crate) fn validate_readable_table(table_info: &TableInfo) -> Result<()> { + match table_info.table_config.is_datalake_enabled() { + Ok(true) => {} + Ok(false) => return Ok(()), + Err(error) => { + return Err(FlussLakeError::NotLakeReadable(format!( + "table {} has an invalid lake configuration: {error}", + table_info.table_path + ))); + } + } + match table_info.table_config.get_datalake_format() { + Ok(Some(_)) => Ok(()), + Ok(None) => Err(FlussLakeError::NotLakeReadable(format!( + "table {} does not define table.datalake.format", + table_info.table_path + ))), + Err(error) => Err(FlussLakeError::NotLakeReadable(format!( + "table {} has an invalid lake configuration: {error}", + table_info.table_path + ))), + } +} diff --git a/fluss-rust/crates/fluss-lake/tests/support/mod.rs b/fluss-rust/crates/fluss-lake/tests/support/mod.rs new file mode 100644 index 00000000000..254dc6d1d00 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/tests/support/mod.rs @@ -0,0 +1,160 @@ +// 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. + +use fluss::ServerType; +use fluss::client::FlussAdmin; +use fluss::error::{Error, FlussError}; +use fluss::metadata::{TableDescriptor, TablePath}; +use fluss::rpc::message::OffsetSpec; +use fluss_test_cluster::{FlussTestingCluster, FlussTestingClusterBuilder}; +use std::collections::HashMap; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; + +const READINESS_TIMEOUT: Duration = Duration::from_secs(30); +const READINESS_POLL_INTERVAL: Duration = Duration::from_millis(200); +extern "C" fn cleanup_on_exit() { + SHARED_CLUSTER.stop(); +} + +static CLUSTER_PORT: LazyLock = + LazyLock::new(|| 20_000 + (std::process::id() % 20_000) as u16); + +// Preparation needs lake-enabled metadata, but no host access to lake files. +static SHARED_CLUSTER: LazyLock = LazyLock::new(|| { + std::thread::spawn(|| { + let runtime = tokio::runtime::Runtime::new() + .expect("Failed to create UnionRead integration test runtime"); + runtime.block_on(async { + let cluster_conf = HashMap::from([ + ("datalake.enabled".to_string(), "true".to_string()), + ("datalake.format".to_string(), "paimon".to_string()), + ( + "datalake.paimon.metastore".to_string(), + "filesystem".to_string(), + ), + ( + "datalake.paimon.warehouse".to_string(), + "/tmp/union-read-warehouse".to_string(), + ), + ]); + let cluster = FlussTestingClusterBuilder::new_with_cluster_conf( + format!("rust-union-read-{}", std::process::id()), + &cluster_conf, + ) + .with_port(*CLUSTER_PORT) + .build() + .await; + wait_for_cluster_ready(&cluster).await; + + unsafe { + unsafe extern "C" { + fn atexit(callback: extern "C" fn()) -> std::os::raw::c_int; + } + atexit(cleanup_on_exit); + } + + cluster + }) + }) + .join() + .expect("Failed to initialize UnionRead integration test cluster") +}); + +pub fn get_shared_cluster() -> Arc { + Arc::new(SHARED_CLUSTER.clone()) +} + +async fn wait_for_cluster_ready(cluster: &FlussTestingCluster) { + let connection = cluster.get_fluss_connection().await; + let admin = connection + .get_admin() + .expect("Failed to get admin while waiting for UnionRead test cluster"); + let start = Instant::now(); + loop { + match admin.get_server_nodes().await { + Ok(nodes) + if nodes + .iter() + .any(|node| *node.server_type() == ServerType::TabletServer) => + { + return; + } + result if start.elapsed() >= READINESS_TIMEOUT => { + panic!( + "Timed out waiting for a registered tablet server in the UnionRead test \ + cluster. Last metadata result: {result:?}" + ); + } + _ => tokio::time::sleep(READINESS_POLL_INTERVAL).await, + } + } +} + +pub async fn create_table( + admin: &FlussAdmin, + table_path: &TablePath, + table_descriptor: &TableDescriptor, +) { + let create_start = Instant::now(); + loop { + match admin + .create_table(table_path, table_descriptor, false) + .await + { + Ok(()) => break, + Err(Error::FlussAPIError { api_error }) + if FlussError::for_code(api_error.code) == FlussError::InvalidReplicationFactor + && create_start.elapsed() < READINESS_TIMEOUT => + { + tokio::time::sleep(READINESS_POLL_INTERVAL).await; + } + Err(error) => { + panic!("Failed to create UnionRead integration test table: {error}"); + } + } + } + + let start = Instant::now(); + loop { + let readiness = if table_descriptor.is_partitioned() { + admin + .get_table_info(table_path) + .await + .map(|_| HashMap::new()) + } else { + admin + .list_offsets(table_path, &[0], OffsetSpec::Latest) + .await + }; + match readiness { + Ok(_) => return, + Err(error) if start.elapsed() < READINESS_TIMEOUT => { + tokio::time::sleep(READINESS_POLL_INTERVAL).await; + if start.elapsed() >= READINESS_TIMEOUT { + panic!( + "Timed out waiting for UnionRead integration test table {table_path}: {error}" + ); + } + } + Err(error) => { + panic!( + "Timed out waiting for UnionRead integration test table {table_path}: {error}" + ); + } + } + } +} diff --git a/fluss-rust/crates/fluss-lake/tests/test_prepare.rs b/fluss-rust/crates/fluss-lake/tests/test_prepare.rs new file mode 100644 index 00000000000..9d37a55e924 --- /dev/null +++ b/fluss-rust/crates/fluss-lake/tests/test_prepare.rs @@ -0,0 +1,234 @@ +// 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. + +#![cfg(feature = "integration_tests")] + +use arrow::array::{ArrayRef, Int32Array, StringArray}; +use arrow::record_batch::RecordBatch; +use fluss::metadata::{ + AddColumn, AlterTableChanges, ColumnPositionType, DataTypes, JsonSerde, PartitionSpec, Schema, + TableDescriptor, TablePath, +}; +use fluss_lake::{FlussLakeBaseline, FlussLakeReadContext, FlussLakeTable}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +mod support; + +// Source preparation needs only Fluss: no lake reader, tiering job or S3 fixture. +#[tokio::test] +async fn prepare_freezes_fluss_only_ranges_and_schema() { + let cluster = support::get_shared_cluster(); + let connection = Arc::new(cluster.get_fluss_connection().await); + let admin = connection.get_admin().unwrap(); + let path = TablePath::new("fluss", "prepare_only"); + let descriptor = TableDescriptor::builder() + .schema( + Schema::builder() + .column("id", DataTypes::int()) + .build() + .unwrap(), + ) + .build() + .unwrap(); + support::create_table(&admin, &path, &descriptor).await; + let source = connection.get_table(&path).await.unwrap(); + let writer = source.new_append().unwrap().create_writer().unwrap(); + let batch = |values| { + RecordBatch::try_from_iter(vec![("id", Arc::new(Int32Array::from(values)) as ArrayRef)]) + .unwrap() + }; + writer.append_arrow_batch(batch(vec![1, 2, 3])).unwrap(); + writer.flush().await.unwrap(); + + let table = FlussLakeTable::open(connection.clone(), &path) + .await + .unwrap(); + let plain = table.capture_read_context().await.unwrap(); + assert_eq!(plain.lake_snapshot_id(), None); + assert_eq!(plain.lake_format(), None); + assert_eq!(plain.log_ranges()[0].baseline(), FlussLakeBaseline::None); + assert_eq!(plain.log_ranges().len(), 1); + assert_eq!(plain.log_ranges()[0].start_offset(), 0); + assert_eq!(plain.log_ranges()[0].stop_offset(), 3); + plain.log_ranges()[0].validate_available().unwrap(); + let imported = FlussLakeReadContext::from_boundary( + source.get_table_info(), + plain.lake_snapshot_id(), + plain.log_ranges().to_vec(), + ) + .unwrap(); + imported + .validate_table(&admin.get_table_info(&path).await.unwrap()) + .unwrap(); + + writer.append_arrow_batch(batch(vec![4, 5])).unwrap(); + writer.flush().await.unwrap(); + assert_eq!(imported.log_ranges(), plain.log_ranges()); + assert_eq!(imported.table_schema(), plain.table_schema()); + assert_eq!(imported.log_ranges()[0].stop_offset(), 3); + assert_eq!( + table.capture_read_context().await.unwrap().log_ranges()[0].stop_offset(), + 5 + ); + drop(writer); + + admin + .alter_table( + &path, + false, + AlterTableChanges { + add_columns: vec![AddColumn { + column_name: "added".into(), + data_type_json: serde_json::to_vec(&DataTypes::int().serialize_json().unwrap()) + .unwrap(), + comment: None, + position: ColumnPositionType::Last, + }], + ..Default::default() + }, + ) + .await + .unwrap(); + let evolved = connection.get_table(&path).await.unwrap(); + imported.validate_table(evolved.get_table_info()).unwrap(); + assert_eq!(imported.log_ranges(), plain.log_ranges()); + assert_eq!(imported.table_schema(), plain.table_schema()); + admin.drop_table(&path, false).await.unwrap(); +} + +#[tokio::test] +async fn partition_filter_preserves_scope_and_resolved_bucket_counts() { + let cluster = support::get_shared_cluster(); + let connection = Arc::new(cluster.get_fluss_connection().await); + let admin = connection.get_admin().unwrap(); + let path = TablePath::new("fluss", "prepare_partitions"); + let descriptor = TableDescriptor::builder() + .schema( + Schema::builder() + .column("id", DataTypes::int()) + .column("region", DataTypes::string()) + .build() + .unwrap(), + ) + .partitioned_by(vec!["region"]) + .distributed_by(Some(1), vec![]) + .property("table.datalake.enabled", "true") + .property("table.datalake.format", "paimon") + .build() + .unwrap(); + support::create_table(&admin, &path, &descriptor).await; + for (region, count) in [("US", 1), ("EU", 2)] { + admin + .alter_table( + &path, + false, + AlterTableChanges { + new_bucket_count: Some(count), + ..Default::default() + }, + ) + .await + .unwrap(); + admin + .create_partition( + &path, + &PartitionSpec::new(HashMap::from([("region", region)])), + false, + ) + .await + .unwrap(); + let started = Instant::now(); + loop { + match admin + .list_partition_offsets( + &path, + region, + &(0..count).collect::>(), + fluss::rpc::message::OffsetSpec::Latest, + ) + .await + { + Ok(_) => break, + Err(error) => { + assert!( + started.elapsed() < Duration::from_secs(30), + "partition not ready: {error}" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + } + } + let source = connection.get_table(&path).await.unwrap(); + let writer = source.new_append().unwrap().create_writer().unwrap(); + // append_arrow_batch routes a batch using its first row's partition. + for (region, ids) in [("US", vec![1]), ("EU", vec![2, 3])] { + let count = ids.len(); + writer + .append_arrow_batch( + RecordBatch::try_from_iter(vec![ + ("id", Arc::new(Int32Array::from(ids)) as ArrayRef), + ( + "region", + Arc::new(StringArray::from(vec![region; count])) as ArrayRef, + ), + ]) + .unwrap(), + ) + .unwrap(); + } + writer.flush().await.unwrap(); + drop(writer); + let table = FlussLakeTable::open(connection.clone(), &path) + .await + .unwrap(); + let all = table.capture_read_context().await.unwrap(); + assert_eq!(all.log_ranges().len(), 3); + let filter = PartitionSpec::new(HashMap::from([("region", "US")])); + let selected = table + .capture_read_context_with_partition_scope(&filter) + .await + .unwrap(); + assert_eq!(selected.log_ranges().len(), 1); + assert_eq!(selected.log_ranges()[0].bucket_count(), 1); + assert_eq!(selected.log_ranges()[0].stop_offset(), 1); + let imported = FlussLakeReadContext::from_boundary_with_partition_filter( + source.get_table_info(), + selected.lake_snapshot_id(), + selected.log_ranges().to_vec(), + &filter, + ) + .unwrap(); + assert_eq!( + imported + .partition_filter() + .get("region") + .map(String::as_str), + Some("US") + ); + assert_eq!(imported.log_ranges(), selected.log_ranges()); + admin.drop_partition(&path, &filter, false).await.unwrap(); + let dropped = table + .capture_read_context_with_partition_scope(&filter) + .await + .unwrap(); + assert!(dropped.log_ranges().is_empty()); + assert_eq!(dropped.partition_filter(), selected.partition_filter()); + admin.drop_table(&path, false).await.unwrap(); +} diff --git a/fluss-rust/crates/fluss/src/metadata/table.rs b/fluss-rust/crates/fluss/src/metadata/table.rs index d322a987942..d061f783bb2 100644 --- a/fluss-rust/crates/fluss/src/metadata/table.rs +++ b/fluss-rust/crates/fluss/src/metadata/table.rs @@ -1259,11 +1259,52 @@ pub struct TableConfig { pub properties: HashMap, } +/// Merge engine configured for a primary-key table. +/// +/// The absence of `table.merge-engine` is the ordinary deduplicate/upsert +/// behavior. These explicit engines require semantics different from the +/// deduplicate current-view reconciliation. +#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumString)] +#[strum(ascii_case_insensitive)] +pub enum MergeEngineType { + #[strum(serialize = "first_row")] + FirstRow, + #[strum(serialize = "versioned")] + Versioned, + #[strum(serialize = "aggregation")] + Aggregation, +} + +impl Display for MergeEngineType { + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::FirstRow => "first_row", + Self::Versioned => "versioned", + Self::Aggregation => "aggregation", + }) + } +} + impl TableConfig { pub fn from_properties(properties: HashMap) -> Self { TableConfig { properties } } + /// Returns whether lake storage is enabled for the table. + /// + /// A table may preconfigure `table.datalake.format` while leaving lake + /// storage disabled, so callers must not infer enablement from the format. + pub fn is_datalake_enabled(&self) -> Result { + match self.properties.get("table.datalake.enabled") { + None => Ok(false), + Some(value) if value.eq_ignore_ascii_case("true") => Ok(true), + Some(value) if value.eq_ignore_ascii_case("false") => Ok(false), + Some(value) => Err(Error::invalid_table(format!( + "Invalid table.datalake.enabled {value:?}: expected true or false" + ))), + } + } + pub fn get_arrow_compression_info(&self) -> Result { ArrowCompressionInfo::from_conf(&self.properties) } @@ -1311,6 +1352,20 @@ impl TableConfig { LogFormat::parse(log_format) } + /// Returns the explicitly configured primary-key merge engine. + /// + /// `None` is the default deduplicate/upsert behavior. + pub fn get_merge_engine_type(&self) -> Result> { + self.properties + .get("table.merge-engine") + .map(|value| { + value.parse().map_err(|error| { + Error::invalid_table(format!("Invalid table.merge-engine {value:?}: {error}")) + }) + }) + .transpose() + } + pub fn get_auto_partition_strategy(&self) -> AutoPartitionStrategy { AutoPartitionStrategy::from(&self.properties) } @@ -1809,6 +1864,59 @@ mod tests { ); } + #[test] + fn table_config_parses_explicit_merge_engines_and_keeps_default_deduplicate() { + let default = TableConfig::from_properties(HashMap::new()); + assert_eq!(default.get_merge_engine_type().unwrap(), None); + + for (value, expected) in [ + ("first_row", MergeEngineType::FirstRow), + ("VERSIONED", MergeEngineType::Versioned), + ("aggregation", MergeEngineType::Aggregation), + ] { + let config = TableConfig::from_properties(HashMap::from([( + "table.merge-engine".to_string(), + value.to_string(), + )])); + assert_eq!(config.get_merge_engine_type().unwrap(), Some(expected)); + } + } + + #[test] + fn table_config_rejects_unknown_merge_engine() { + let config = TableConfig::from_properties(HashMap::from([( + "table.merge-engine".to_string(), + "deduplicate".to_string(), + )])); + + assert!(config.get_merge_engine_type().is_err()); + } + + #[test] + fn table_config_distinguishes_lake_format_from_lake_enablement() { + let format_only = TableConfig::from_properties(HashMap::from([( + "table.datalake.format".to_string(), + "paimon".to_string(), + )])); + assert!(!format_only.is_datalake_enabled().unwrap()); + + let enabled = TableConfig::from_properties(HashMap::from([ + ("table.datalake.enabled".to_string(), "TRUE".to_string()), + ("table.datalake.format".to_string(), "paimon".to_string()), + ])); + assert!(enabled.is_datalake_enabled().unwrap()); + } + + #[test] + fn table_config_rejects_invalid_lake_enablement() { + let config = TableConfig::from_properties(HashMap::from([( + "table.datalake.enabled".to_string(), + "yes".to_string(), + )])); + + assert!(config.is_datalake_enabled().is_err()); + } + #[test] fn auto_increment_column_requires_a_primary_key_table() { let err = Schema::builder()