Thanks Keith, On zero-copy: agree the per-row re-encode is the wrong trade, and pushing materialisation to the client is a clean fix - it piggybacks the Arrow-to-row conversion clients already do, and since Flink, Spark and the lake reader all go through the Fluss Java client the stitch is largely write-once there, not per-connector. The one thing I'd keep in view is the splice I flagged first (keep base columns as the zero-copy file slices projection already serves, build only the enrichment column): its edge over client-side isn't CPU, it's that it changes no wire format and stays transparent to anything reading the fetch path, at the cost of a CRC pass server-side. Both are reasonable - worth picking on that wire-format/transparency question rather than on cost, since client-side's cost is smaller than it first looks.
On the disk options, the thing that worries me is common to 2 and 3: the late enrichment, whenever it lands, still goes through the local appendColumns path, and that path can't write once the base for that range has left the local disk. Whether disk is freed by the timeout escape valve (2) or by tiering base ahead and truncating (3), the moment base is gone locally appendColumns reject offsets below localLogStart and enrichment for that range can never be written. The README's retention note lands on exactly that: "deleting the local segment means enrichment data can never be written," with the fallbacks being hold-beyond-retention (disk pressure again) or accept enrichment data loss. So freeing the disk and still being able to enrich the freed range are in direct tension - the relief mechanisms work by deleting the base the enrichment needs. That's why I don't read option 2 as a disk fix, it relocates the problem - transient base-only or NULL columns in the lake, plus enrichment that may never land for timed-out ranges. Option 1 keeps completeness but takes the disk hit head-on. The way out, I think, is to stop gating disk on enrichment at all: tier base to the remote log at HW like a plain log table, so local disk is freed immediately. Because the base is then durably in remote storage, enrichment can be filled for that range at any time by reading the remote copy and writing a companion file - no local-log dependency, so no wedge, and it also removes the multi-job truncation worry, since local truncation then only depends on base tiering. The remote read/write plumbing already exists (RemoteLogDownloader on the read side, FsRemoteLogOutputStream and RemoteLogManifest on the write side), the genuinely new work is the control plane to write enrichment for an already-tiered range and track its completeness per-range in the manifest, rather than through the local-bound CEW. Given that, you can actually deliver the enriched lake the FIP is aiming at, without the disk trap. With base durable in remote, the lake materialises a range only once its enrichment has landed - the same backfill machinery, with the lake write deferred until the range is complete - so the lake ends up with every column, complete but lagging, never partial. That's the real target, and notably it does not need the Iceberg column-update, that primitive is only for making the enriched lake fresh rather than lagging, a later optimisation, not a gate on getting enriched data into the lake at all. For a table whose consumers are purely streaming and don't need enriched data at rest, base-only tiering plus live enriched reads from Fluss is the disk-safe minimum and a fine first increment - but it's the floor, not the destination, since on its own it doesn't give you the enriched lake. Option 2's timeout-fills-nulls is reasonable as an explicit opt-in for teams who'd rather have fresh-but-partial lake data, but I wouldn't make it the default or the disk answer. One framing note, +1 to Lorenzo: PK tables already do partial-column enrichment in place by targetColumns, with natural per-key parallelism and none of the CEW, merge-on-read, tiering or disk machinery. So a lot of what we're discussing is the cost of doing enrichment on append/log tables specifically. The FIP would be stronger scoping itself there explicitly - the cases where there's no key to merge on and offset-addressed replay matters - rather than as a general enrichment mechanism, since the keyed cases already have a path that avoids all of this. -- Anton вт, 23 июн. 2026 г. в 11:54, Lorenzo Affetti via dev <[email protected]>: > > Hi Keith, > > I went through FIP-45 and I think it is an amazing piece of work — the > CEW/watermark design in particular is elegant and well thought out. I have > a few questions and suggestions I'd love to discuss. > > First, a minor note on the motivation (§1): the problem statement focuses > on storage cost, but I think the stronger argument is pipeline coupling — > today, a slow enrichment job stalls all downstream consumers. Surfacing this > angle would make the design choices, especially CEW gating, even more > compelling. > > On the API side (§3), I wonder whether writing enrichment through a separate > proxy sink table is the right ergonomics. The FIP promotes the idea of > multiple independent jobs enriching different column groups of the same base > table, but the DDL requires each job to declare its own virtual table with > enrichment.target pointing back. A direct write to the base table with a > job-level 'enrichment.group' option would feel more natural and would also > let the connector handle _bucket/_offset resolution internally, avoiding > leaking those internals into user SQL. Happy to be wrong here if there is > a reason I'm missing. > > A concern I have on the write path (§2, §4): unlike standard Fluss log > writes — which are naturally multi-writer thanks to per-writer writerId and > batchSequence tracking — enrichment writes share a single global cursor per > (bucket, group). It would be good to clarify how the Flink connector is > expected to enforce single-writer-per-bucket routing, and who owns that > responsibility. > > One broader question I think would strengthen the FIP: it would be valuable > to include a short comparison with PK tables. For PK tables, the enrichment > problem is already handled natively — partial column updates via > targetColumns > let a writer fill specific columns for an existing key, with the KV store > merging them in-place, and parallelism is natural since writers operate on > independent keys. The FIP could explicitly call out why this mechanism does > not apply to log tables (no primary key, no update semantics, > offset-addressed > records) and how the proposed CEW/segment approach is the log-table > equivalent. > This would make the design feel more complete and help readers understand > the > scope of the proposal. > > Best, > Lorenzo > > On Tue, Jun 23, 2026 at 12:38 AM Keith Lee <[email protected]> wrote: > > > Hello, > > > > Thank you Giannis and Anton for your feedback. The pushback on zero-copy is > > valuable, the merge-on-read approach on the branch was built off the > > motivation of making current APIs unchanged for polling logs, I am > > convinced now that it was the wrong trade off to make and retaining > > zero-copy can be done by pushing materialisation to client side (which > > would be virtually free anyway as it is a conversion to columnar row that > > clients already do.) > > > > > LogTieringTask is untouched, there's no escape valve or backfill path > > > > This is correct. I considered making an escape valve e.g. configurable > > timeouts where timeouts result in null (we will have to force column type > > to be nullable) or sentinel values. Both of these add complexities to the > > design and neither null or sentinel values are easily distinguishable from > > true nulls or sentinel equivalent values downstream. > > > > Your input would be very useful here as this aspect of the design is about > > which trade off Fluss should make: > > > > 1. Preserve no partial row, simplicity in design and interpreting nulls: > > this is the trade off that the design current takes. You have a point that > > this story can be fleshed out further. I believe we can let users decide > > which columns are necessary to be tiered, (as an example, they can decide > > to just tier the base columns). If tiering is blocked by enrichment, the > > failure mode is in a way similar to failing tiering job. > > > > 2. Timeout and fill with nulls/sentinel value: this is an alternative so > > that tiering works and Fluss cluster storage do not get filled up if > > enrichment jobs are slow or failing. This works but downstream consumers > > will need to decide how to detect and handle nulls or sentinels from > > timeouts and the backfill workflow will have to be orchestrated by user > > outside of Fluss. > > > > 3. Same as 1. but allow for breaking up of tiering jobs and utilise tiered > > storage feature to append enrichment columns. In this model, we would build > > on 1. and also allow for multiple tiering jobs per log table for base and > > enriched parts. This model would need to use currently unavailable feature > > such as efficient column update on Iceberg [1]. There is added complexity > > in this model in deciding when can log table be truncated. As tiering is > > split up to multiple jobs, no single job can truncate rows after tiering as > > other jobs might not have tiered up to that offset yet. > > > > > > > > > what does a lake reader see for those columns until backfill lands, given > > the "no partial rows" contract? > > > > As is, lake reader will not see those columns or rows as tiering is blocked > > by enrichment, fulfilling the no partial rows contract. However, if we make > > trade off #2 mentioned earlier, they would either see nulls or sentinel > > values. > > > > I’d appreciate your thoughts on the above before I address the other > > comments or updating the design and draft implementation. > > > > Best regards > > Keith > > > > > > [1] > > > > https://docs.google.com/document/d/1Bd7JVzgajA8-DozzeEE24mID_GLuz6iwj0g4TlcVJcs/mobilebasic > > > > On Mon, 22 Jun 2026 at 04:22, Anton Borisov <[email protected]> wrote: > > > > > Hi Keith, > > > > > > Thanks for the writeup, I went through the FIP and the branch. > > > The CEW completeness contract is the right core idea. > > > > > > A few concerns, aside from what Giannis already called out about > > zero-copy. > > > Fully zero-copy isn't achievable here, but the merger re-encodes every > > > column row by row when it could keep the base columns as the zero-copy > > > file slices projection already serves and build only the enrichment > > > column - so the per-row rebuild is the costly extreme, not the floor. > > > > > > Okay, back to the points: > > > 1. Leader failover. The FIP says reads gate at CEW so a survivor > > > "never reveals enrichment that wasn't durably replicated". But > > > onBecomeNewLeader seeds CEW > > > from the promoted replica's own local EWM, and enrichment lag doesn't > > > gate ISR membership, so a replica that's in-sync on the base log but > > > behind on enrichment can > > > be cleanly promoted with CEW set above what the ISR actually holds. It > > > can also go backwards: promote a leader at EWM 10 and CEW jumps to 10, > > > it dies, an in-sync > > > follower still at EWM 4 is promoted and CEW drops to 4 and rows > > > clients already read as enriched are hidden again. > > > testNewLeaderSeedsCewFromLocalEwm encodes that over-claim as the > > > expected result. Could CEW be checkpointed the way the high watermark > > > already > > > is (checkpointHighWatermarks runs periodically, off the hot path) and > > > seeded from that on promotion? A stale checkpoint only under-claims, > > > which is safe. > > > 2. On tiering and disk: I first thought a slow enricher could wedge a > > > bucket permanently, but the companion-file backfill design handles > > > that, so - fine. > > > What's left is that none of it is in the branch yet - LogTieringTask > > > is untouched, there's no escape valve or backfill path, and > > > computeTierSafeEndOffset only caps the > > > lake split. So the disk-overflow story is unbuilt and untested in the > > > branch, and it's the heaviest part of the proposal (companion remote > > > segments, manifest rewrites, a second merge on the lake side). > > > One question while it's still on design paper: during the escape-valve > > > window, segments tier before enrichment is done - what does a lake > > > reader see for those columns until backfill lands, given the "no > > > partial rows" contract? > > > > > > A few smaller things: > > > - commitTimestamp looks like it gets dropped in the merge, the Arrow > > > builder takes no timestamp and setCommitTimestamp only runs on the > > > leader's append path, so enriched reads > > > and the lake __timestamp carry the builder default, not the > > > original. baseLogOffset does survive. Worth confirming. > > > - listOffsets(LATEST) returns CEW, not HW, for every caller on a > > > column-group table - the tiering split's CEW bound > > > (computeTierSafeEndOffset) leaks into the client's latest-offset > > > answer, so a base-only bounded scan stops short of HW. > > > Should listOffsets just return HW and keep the CEW bound to the > > > tiering generator, since enrichment reads are already gated at fetch > > > time? > > > - Giannis, on offset 0 never being enrichable - I don't think it's > > > real. The write check is source_offset == EWM and EWM starts at 0, so > > > offset 0 goes through; it's > > > the javadoc ("starts at -1") and the error string that misleads. > > > The convention already matches the high watermark. > > > > > > Let me know what you think. > > > > > > -- Anton > > > > > > пт, 19 июн. 2026 г. в 06:23, Giannis Polyzos <[email protected]>: > > > > > > > > Thanks Keith, solid proposal and the compat/CEW design is great. > > However > > > I > > > > have some concerns/blockers. > > > > > > > > Two blockers: 1. merge-on-read drops the zero-copy fetch path that > > > > projection keeps today via FileLogProjection; the per-row re-encode > > hits > > > > every enrichment projection and all SELECT *. > > > > 2. the re-encode threads no batch metadata (commitTimestamp, > > > baseLogOffset, > > > > lastOffsetDelta), corrupting offset/time-travel reads. > > > > > > > > Two things that could be addressed: The OffsetIndex is sparse and > > > > slot-indexed, but EnrichmentSegment appends per-row and treats the > > offset > > > > as the ordinal, overflowing past 2^31 with an O(batch²) walk. EWM is > > > > inconsistent (starts at 0, gate min(HWM,EWM), write at EWM+1), so > > offset > > > 0 > > > > is never enrichable; Can we use the exclusive HWM convention. > > > > > > > > Also, tier-eligibility needs all groups caught up, so a stalled job > > pins > > > > segments and fills disk. > > > > > > > > Let me know your thoughts. > > > > > > > > Best, > > > > Giannis > > > > > > > > > > > > On Tue, 16 Jun 2026 at 12:19 AM, Keith Lee <[email protected]> wrote: > > > > > > > > > Hello dev, > > > > > > > > > > I have also updated the motivation section to include the convergence > > > of > > > > > lakehouse table formats that we are seeing in this area. > > > > > > > > > > Would appreciate further feedback from the community. > > > > > > > > > > Best regards > > > > > Keith > > > > > > > > > > On Mon, Jun 15, 2026 at 10:17 PM Keith Lee <[email protected]> > > wrote: > > > > > > > > > > > Hello Zhe, > > > > > > > > > > > > Thank you for your review, I have incorporated changes based on > > your > > > > > > feedback; I appreciate it as they make the proposal more robust. > > > > > > > > > > > > Best regards > > > > > > Keith > > > > > > > > > > > > On Fri, Jun 12, 2026 at 4:29 AM Zhe Wang <[email protected]> > > > wrote: > > > > > > > > > > > >> Hi Keith, > > > > > >> > > > > > >> Thanks for the detailed FIP. I read through the proposal and the > > > overall > > > > > >> direction makes sense to me. I have two questions/suggestions > > about > > > the > > > > > >> Flink SQL user-facing part and the column-group metadata. > > > > > >> > > > > > >> 1. Literal-only enrichment SELECT > > > > > >> > > > > > >> The caveat around literal-only enrichment expressions looks a bit > > > > > >> surprising to users: > > > > > >> > > > > > >> INSERT INTO device_logs_geo_sink > > > > > >> SELECT _partition, _bucket, _offset, 'London' > > > > > >> FROM device_logs; > > > > > >> > > > > > >> According to the FIP, this can silently emit zero rows because no > > > base > > > > > >> column is referenced, so the connector may fall back to a > > projection > > > > > that > > > > > >> is clamped by the enrichment gate. > > > > > >> > > > > > >> Would it be possible to make this fail at planning time with a > > clear > > > > > >> validation error, or have the source side automatically include a > > > > > minimal > > > > > >> base-column projection when metadata columns are requested for > > > > > enrichment? > > > > > >> Silent zero-output behavior feels hard to diagnose in production, > > > > > >> especially because the SQL statement looks valid and deterministic > > > from > > > > > >> the > > > > > >> user's point of view. > > > > > >> > > > > > >> 2. Column group name validation > > > > > >> > > > > > >> Column group names appear in multiple places: > > > > > >> > > > > > >> - DDL option keys, e.g. `column-groups.enriched_geo` > > > > > >> - Java API values, e.g. `columnGroup("enriched_geo", ...)` > > > > > >> - RPC fields > > > > > >> - physical segment files, e.g. `.col.<group>.log` > > > > > >> > > > > > >> Should the FIP explicitly define the allowed format for column > > group > > > > > >> names? > > > > > >> For example, whether names may contain `.`, `/`, spaces, uppercase > > > > > >> characters, or characters that need escaping. > > > > > >> > > > > > >> A small validation rule such as `[A-Za-z_][A-Za-z0-9_]*` or > > another > > > > > >> clearly > > > > > >> documented convention would make DDL round-tripping, file naming, > > > and > > > > > >> error > > > > > >> messages easier to reason about. > > > > > >> > > > > > >> Best regards, > > > > > >> Zhe Wang > > > > > >> > > > > > >> > > > > > >> Keith Lee <[email protected]> 于2026年5月25日周一 22:49写道： > > > > > >> > > > > > >> > Hello devs, > > > > > >> > > > > > > >> > I have raised a proposal on Log Enrichment via Append Columns > > [1]. > > > > > >> > > > > > > >> > A common Fluss enrichment pipeline today looks like the > > following: > > > > > >> > > > > > > >> > edge device --> log table A --> Flink job (enrichment) --> log > > > table B > > > > > >> --> > > > > > >> > consumer > > > > > >> > > > > > > >> > If A has 50 columns and the enrichment adds 3, the 50 base > > > columns are > > > > > >> > stored twice - once in A, once in B. At scale this is real > > > operational > > > > > >> and > > > > > >> > infrastructure cost - 2x storage, 2x write I/O, 2x fetch I/O for > > > any > > > > > >> > downstream that consumes B plus the additional architectural and > > > > > >> > operational complexity of running and maintaining a two > > pipelines. > > > > > >> > > > > > > >> > We propose a first-class primitive that lets the same row gain > > > columns > > > > > >> over > > > > > >> > time, with the system enforcing "completeness" at the read > > > boundary. > > > > > >> With > > > > > >> > this proposal, enrichment job(s) writes enrichment columns back > > > to the > > > > > >> same > > > > > >> > log table A via appendColumns; there is no log table B. > > Different > > > > > >> > enrichment tasks can also happen at different cadence e.g. > > > streaming > > > > > for > > > > > >> > GeoIP lookup, batch for vector embedding. > > > > > >> > > > > > > >> > Base columns are stored once and enrichment columns are stored > > > once > > > > > (per > > > > > >> > column group). Consumers project whatever they need against the > > > single > > > > > >> > table: projections that touch only base columns advance up to > > high > > > > > >> > watermark (HWM) as today; projections that touch enrichment > > column > > > > > >> group g > > > > > >> > are clamped by enrichment water mark (EWM) at min(HWM, EWM_g), > > so > > > > > >> partial > > > > > >> > rows are never observed and the system not the consumer owns the > > > "is > > > > > >> > enrichment caught up?" decision. > > > > > >> > > > > > > >> > There are more details on the FIP and POC implementation [2]. > > > > > >> > Looking forward to your comments > > > > > >> > > > > > > >> > Best > > > > > >> > Keith > > > > > >> > > > > > > >> > References > > > > > >> > [1] > > > > > >> > > > > > > >> > > > > > > >> > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLUSS/FIP-45%3A+Log+Enrichment+via+Append+Columns > > > > > >> > [2] > > > > > >> > > > > > > >> > > > > > > > > > > https://github.com/leekeiabstraction/fluss/tree/option02-lateMaterialized > > > > > >> > > > > > > >> > > > > > > > > > > > > > > > > > > > -- > Lorenzo Affetti > Senior Software Engineer @ Flink Team > Ververica <http://www.ververica.com>
