[junmuz]

Preserve correct before-images in changelog retraction records and append event values as configurable metadata columns, accessible via Flink SupportsReadingMetadata and Spark table schema.

Purpose

• Added support for reading the metadata through Flink using <prefix><field> metadata columns
  such as Cassandra WRITETIME.
• Updated Spark and core read paths to include and populate the synthetic metadata fields across raw and merged file reads.
• Added validation for metadata projections, file reads, filters, and lookup changelog generation.

Tests

• Added unit coverage for insert, update, delete, before-image, filter, aggregation, and unconfigured cases.
• Added Flink integration coverage for reading event metadata and preserving correct physical before-images.
• paimon-docs verify passes on Java 8 and Java 11.
• Clean Java 11 Maven installation completed successfully across all 76 modules.

[junmuz]

@JingsongLi Can I get a review on this?

[JingsongLi]

Thanks for the feature. I reviewed the write and read paths end to end and found two blocking issues plus a consistency concern.

1. [P1] The option is applied for every changelog producer, but only the lookup path fills the extra fields

MergeTreeCompactManagerFactory's constructor widens the changelog value type whenever the option is set:

RowType changelogValueType = computeChangelogValueType(valueType, options);
if (changelogValueType != null) {
    writerFactoryBuilder.withChangelogValueType(changelogValueType);
}

There is no changelog-producer == lookup check here, and the option is not validated anywhere else (docs/description only say "Only valid when changelog-producer is lookup"). The extra fields are actually populated only by the lookup merge-function wrapper (preserveFieldIndices is only passed to LookupMergeFunctionWrapperFactory inside the lookup branch of createRewriter).

Because the same KeyValueFileWriterFactory.Builder instance is shared with KeyValueFileStoreWrite (the compact-manager factory is created in KeyValueFileStoreWrite's constructor, and createWriter later calls writerFactoryBuilder.build(...)), the merge-tree writer's changelog format context is widened to N+K for all producers:

• changelog-producer='input': MergeTreeWriter flushes changelog records built from the incoming values (MergeTreeWriter.java:219-220 -> createRollingChangelogFileWriter) whose value rows have N fields, but the writer now expects N+K value fields.
• changelog-producer='full-compaction': same for ChangelogMergeTreeRewriter.java:147.

Result: the flush/compaction fails while serializing the shorter row (e.g. AIOOBE in the row writer), or worse, corrupt changelog files are written. Please gate the widening on options.changelogProducer() == ChangelogProducer.LOOKUP (and/or reject the option at schema validation for other producers).

2. [P1] The widened Spark DSv2 schema breaks ordinary Spark writes to such tables

PaimonSparkTableBase.schema (and BaseScan.tableRowType) add the K metadata columns to the DSv2 table schema. Paimon tables advertise ACCEPT_ANY_SCHEMA, so Spark's own ResolveOutputRelation is skipped and writes are resolved by Paimon's PaimonAnalysis -> PaimonOutputResolver. For a by-position write, SchemaEvolutionHelper.expectedAttrsForCatalogWrite returns table.output (now N+K columns), and PaimonOutputResolver.resolveColumnsByPosition throws on the size mismatch:

Cannot write to `t`, the number of data columns (3) doesn't match the table schema's (4).

So on a table with changelog-producer.expose-field-as-metadata set, plain Spark SQL writes such as INSERT INTO t SELECT id, data, event_ts FROM src or INSERT INTO t VALUES (...) fail, even though the metadata columns are generated rather than user input - the new test itself says "it is not an input column". By-name writes happen to survive (top-level NULL fill), which makes the failure mode even more confusing.

Additionally, INSERT INTO t SELECT * FROM t produces N+K columns, which resolves positionally; with spark.paimon.write.merge-schema enabled, SchemaEvolutionHelper.commitSchemaEvolution will merge __internal__event_ts into the persisted table schema, because SparkSystemColumns only filters _bucket_/_row_kind_.

Suggested fix: keep the generated metadata columns out of write-side resolution, e.g. filter them in expectedAttrsForCatalogWrite/SparkSystemColumns, or expose them only on the scan side while keeping PaimonSparkTableBase.schema at the physical N columns.

3. [P2] Inconsistent unknown-column handling and duplicated extension logic

• MergeTreeCompactManagerFactory.computeChangelogValueType throws IllegalArgumentException for a configured column that does not exist, while KeyValueFileStore.newReaderFactoryBuilder and ChangelogEventMetadataTable.computeExtendedRowType silently skip it (if (idx >= 0)). A typo in the option therefore fails at write/compaction time but is ignored on the read path (metadata column silently NULL) - please pick one behavior, ideally validating at schema/option level.
• The "append prefix + name fields with max(id) + 1 and nullable type copy" logic is duplicated in MergeTreeCompactManagerFactory (twice), KeyValueFileStore, ChangelogEventMetadataTable.computeExtendedRowType and again in a different variant in MergeFileSplitRead/RawFileSplitRead. These must stay in sync for the reader mapping (createMetadataFallbackMapping) to line up; a single helper would be safer.

Minor: for changelog files written before the option was enabled, the metadata columns read as NULL (the fallback mapping is only applied to data files, via !isChangelogFile(file)). Worth one sentence in the docs if that is intended.

[JingsongLi]

Given the significant nature of the changes, I need specific use cases to justify their validity; please provide a detailed explanation of how this would be implemented in a real-world production environment.

[junmuz]

Given the significant nature of the changes, I need specific use cases to justify their validity; please provide a detailed explanation of how this would be implemented in a real-world production environment.

@JingsongLi Let me explain this in detail below. Yeah I agree the changes are more complex that's why I initially went with minimal change approach.

Use case and motivation

Paimon’s lookup changelog producer reconstructs UPDATE_BEFORE and DELETE records by looking up the previous table state. As a result, regular columns in those records describe the old row, not necessarily the event that caused the change. For example:

UPDATE_BEFORE: event_ts = 50, __internal__event_ts = 100
UPDATE_AFTER: event_ts = 100, __internal__event_ts = 100

Downstream CDC consumers often need the event timestamp or source sequence to apply changes deterministically.

Concurrent Sinks to a Single Cassandra table

Multiple sources can fan into the same Cassandra table while being processed by independent jobs. A common example is migrating from Kafka to Paimon: Kafka retains the original event payload for retract events, whereas Paimon’s lookup-generated changelog may represent UPDATE_BEFORE and DELETE records using the previous row image. This makes it difficult to replace a Kafka-based job with a Paimon-based job without changing downstream conflict-resolution behavior. Exposing selected fields as changelog metadata preserves the event-level values and gives the sink a consistent, globally comparable event version for conditional writes and tombstones, preventing stale updates from overwriting newer changes or resurrecting deleted records. This behavior allows creating Sinks from multiple Paimon tables merging into a single Cassandra column.

[JingsongLi]

The Cassandra WRITETIME use case gives this feature concrete end-to-end value. The earlier lookup-only validation, physical before-image preservation, and Spark write-side filtering are present on this head. I found one remaining schema issue before this should ship:

P2 — Allocate synthetic metadata field IDs above nested field IDs (ChangelogEventMetadata.java:117). extraValueFields takes the maximum top-level field ID. A valid table can have id (0), payload ROW<nested INT> (top-level 1, nested field ID 3), and event_ts (2). Exposing event_ts assigns the new __internal__event_ts field ID 3, duplicating payload.nested. I added this case to ChangelogEventMetadataTest in an isolated checkout: appendMetadataFields(...).collectFieldIds(...) fails with Broken schema, field id 3 is duplicated. The generated row type is used by the changelog writer and both read integrations, so its IDs need to remain valid for nested schemas and field-ID based mapping. Please derive the next ID from the full recursive field-ID set (for example, RowType.currentHighestFieldId(valueType.getFields()) + 1) and test a nested table through a changelog read.

Validation: the added core tests passed 16/16 and Spark 3 PaimonCDCSourceTest passed 6/6 on 5ec5070; the strengthened nested-ID test fails as above. I attempted the 3 Flink integration tests locally, but all stopped on the local Avro classpath mismatch DataFileWriter.setEncoder before exercising this change. The PR's Flink and other CI jobs are green on this head.

[junmuz]

The Cassandra WRITETIME use case gives this feature concrete end-to-end value. The earlier lookup-only validation, physical before-image preservation, and Spark write-side filtering are present on this head. I found one remaining schema issue before this should ship:

Thanks a lot for pointing it out. I have made that fix.

[JingsongLi]

Re-reviewed head b0fd1ea for production. The Cassandra WRITETIME use case supplies clear end-to-end value. The latest fix allocates synthetic changelog metadata IDs above the full recursive physical field-ID set, addressing my prior nested-field collision; it also adds a Flink SQL read/write regression with a nested ROW and metadata alias.

Local JDK 8 core suites passed: ChangelogEventMetadataTest 2/2, SchemaValidationTest 72/72, and LookupChangelogMergeFunctionWrapperTest 13/13. I attempted the four Flink integration cases locally, but each failed at its initial manifest write before the changed logic because this isolated checkout loads an incompatible Avro jar (DataFileWriter.setEncoder NoSuchMethodError), the same local classpath issue observed in the earlier review. Exact-head Flink 1/2, Spark, JDK 8/11, E2E, docs and licensing CI are all green. git diff --check passed. I found no remaining code blocker; keep this PR open for merge.

[JingsongLi]

I took another look at head b0fd1ea, focusing on the public semantics and schema evolution. The Cassandra WRITETIME use case is valid, and preserving the physical before-image is the right direction. I need to revise my earlier no-blocker assessment: I found two correctness issues with the current metadata surface.

1. [P1] Event metadata breaks retract semantics when used by normal SQL operators.

LookupChangelogMergeFunctionWrapper writes the new event's metadata into the -U record while its physical columns correctly contain the old row. For example, an initial +I(event_ts=50, metadata=50) followed by an update to 100 produces -U(event_ts=50, metadata=100). In Flink, WHERE metadata < 75 forwards the insert but drops its retraction, leaving a stale row. Grouping or aggregating by the metadata column has the same problem. The tests cover filtering on the physical event_ts, but not on the exposed metadata column. This is intrinsic to treating an event attribute as an ordinary column in a retracting dynamic table; please define an event-oriented consumption surface (or another mechanism that prevents relational operators from interpreting it as before-image data) and add a regression test for this sequence.

2. [P1] Changing changelog-producer.metadata-field-prefix loses historical metadata.

The prefix is not marked immutable, so it can be changed after changelog files exist. The write path stored the old prefixed field name, but ChangelogEventMetadata.extraValueFields and the reader construct only the current name for every file schema. After changing __internal__ to __event__, older changelog files have no __event__event_ts field, so historical event metadata reads as NULL; the data-file fallback is intentionally disabled for changelog files. Please keep the on-disk identity stable across option changes, or reject prefix changes once files exist. A write → ALTER prefix → read-old-changelog test would make the contract clear.

There is also a source/sink mismatch in the documentation example: METADATA FROM '__internal__event_ts' needs to be declared on the Paimon source. An external sink's METADATA FROM key belongs to that sink and works only if the sink advertises that writable key. Please show the source alias and an explicit mapping to the sink's actual timestamp input.

Finally, the option names obscure the contract. expose-field-as-metadata persists a list of post-merge event values; it does more than expose a read-time field and does not always preserve the raw incoming value. A name such as changelog-producer.event-metadata-fields would be clearer. The configurable prefix currently couples the on-disk field name, Flink metadata key, and Spark-visible column name. I suggest separating the stable storage identity from engine-specific presentation. Given that the concrete use case is a Flink-to-Cassandra flow, I would also consider landing core/Flink support first and adding the Spark schema/write-path changes with a separate demonstrated consumer.

[junmuz]

@JingsongLi Thanks for the review.

1. I see your point, and I am of the opinion if someone uses it inside for filtering and aggregation then it is an incorrect configuration. I have also updated the documentation to not use the metadata field with aggregation or filtering. Also, it is quite difficult unless someone chooses to use it that way, because now they have to setup additional table property as well as exposing the metadata column. We have similar behavior with sequence fields as well if someone configures ts as sequence field, and uses `SELECT MAX(ts) as ts from source_table; then retraction doesn't work.

2. I’ve made changelog-producer.metadata-field-prefix immutable. The stored changelog field uses the prefix with the source field ID, so its physical identity stays stable across source field renames. Spark and Flink can use the same logical metadata name through their respective APIs (Spark as a column and Flink as a metadata key) without engine-specific names.

[JingsongLi]

Reviewed current head 2b5f23f8289ddc7b07c6deec9a728791da452b21. The Cassandra event-version use case has end-to-end value. The immutable prefix, source-field-ID storage names, rename handling, and shared write-side filtering address the earlier problems. I found two new historical-schema issues and one production contract gap.

[P1] Dropping an ordinary column silently replaces event metadata with that column's historical value. ChangelogEventMetadata.metadataValueFields allocates IDs above the remaining row type's maximum (line 253), which can reuse a dropped physical field ID. Reproduction with real Parquet and Avro tables: physical schema (id INT, data INT, event_ts BIGINT, extra BIGINT), expose event_ts; insert (1,10,50,777), then update (1,20,100,888). Before ALTER ... DROP extra, metadata is 50/100/100 for +I/-U/+U. After the successful DROP, it is 777/777/888; latest batch reads return physical event_ts=100 but metadata 888. Physical-only reads remain correct. The table's historical highest field ID remains 3, but the new metadata ID also becomes 3. A control allocating non-colliding metadata IDs preserves the correct values in both formats. Please ensure synthetic IDs/read mappings cannot alias historical physical fields, and add a real DROP/read regression. This can send an incorrect WRITETIME to the downstream store without an exception.

[P2] Avro historical metadata loses the source-type conversion after a legal ALTER. For an Avro lookup table exposing event_ts BIGINT, write values 50 and 100, then ALTER it to DECIMAL(20,0). Historical physical values still read correctly, but metadata reads throw ClassCastException: Long cannot be cast to Decimal. KeyValueFileReaderFactory lines 434–441 appends the current extraFields to each historical schema, so the metadata is treated as DECIMAL on disk and misses the BIGINT→DECIMAL cast. Constructing the historical metadata fields using their historical source-field type makes the same reproduction pass. Please cover this in both raw and merged reader mappings, with an actual historical-file test.

[P2] The documented pass-through support boundary must include automatic sink materialization. A plain INSERT INTO event_sink SELECT data,writetime FROM source_table, with source PK id and an external upsert sink PK data, inserts Flink's default AUTO upsert materializer. For (1,10,50) followed by (1,20,100), it cannot match -U(10,100) to stored (10,50), so old sink key 10 remains. There is no metadata filter or aggregation. I reproduced this against Flink's values upsert sink, with a later marker proving catch-up. NONE produces the correct final sink; keeping AUTO but projecting physical event_ts also correctly removes old key 10. Flink TRY_RESOLVE rejects the unsafe metadata plan. The warning at docs lines 118–120 currently excludes only filters/aggregations; please state the compatible key/plan requirements and necessary safeguards, and add an actual downstream-sink regression. Flink's event-oriented CDC metadata API is a defensible contract, so I am not repeating the previous metadata-filter example as a new blocker.

Validation: 147 Core tests passed on JDK 8; all four existing Flink integration cases passed on Flink 1.20.1 and 2.2.0 after aligning the local Avro runtime; Spark 3.5.8 CDC suite passed 6/6. Spark 4.1.2 CDC suite passed 6/6 with normal Maven checks. Independent actual persisted tests passed scalar rename, ordinary-column addition, metadata-field addition/reordering, and nine ROW/ARRAY/MAP × Parquet/Avro/ORC full-row and metadata-only projections. Exact-head CI is green, but the new schema-evolution probes above fail on this head and pass with isolated controls. No shared source changes were retained.

[junmuz]

@JingsongLi Thanks for the detailed reproductions. All three are addressed as of 36ba0fb.

1. [P1] Drop aliasing: Metadata field IDs are now allocated above TableSchema#highestFieldId(), so they never reuse a dropped field's ID. ChangelogEventMetadataSchemaEvolutionTest reproduces your Parquet/Avro DROP scenario on both the merged and raw read paths.

2. [P2] Historical type cast: Each file's metadata fields now use the source column's type from that file's schema, so schema evolution casts them like the source column. testSourceTypeChangeCastsHistoricalMetadata covers your BIGINT → DECIMAL case on Parquet/Avro.

3. [P2] Sink materialization: The docs now explain why retractions with metadata cannot be matched by full-row comparison, and require keeping sink and source keys equal or using upsert-materialize=NONE, with TRY_RESOLVE as a safeguard. The example shows the explicit source-to-sink mapping. LookupChangelogEventMetadataITCase covers both the NONE and TRY_RESOLVE cases.

One behavior change: renaming or dropping a column listed in changelog-producer.event-metadata-fields is now rejected, because it would silently change metadata names that downstream jobs reference. Remove the column from the option first.

On Spark scope: I'm happy to split the Spark changes into a follow-up PR if you'd still prefer that.

[JingsongLi]

Re-reviewed this head for production. The Cassandra WRITETIME use case has end-to-end value, and the prior dropped-ID aliasing, historical type conversion, and documented sink-materialization fixes are present. Local JDK 8 package verification without fast-build passed 165 core, 6 Flink SQL integration, and 6 Spark CDC tests. Three remaining issues are below. Added real-file probes reproduce nested projection corruption on both Parquet and Avro, and a real Flink temporal join fails while its physical-only control passes. The exact-head E2E CI also fails during POM parsing on the duplicate Avro dependency. Spark 4.1 CI additionally has a concurrent data-evolution compaction snapshot-expiration failure; I have not established that it is caused by this change.

[JingsongLi]

[P2] Project nested fallback values to the requested metadata shape

Adding the whole source field here is unsafe when metadata is nested-pruned. With payload ROW<a INT,b INT> and event-metadata-fields=payload, requesting only __internal__payload.b gives a metadata type ROW<b INT>, but FallbackMappingRow.getRow returns the full physical ROW<a,b> unchanged. A real table updated to (a=33,b=44) returns 33 instead of 44 for that metadata child. I reproduced this with Parquet and Avro on the raw path and with the merge reader forced via forceKeepDelete(). Spark nested pruning produces this requested type. MergeFileSplitRead.adjustReadType has the same dependency logic. Please combine physical/metadata dependency projections and project each fallback value to its requested nested shape, including when the physical source and metadata request different children.

[JingsongLi]

[P2] Apply declared metadata to temporal lookup results too

The scan path consumes these metadata keys, but the inherited getLookupRuntimeProvider still passes the physical FileStoreTable and physical projection to FileStoreLookupFunction. Flink accepts metadata on a lookup table and expects the produced metadata column in each returned row. Reproduction: declare (id INT PRIMARY KEY NOT ENFORCED, event_ts BIGINT, writetime BIGINT METADATA FROM '__internal__event_ts' VIRTUAL), preserve event_ts, insert (1,50), and run a processing-time temporal join selecting d.event_ts,d.writetime. The actual job fails with ArrayIndexOutOfBoundsException: 2 in GenericRow.isNullAt through the generated lookup collector because only the two physical fields are returned. The same join selecting only d.event_ts passes. Please supply the requested metadata in the lookup runtime (or explicitly reject the unsupported lookup plan before launching a job) and add this integration case.

[junmuz]

Thanks for the suggestions. I have tried addressing the feedback in 5a99846 (also rebased on master)

1. [P2] Nested fallback shape: Both the raw and merge readers now read the source field with the union of the nested children that it and its metadata field need, then trim each fallback value to the requested metadata shape. This covers the case where the physical column and the metadata request different children. ChangelogEventMetadataNestedReadTest covers Parquet/Avro/ORC on the raw path and on the merge path via forceKeepDelete().

2. [P2] Temporal lookup join: I went with rejecting the plan. A lookup only sees the latest table state, so it can't return per-event metadata. DataTableSource.getLookupRuntimeProvider now fails with a clear error when metadata columns are requested, and lookups of physical columns still work. Both cases are covered in LookupChangelogEventMetadataITCase.

3. [P2] Duplicate Avro dependency: Removed.

While testing, I also fixed incremental-between-scan-mode=diff reads, which failed with Cannot find field index by FieldId when metadata fields were requested.

I have also raised a different PR that preserves the field with additional configuration option. That's much more simpler, but it comes at the expense of breaking CDC contract (which should be ok IMO as long as we clearly document it).

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS. The previous nested-ROW, temporal-lookup rejection and duplicate-dependency fixes are present. Regular JDK 8 verification passed 186 Core metadata/schema/merge tests, 7 real Flink SQL integration tests and 6 Spark CDC tests. The existing IncrementalTableTest has 14 passes and one reproducible regression matching exact-head JDK 8/11 CI. Additional real-file and Spark SQL probes, with controls and independent challenges, expose four issues below. The MAP probe uses the exact current Spark 3.5 engine rule because the common UT classpath otherwise loads its no-op placeholder.

[JingsongLi]

[P1] Compute diff membership before applying the requested projection

Pushing the requested columns into both merge readers makes DiffMerger compare only those columns. A key whose unselected value changed is therefore dropped as unchanged. This also affects ordinary PK tables without event metadata: after committing (id=1,v=10), updating to v=20 and full-compacting, the same incremental diff returns [1,20] with the full schema but no rows when projected to id; the merge-base implementation returns [1] for the latter. The existing IncrementalTableTest.testTagIncremental also fails locally and in both exact-head JDK 8/11 CI jobs (expected [2,6], actual [6]). Please retain all values needed to decide whether the row changed and project only after computing the diff, while adding metadata dependencies separately.

[JingsongLi]

[P2] Populate ordinary-file fallback metadata after merging values

This copies metadata from each input file before the merge engine runs. For an aggregation table, the synthetic field does not inherit its source's aggregation function, so it is merged using the default last_non_null_value. With legal lookup/event-metadata=amount, fields.amount.aggregate-function=sum and write-only=true, two commits of amount 10 and 20 return an ordinary snapshot row amount=30, __internal__amount=20. Full compaction changes only the file layout and then the same row reads 30,30; the last_non_null_value control remains 20,20 throughout. Independent real-file runs also reproduce it with metadata-only projection. Please derive fallback values from the final merged physical row (or preserve the source merge semantics), while retaining the actual event metadata stored in changelog files.

[JingsongLi]

[P2] Retain a full MAP dependency beside selected-key physical reads

Returning the physical selected-key ROW here leaves a full MAP metadata field backed by that ROW. With a shared-shredding payload MAP and event-metadata-fields=payload, real Spark 3.5 SQL SELECT payload['a'], __internal__payload FROM T pushes PushedMapSelectedKeys:[payload=[a]] and then fails with ClassCastException (CastedRowColumnVector to MapColumnVector) through FallbackMappingRow.getMap. Selecting the full physical map plus metadata, or only payload['a'], both succeeds. I reproduced this on real compacted Parquet files using the exact current Spark 3.5 optimizer implementation, in addition to a Core read-type probe. Please retain an independent full source-map dependency or prevent this selected-key pruning when full metadata needs that source, rather than reusing the incompatible ROW fallback.

[JingsongLi]

[P2] Keep synthetic metadata out of MERGE target assignment alignment

Exposing these read-only fields through schema also adds them to DataSourceV2Relation.output, but the MERGE resolver still expands and aligns assignments against that full output. Under the default merge-schema=false, an ordinary physical-only source now fails both WHEN MATCHED THEN UPDATE SET * and WHEN NOT MATCHED THEN INSERT * with "source is missing" __internal__event_ts. Actual Spark SQL controls show that explicit physical assignments succeed on the same metadata-enabled table, and both stars succeed on the same lookup table without this option. The new INSERT/V2 write filtering happens later and does not cover MERGE analysis. Please align MERGE stars against physical target fields while continuing to expose metadata for reads.