[rust] Add reusable UnionRead source boundaries by naivedogger · Pull Request #4887 · apache/fluss

naivedogger · 2026-10-26T11:42:47Z

naivedogger · 2026-10-26T12:03:56Z

@fresh-borzoni @beryllw Could you please take a look at this one when you got a chance? Thanks!🙏

fresh-borzoni

@naivedogger Thank you for the PR, a couple of comments and questions, PTAL

fresh-borzoni · 2026-10-27T17:24:49Z

+        |(id, name, identity, count)| async move {
+            let bucket_ids: Vec<i32> = (0..count).collect();
+            let offsets =
+                load_server_offsets(admin, table_path, name.as_deref(), &bucket_ids).await?;


Partitions are listed first, and their offsets are looked up by name afterwards. If a partition is dropped in between, the whole prepare() fails. Auto-partition retention drops partitions as a matter of course, so long-running services will hit this.

A partition that disappears mid-prepare can be treated as if it was dropped before we started: its lake data is still read from the snapshot, and its log is gone either way. A drop can surface as different errors depending on timing, not only PartitionNotExists. So when a lookup fails, it may be simplest to re-check the partition list by id, skip the partition only if it's really gone, and otherwise return the original error.

I've updated boundary capture to recheck partition IDs after the offset lookups. A partition's log ranges are omitted only when its original ID is confirmed absent, including after same-name recreation. Other lookup failures still propagate, and lake data remains covered by the pinned snapshot.

fresh-borzoni · 2026-10-27T17:25:14Z

+    // Without a lake baseline the complete history is required, not merely
+    // the retained suffix. Availability is checked after scan-specific pruning
+    // so an irrelevant range cannot break a lake-only or pruned read.
+    let start_offset = snapshot_offset.unwrap_or(0);


For a bucket with no lake offset, the range starts at 0. For log tables that works, since the context also records the earliest offset, so a caller can choose to fail or read what's left. For PK tables it doesn't. If the changelog start was cleaned up, the current state can't be rebuilt from the log, although Fluss still has it in the KV snapshot.

Could a range say what sits under it (lake offset, KV snapshot, or nothing), even if only the lake case is implemented now? The KV snapshot would need to be captured here, before the stop offsets, so the snapshot and the log after it stay consistent. Adding it later would change the format. WDYT?

Each range now explicitly identifies its baseline as lake or none, including the distinction between a lake seam at zero and no baseline. Without a baseline, reads start at the captured earliest offset, so PK results only cover keys reconstructible from retained logs. KV snapshot fallback remains future work, and readers reject unknown baseline kinds.

fresh-borzoni · 2026-10-27T17:25:41Z

+# }
+```
+
+The context freezes table/schema identity, physical lake table mapping, a readable snapshot and half-open log ranges `[start, stop)`. Scan configuration, including projection and filters, belongs to the subsequent planning layer; preparation always captures table-wide boundaries. 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. Its JSON representation is an evolving Rust contract, not a stable cross-language ABI.


Engines that plan in Java (Trino, Spark, StarRocks, Doris) can't call prepare(), and the README says not to treat the JSON as cross-language. How do you see those engines getting a context? If a Java producer is planned, it's worth settling the format before later PRs build on it.

I've added UnionReadBoundaryPlanner.captureReadContext(admin, tablePath), with an overload for partition scope, to capture the boundaries and return the encoded context in one call. Context v1 is now a documented cross-language format, so Java coordinators can produce it without calling Rust or assembling the JSON themselves.

fresh-borzoni · 2026-10-27T17:26:25Z

+/// Uses the public Fluss schema JSON, not Rust's internal Schema serde.
+/// No arbitrary catalog/table property map is allowed in this descriptor.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]


Rejecting unknown fields and accepting exactly one version means any new field breaks existing readers. The side that prepares the context and the side that reads it would always have to upgrade together, which is hard during rolling upgrades.

A more forgiving rule would be to ignore unknown optional fields, and to bump the version (or list a required feature) only when a change alters meaning. Also, the strictness is uneven today: unknown fields inside table_bucket are accepted.

Readers now ignore unknown optional fields while rejecting missing required fields, unsupported versions and unknown semantic variants. Additions are compatible only when ignoring them preserves the read semantics; incompatible changes require a version bump, with readers upgraded before producers use the new format.

fresh-borzoni · 2026-10-27T17:27:06Z

+        if self.table_path() != &table.table_path
+            || self.table_id() != table.table_id
+            || self.lake_table_path() != &crate::table::resolve_lake_table_path(table)?
+            || self.schema_id() != table.schema_id


validate_table fails if a column was added after prepare(), but I don't think a schema change can affect a frozen read. The frozen data already mixes schemas: if a column was added before prepare(), older log rows and older lake files don't have it. So readers have to map each batch to one schema anyway. The Rust scanner does that by column id, and Paimon reads a snapshot with that snapshot's own schema. If readers use the frozen schema rather than the live one, a later added column changes nothing in the frozen read. The same is true when the column is added while prepare() is running, since the schema is read before the offsets.

Could validate_table check only that it's still the same table (table id, lake table), and leave the schema to the frozen one? A schema id is never reused, so comparing the schema JSON isn't needed either.

Could we avoid rejecting the context when columns are added after prepare()? The read should keep using the schema captured during planning, with data mapped to it by column ID.

I've removed the live-schema equality check from validate_table() and kept the source-identity and lake-mapping checks. Readers must use the captured schema rather than the latest one; the default execution path shown in #5000 passes that schema to the core scanner for column-ID-based mapping.

Agreed. The context now remains valid when columns are added, and its captured schema stays unchanged. The reader contract requires using that schema for column-ID-based mapping, rather than reopening the read against the latest schema.

fresh-borzoni · 2026-10-27T17:27:34Z

+
+// Source preparation needs only Fluss: no lake reader, tiering job or S3 fixture.
+#[tokio::test]
+async fn prepare_freezes_table_wide_ranges_without_a_lake_backend() {


This is the only end-to-end test, and no workflow runs it, so it could break without anyone noticing. Could it run in client-integration.yml? A partitioned-table case would also help, since most of the logic is in the partition path.

I've wired test_prepare into the existing rust-integration job in client-integration.yml. The coverage includes partitioned tables with different per-partition bucket counts and scoped capture.

fresh-borzoni · 2026-10-27T17:28:44Z

+            Ok::<_, FlussLakeError>((id, identity, count, offsets))
+        },
+    ))
+    .buffered(8);


prepare() always looks up every partition, so a scan that needs one partition still pays for all of them. Table-wide makes sense as the default, since two references to the same table in one query should share one cut.
An optional partition filter would let a single scan avoid the cost, wdyt?

Could we apply safe partition pruning before fetching log offsets? For a single scan, freezing only the selected splits should be sufficient and would avoid querying unrelated partitions.

capture_read_context() remains table-wide so multiple scans can share the same boundaries. capture_read_context_with_partition_scope(&PartitionSpec) selects partitions before fetching offsets, with the scope carried in the context and applied to both lake and log inputs.

This PR now supports explicit partition scope before fetching offsets. Deriving that scope from scan predicates belongs to the subsequent planner implementation, illustrated in #5000: it extracts safe STRING partition equalities, including AND conjuncts, and keeps a broader scope when a predicate cannot be safely converted.

fresh-borzoni · 2026-10-27T17:29:11Z

+use serde::{Deserialize, Serialize};
+
+/// Stable partition identity exposed for scheduling and diagnostics.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]


If the format stays JSON, the Rust variant names ("Unpartitioned", "KeyValues") become part of it, so renaming the enum would change the format.
Maybe set explicit names, e.g. #[serde(rename_all = "snake_case")]?

I've added explicit serde names for the partition and baseline variants, so renaming a Rust variant will not change the wire format.

fresh-borzoni

one more Q

fresh-borzoni · 2026-10-27T17:32:52Z

+use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, HashSet};
+
+const CONTEXT_VERSION: u32 = 3;


+1
Why does the context version start at 3? Could we also define a compatibility and upgrade strategy? Currently, rejecting unknown fields means adding an optional field can break older readers. Compatible additions should ideally work during rolling upgrades, with version bumps reserved for incompatible changes.

The initial format is now v1; the previous number came from unpublished PoC iterations. Readers now ignore unknown optional fields when doing so preserves read semantics, while rejecting missing required fields and unknown semantic variants. Incompatible changes require a version bump, with readers upgraded before producers emit the new format. During rolling upgrades, producers must continue using the older format until all readers support the new one.

beryllw · 2026-10-29T07:17:35Z

Given the roadmap in #4713, could we open the API/interface PR first, with an end-to-end usage example? Seeing how ReadContext fits into Scan → Plan → Split → Reader would give us the full picture before reviewing the implementation details.

beryllw · 2026-10-29T08:56:26Z

We may also need to consider historical partitions here. Could we add a TODO comment or open a follow-up issue to track this?

naivedogger · 2026-11-02T12:12:15Z

Could we open the API/interface PR first, with an end-to-end usage example?

I've opened draft #5000 as a reference for the full API and usage examples, including ReadContext → Scan → Plan → Split → Reader. It includes supporting implementation beyond this PR's scope to show the complete flow; this PR remains focused on boundary capture and transport. Does this give you enough context to assess the interfaces?

naivedogger · 2026-11-02T12:12:18Z

We may also need to consider historical partitions here. Could we add a TODO comment or open a follow-up issue to track this?

Opened #5001 to track shared historical-log reads. The current design covers expired partitions retained in the lake snapshot, while historical logs still need their own boundary, pruning and reconciliation rules.

naivedogger · 2026-11-02T12:12:21Z

@fresh-borzoni @beryllw, thank you for the helpful review! The latest commit addresses your review comments. Draft #5000 provides the full API and usage examples as a reference beyond this PR's scope. I'd appreciate your thoughts on the updated boundary semantics and API design. PTAL!

naivedogger added 3 commits October 26, 2026 15:46

[rust] Add reusable UnionRead source boundaries

33fb0b3

[rust] Simplify UnionRead preparation and reuse core APIs

1dff685

[rust] Update dependency inventories for fluss-lake

36d444a

fresh-borzoni reviewed Oct 27, 2026

View reviewed changes

beryllw self-requested a review October 28, 2026 08:33

[rust] Refine UnionRead boundary capture and cross-language contracts

4222754

Conversation

naivedogger commented Oct 26, 2026

Uh oh!

naivedogger commented Oct 26, 2026

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni Oct 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

beryllw commented Oct 29, 2026

Uh oh!

beryllw commented Oct 29, 2026

Uh oh!

naivedogger commented Nov 2, 2026

Uh oh!

naivedogger commented Nov 2, 2026

Uh oh!

naivedogger commented Nov 2, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

fresh-borzoni Oct 27, 2026 •

edited

Loading