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
> > >> >
> > >>
> > >
> >
