You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "junmuz (via GitHub)" <gi...@apache.org> on 2026/09/14 12:35:36 UTC
[PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
junmuz opened a new pull request, #9822:
URL: https://github.com/apache/paimon/pull/9822
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
### Tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5832651229
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4254417732
##########
paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java:
##########
@@ -387,8 +412,99 @@ private FileRecordReader<InternalRow> createFileReader(
}
if (deletionVector != null && !deletionVector.isEmpty()) {
- return new ApplyDeletionVectorReader(fileRecordReader, deletionVector);
+ fileRecordReader = new ApplyDeletionVectorReader(fileRecordReader, deletionVector);
}
- return fileRecordReader;
+ return applyMetadataFallbackAndOuterProjection(fileRecordReader);
+ }
+
+ private RowType readTypeWithMetadataDependencies(RowType requestedReadType) {
+ if (changelogExtraValueFields.isEmpty()) {
+ return requestedReadType;
+ }
+
+ List<String> readFieldNames = requestedReadType.getFieldNames();
+ List<DataField> dependencies = new ArrayList<>();
+ for (String preserveColumn : metadataPreserveColumns) {
+ String metadataName = metadataFieldPrefix + preserveColumn;
+ if (readFieldNames.contains(metadataName)
+ && !readFieldNames.contains(preserveColumn)
+ && schema.logicalRowType().containsField(preserveColumn)) {
+ dependencies.add(schema.logicalRowType().getField(preserveColumn));
Review Comment:
[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.
##########
paimon-flink/paimon-flink-common/pom.xml:
##########
@@ -190,6 +190,13 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
Review Comment:
[P2] Remove the duplicate Avro dependency
This new test dependency exactly duplicates the existing unconditional `org.apache.avro:avro:${avro.version}` test dependency at lines 108–113. The exact-head [Flink/Spark E2E CI](https://github.com/apache/paimon/actions/runs/37482976432/job/112339853634) stops before compilation/tests with `dependencies.dependency ... must be unique ... duplicate declaration ... @ line 193`. Please retain one declaration so the E2E build can run.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java:
##########
@@ -173,25 +179,61 @@ public RowLevelModificationScanContext applyRowLevelModificationScan(
}
public Map<String, DataType> listReadableMetadata() {
- // Flink calls this after applyRowLevelModificationScan for row-level operations.
- if (rowLevelModificationSnapshotId == null || !isDataEvolutionTable()) {
- return Collections.emptyMap();
- }
Map<String, DataType> metadata = new LinkedHashMap<>();
- metadata.put(SpecialFields.ROW_ID.name(), DataTypes.BIGINT().notNull());
+
+ // Row-level modification metadata
+ if (rowLevelModificationSnapshotId != null && isDataEvolutionTable()) {
+ metadata.put(SpecialFields.ROW_ID.name(), DataTypes.BIGINT().notNull());
+ }
+
+ // Event metadata from changelog-producer.event-metadata-fields
+ List<String> preserveColumns = eventPreserveColumns();
+ if (!preserveColumns.isEmpty() && table instanceof FileStoreTable) {
+ TableSchema schema = ((FileStoreTable) table).schema();
+ CoreOptions coreOptions = CoreOptions.fromMap(table.options());
+ for (DataField field :
+ ChangelogEventMetadata.extraValueFields(
+ schema.logicalRowType(), schema.highestFieldId(), coreOptions)) {
+ DataType flinkType =
+ TypeConversions.fromLogicalToDataType(
+ LogicalTypeConversion.toLogicalType(field.type()));
+ metadata.put(field.name(), flinkType.nullable());
+ }
+ }
+
return metadata;
}
public void applyReadableMetadata(List<String> metadataKeys, DataType producedDataType) {
for (String metadataKey : metadataKeys) {
- if (!SpecialFields.ROW_ID.name().equals(metadataKey)) {
- throw new UnsupportedOperationException(
- "Unsupported Paimon metadata column: " + metadataKey);
+ if (SpecialFields.ROW_ID.name().equals(metadataKey)
+ || eventMetadataFieldNames().contains(metadataKey)) {
+ continue;
}
+ throw new UnsupportedOperationException(
+ "Unsupported Paimon metadata column: " + metadataKey);
}
this.metadataKeys = metadataKeys;
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4039490104
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogEventMetadataITCase.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for lookup changelog event metadata. */
+public class LookupChangelogEventMetadataITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testEventMetadataCanBeReadAsWriteTime() throws Exception {
+ sql(
+ "CREATE TABLE source_table ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "data INT, "
+ + "event_ts BIGINT, "
+ + "writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"
Review Comment:
What does this mean?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5738464956
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:
```java
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-6019029931
@JingsongLi Thanks for the detailed reproductions. All three are addressed as of 36ba0fba5.
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4039483545
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogEventMetadataITCase.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for lookup changelog event metadata. */
+public class LookupChangelogEventMetadataITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testEventMetadataCanBeReadAsWriteTime() throws Exception {
+ sql(
+ "CREATE TABLE source_table ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "data INT, "
+ + "event_ts BIGINT, "
+ + "writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"
+ + ") WITH ("
+ + "'bucket'='1', "
+ + "'changelog-producer'='lookup', "
+ + "'sequence.field'='event_ts', "
+ + "'changelog-producer.expose-field-as-metadata'='event_ts')");
+
+ // The metadata column models a Cassandra sink column populated from WRITETIME. The
+ // physical event_ts column remains available to normal Flink operators, while writetime
+ // carries the incoming event value even on an UPDATE_BEFORE retraction.
+ BlockingIterator<Row, Row> iterator =
+ streamSqlBlockIter("SELECT id, data, event_ts, writetime FROM source_table");
+
+ sql("INSERT INTO source_table VALUES (1, 10, 50)");
+ assertThat(iterator.collect(1)).containsExactly(Row.of(1, 10, 50L, 50L));
+
+ sql("INSERT INTO source_table VALUES (1, 20, 100)");
+ assertThat(iterator.collect(2))
+ .containsExactly(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, 10, 50L, 100L),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 20, 100L, 100L));
+
+ iterator.close();
+ }
+
+ @Test
+ public void testPhysicalFilterStillSeesBeforeImage() throws Exception {
+ sql(
+ "CREATE TABLE filtered_source ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "data INT, "
+ + "event_ts BIGINT, "
+ + "writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"
Review Comment:
What happens if the table is created from Flink and read from Spark?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5833425147
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4035699057
##########
docs/docs/primary-key-table/changelog-producer.md:
##########
@@ -109,6 +109,43 @@ checkpoint completion waits for compaction. Tune it with checkpoint duration and
`lookup` is incompatible with `full-compaction.delta-commits`. For periodic full compaction with
changelog generation, use `full-compaction` instead.
+By setting `'changelog-producer.expose-field-as-metadata'` to a comma-separated list of column names,
+the lookup changelog producer stores those columns' values from the incoming event as additional metadata
+columns (`__internal__<column>`) in all changelog records. For retraction records (`-U`, `-D`), the regular
+value columns retain the correct before-image so standard downstream operators (filters, aggregations)
+work correctly, while the metadata columns carry the event values. For forward records (`+I`, `+U`),
+the metadata columns mirror the regular values for schema consistency. Sinks that need the event
+timestamp — such as Cassandra using `WRITETIME` for conflict resolution — can read the metadata columns.
+
+**Note:** The event values come from the *merged result* of the merge function, which equals the raw
+incoming event for the `deduplicate` merge engine but may differ for aggregation merge engines.
+
+This option is only supported by the `lookup` changelog producer.
+
+```sql
+-- Source table with event metadata preservation
+CREATE TABLE my_table (
+ id INT PRIMARY KEY NOT ENFORCED,
+ data STRING,
+ event_ts BIGINT
+) WITH (
+ 'changelog-producer' = 'lookup',
+ 'sequence.field' = 'event_ts',
+ 'changelog-producer.expose-field-as-metadata' = 'event_ts'
+);
+
+-- Sink table reading event metadata
+CREATE TABLE cassandra_sink (
+ id INT,
+ data STRING,
+ event_ts BIGINT,
+ retract_event_ts BIGINT METADATA FROM 'paimon.event.event_ts'
+) WITH (...);
+```
+
+(Note: Please increase `'execution.checkpointing.max-concurrent-checkpoints'` Flink configuration, this is very
Review Comment:
This mention of max-concurrent-checkpoints may not be needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5725649052
@JingsongLi Can I get a review on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4039490104
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogEventMetadataITCase.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for lookup changelog event metadata. */
+public class LookupChangelogEventMetadataITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testEventMetadataCanBeReadAsWriteTime() throws Exception {
+ sql(
+ "CREATE TABLE source_table ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "data INT, "
+ + "event_ts BIGINT, "
+ + "writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"
Review Comment:
What does this mean?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4039682675
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogEventMetadataITCase.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for lookup changelog event metadata. */
+public class LookupChangelogEventMetadataITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testEventMetadataCanBeReadAsWriteTime() throws Exception {
+ sql(
+ "CREATE TABLE source_table ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "data INT, "
+ + "event_ts BIGINT, "
+ + "writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"
+ + ") WITH ("
+ + "'bucket'='1', "
+ + "'changelog-producer'='lookup', "
+ + "'sequence.field'='event_ts', "
+ + "'changelog-producer.expose-field-as-metadata'='event_ts')");
+
+ // The metadata column models a Cassandra sink column populated from WRITETIME. The
+ // physical event_ts column remains available to normal Flink operators, while writetime
+ // carries the incoming event value even on an UPDATE_BEFORE retraction.
+ BlockingIterator<Row, Row> iterator =
+ streamSqlBlockIter("SELECT id, data, event_ts, writetime FROM source_table");
+
+ sql("INSERT INTO source_table VALUES (1, 10, 50)");
+ assertThat(iterator.collect(1)).containsExactly(Row.of(1, 10, 50L, 50L));
+
+ sql("INSERT INTO source_table VALUES (1, 20, 100)");
+ assertThat(iterator.collect(2))
+ .containsExactly(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, 10, 50L, 100L),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 20, 100L, 100L));
+
+ iterator.close();
+ }
+
+ @Test
+ public void testPhysicalFilterStillSeesBeforeImage() throws Exception {
Review Comment:
Add some test cases where `"writetime BIGINT METADATA FROM 'paimon.event.event_ts' VIRTUAL"` is not configured and what's seen on the table?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4039641638
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java:
##########
@@ -172,26 +176,53 @@ public RowLevelModificationScanContext applyRowLevelModificationScan(
rowLevelModificationSnapshotId);
}
+ public static final String EVENT_METADATA_PREFIX = "paimon.event.";
Review Comment:
There are 2 prefixes configured which can be quite confusing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4037346160
##########
docs/docs/primary-key-table/changelog-producer.md:
##########
@@ -109,6 +109,48 @@ checkpoint completion waits for compaction. Tune it with checkpoint duration and
`lookup` is incompatible with `full-compaction.delta-commits`. For periodic full compaction with
changelog generation, use `full-compaction` instead.
+By setting `'changelog-producer.expose-field-as-metadata'` to a comma-separated list of column names,
+the lookup changelog producer stores those columns' values from the incoming event as additional metadata
+columns (`__internal__<column>`) in all changelog records. For retraction records (`-U`, `-D`), the regular
+value columns retain the correct before-image so standard downstream operators (filters, aggregations)
+work correctly, while the metadata columns carry the event values. For forward records (`+I`, `+U`),
+the metadata columns mirror the regular values for schema consistency. Sinks that need the event
+timestamp — such as Cassandra using `WRITETIME` for conflict resolution — can read the metadata columns.
+
+**Note:** The event values come from the *merged result* of the merge function, which equals the raw
+incoming event for the `deduplicate` merge engine but may differ for aggregation merge engines.
+
+This option is only supported by the `lookup` changelog producer.
+
+The metadata column name is formed by concatenating the configured prefix and the preserved
+column name. The default prefix is `__internal__`, so preserving `event_ts` creates the
+`__internal__event_ts` metadata column. Set `'changelog-producer.metadata-field-prefix'` to a
+different prefix if the default conflicts with an existing column name.
+
+```sql
+-- Source table with event metadata preservation
+CREATE TABLE my_table (
+ id INT PRIMARY KEY NOT ENFORCED,
+ data STRING,
+ event_ts BIGINT
+) WITH (
+ 'changelog-producer' = 'lookup',
+ 'sequence.field' = 'event_ts',
+ 'changelog-producer.expose-field-as-metadata' = 'event_ts'
+);
+
+-- Sink table reading event metadata
+CREATE TABLE cassandra_sink (
Review Comment:
Instead of mentioning Cassandra, we can name it as external sink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5780164392
> 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](https://github.com/apache/paimon/pull/9691).
### 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 is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-6297340698
Thanks for the suggestions. I have tried addressing the feedback in 5a998464e5 (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](https://github.com/apache/paimon/pull/10702) 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4376128747
##########
paimon-core/src/main/java/org/apache/paimon/table/system/ChangelogEventMetadata.java:
##########
@@ -0,0 +1,461 @@
+/*
+ * 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.paimon.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.ChangelogProducer;
+import org.apache.paimon.data.variant.VariantMetadataUtils;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.NestedProjectedRow;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+/** Shared validation and row-type construction for changelog event metadata fields. */
+public final class ChangelogEventMetadata {
+
+ private ChangelogEventMetadata() {}
+
+ /**
+ * Validates the event metadata configuration against the physical value type.
+ *
+ * <p>The fields are appended to records produced by the lookup changelog wrapper. They must not
+ * be enabled for another changelog producer because those producers share the ordinary writer
+ * and do not emit the appended values.
+ */
+ public static void validate(RowType valueType, CoreOptions options) {
+ List<String> preserveColumns = options.changelogEventMetadataFields();
+ if (preserveColumns.isEmpty()) {
+ return;
+ }
+
+ if (options.changelogProducer() != ChangelogProducer.LOOKUP) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Option '%s' can only be used when '%s' is '%s', but it is '%s'.",
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key(),
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ ChangelogProducer.LOOKUP,
+ options.changelogProducer()));
+ }
+
+ Set<String> valueFieldNames = new HashSet<>(valueType.getFieldNames());
+ Set<String> preservedColumns = new HashSet<>();
+ Set<String> metadataFieldNames = new HashSet<>();
+ Map<String, String> metadataFieldSources = new HashMap<>();
+ Map<String, String> storageFieldSources = new HashMap<>();
+ for (String preserveColumn : preserveColumns) {
+ if (!preservedColumns.add(preserveColumn)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Column '%s' is specified more than once in '%s'.",
+ preserveColumn,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+
+ if (!valueFieldNames.contains(preserveColumn)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Column '%s' specified in '%s' not found in value type. Available columns: %s",
+ preserveColumn,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key(),
+ valueType.getFieldNames()));
+ }
+
+ DataField physicalField = valueType.getField(preserveColumn);
+ String metadataFieldName = metadataFieldName(preserveColumn, options);
+ if (valueFieldNames.contains(metadataFieldName)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Metadata field '%s' created by '%s' conflicts with an existing value column.",
+ metadataFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+ if (SpecialFields.isSystemField(metadataFieldName)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Metadata field '%s' created by '%s' conflicts with a system field.",
+ metadataFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+ if (!metadataFieldNames.add(metadataFieldName)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Metadata field '%s' is created more than once by '%s'.",
+ metadataFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+
+ String storageFieldName = storageMetadataFieldName(physicalField, options);
+ if (valueFieldNames.contains(storageFieldName)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Storage metadata field '%s' created by '%s' "
+ + "conflicts with an existing value column.",
+ storageFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+ if (SpecialFields.isSystemField(storageFieldName)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Storage metadata field '%s' created by '%s' "
+ + "conflicts with a system field.",
+ storageFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+ if (storageFieldSources.put(storageFieldName, preserveColumn) != null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Storage metadata field '%s' is created more than once by '%s'.",
+ storageFieldName,
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key()));
+ }
+ metadataFieldSources.put(metadataFieldName, preserveColumn);
+ }
+
+ for (Map.Entry<String, String> metadataField : metadataFieldSources.entrySet()) {
+ String storageSource = storageFieldSources.get(metadataField.getKey());
+ if (storageSource != null && !storageSource.equals(metadataField.getValue())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Metadata field '%s' created by '%s' conflicts with the storage "
+ + "metadata field for column '%s'.",
+ metadataField.getKey(),
+ CoreOptions.CHANGELOG_PRODUCER_EVENT_METADATA_FIELDS.key(),
+ storageSource));
+ }
+ }
+ }
+
+ /**
+ * Returns the nullable public metadata fields appended to a table row.
+ *
+ * <p>{@code highestFieldId} must be the table schema's highest field ID, which also covers
+ * dropped fields, so that metadata field IDs never alias a field of a historical schema.
+ */
+ public static List<DataField> extraValueFields(
+ RowType valueType, int highestFieldId, CoreOptions options) {
+ validate(valueType, options);
+ return metadataValueFields(
+ valueType,
+ highestFieldId,
+ options,
+ physicalField -> metadataFieldName(physicalField.name(), options));
+ }
+
+ /**
+ * Returns the nullable fields used to store event metadata in changelog files.
+ *
+ * <p>Storage names use the configured prefix and source field ID, so they remain stable when
+ * the source column is renamed.
+ */
+ public static List<DataField> storageValueFields(
+ RowType valueType, int highestFieldId, CoreOptions options) {
+ validate(valueType, options);
+ return metadataValueFields(
+ valueType,
+ highestFieldId,
+ options,
+ physicalField -> storageMetadataFieldName(physicalField, options));
+ }
+
+ /**
+ * Resolves storage metadata fields against the value fields of the schema a data file was
+ * written with.
+ *
+ * <p>The metadata values were written with the source field's type at that time, so each field
+ * takes the historical source field type. This lets schema evolution cast the values like the
+ * source field itself. Fields whose source field did not exist yet keep the current type.
+ */
+ public static List<DataField> storageValueFieldsForDataSchema(
+ List<DataField> storageFields,
+ RowType valueType,
+ List<DataField> dataValueFields,
+ List<String> preserveColumns) {
+ List<DataField> fields = new ArrayList<>(storageFields.size());
+ for (int i = 0; i < storageFields.size(); i++) {
+ DataField storageField = storageFields.get(i);
+ if (i >= preserveColumns.size() || !valueType.containsField(preserveColumns.get(i))) {
+ fields.add(storageField);
+ continue;
+ }
+ int sourceFieldId = valueType.getField(preserveColumns.get(i)).id();
+ DataField dataSourceField = null;
+ for (DataField dataField : dataValueFields) {
+ if (dataField.id() == sourceFieldId) {
+ dataSourceField = dataField;
+ break;
+ }
+ }
+ fields.add(
+ dataSourceField == null
+ ? storageField
+ : storageField.newType(dataSourceField.type().copy(true)));
+ }
+ return fields;
+ }
+
+ /** Returns the physical value-field positions copied into event metadata columns. */
+ @Nullable
+ public static int[] preserveFieldIndices(RowType valueType, CoreOptions options) {
+ List<String> preserveColumns = options.changelogEventMetadataFields();
+ if (preserveColumns.isEmpty()) {
+ return null;
+ }
+ validate(valueType, options);
+ int[] indices = new int[preserveColumns.size()];
+ for (int i = 0; i < preserveColumns.size(); i++) {
+ indices[i] = valueType.getFieldIndex(preserveColumns.get(i));
+ }
+ return indices;
+ }
+
+ /**
+ * Appends event metadata fields to a row type while resolving their source fields from the
+ * physical value type.
+ *
+ * <p>The two row types intentionally differ for system tables such as {@code audit_log}, whose
+ * base row prepends system fields to the physical value fields.
+ */
+ public static RowType appendMetadataFields(
+ RowType baseRowType, RowType valueType, int highestFieldId, CoreOptions options) {
+ return appendFields(baseRowType, extraValueFields(valueType, highestFieldId, options));
+ }
+
+ /** Appends the internal storage metadata fields to a row type. */
+ public static RowType appendStorageMetadataFields(
+ RowType baseRowType, RowType valueType, int highestFieldId, CoreOptions options) {
+ return appendFields(baseRowType, storageValueFields(valueType, highestFieldId, options));
+ }
+
+ private static RowType appendFields(RowType baseRowType, List<DataField> extraFields) {
+ if (extraFields.isEmpty()) {
+ return baseRowType;
+ }
+
+ boolean hasExistingMetadata = false;
+ for (DataField extraField : extraFields) {
+ if (baseRowType.containsField(extraField.name())) {
+ hasExistingMetadata = true;
+ }
+ }
+ if (hasExistingMetadata) {
+ boolean allExisting =
+ extraFields.stream().allMatch(field -> baseRowType.containsField(field.name()));
+ if (allExisting) {
+ return baseRowType;
+ }
+ throw new IllegalArgumentException(
+ "Changelog event metadata fields conflict with the requested row type.");
+ }
+
+ List<DataField> fields = new ArrayList<>(baseRowType.getFields());
+ fields.addAll(extraFields);
+ return new RowType(fields);
+ }
+
+ /**
+ * Adds the physical fields that back requested metadata fields when reading data files.
+ *
+ * <p>Data files do not store metadata fields, so readers copy each metadata value from its
+ * source field. A source field that is not requested is appended; one that is requested is
+ * widened when its nested children do not cover those of the metadata field. Both cases use the
+ * union of the requested nested children, so callers must project their output back to the
+ * requested read type. Returns {@code readType} itself when no change is needed.
+ */
+ public static RowType withMetadataDependencies(
+ RowType readType, RowType logicalRowType, CoreOptions options) {
+ List<String> preserveColumns = options.changelogEventMetadataFields();
+ if (preserveColumns.isEmpty()) {
+ return readType;
+ }
+
+ List<DataField> fields = new ArrayList<>(readType.getFields());
+ boolean changed = false;
+ for (String preserveColumn : preserveColumns) {
+ String metadataName = metadataFieldName(preserveColumn, options);
+ if (!readType.containsField(metadataName)
+ || !logicalRowType.containsField(preserveColumn)) {
+ continue;
+ }
+ DataField sourceField = logicalRowType.getField(preserveColumn);
+ DataType metadataType = readType.getField(metadataName).type();
+ int sourceIndex = readType.getFieldIndex(preserveColumn);
+ if (sourceIndex < 0) {
+ fields.add(sourceField.newType(metadataType.copy(sourceField.type().isNullable())));
+ changed = true;
+ continue;
+ }
+ DataType requestedType = fields.get(sourceIndex).type();
+ DataType dependencyType =
+ unionPrunedType(sourceField.type(), requestedType, metadataType);
+ if (!dependencyType.equals(requestedType)) {
+ fields.set(sourceIndex, fields.get(sourceIndex).newType(dependencyType));
+ changed = true;
+ }
+ }
+ return changed ? new RowType(readType.isNullable(), fields) : readType;
+ }
+
+ /**
+ * Creates a view of a row read with {@link #withMetadataDependencies} in which each source
+ * field mapped by {@code fallbackMapping} has the nested shape of its metadata field. Returns
+ * null if no nested projection is needed.
+ */
+ @Nullable
+ public static NestedProjectedRow createFallbackProjection(
+ RowType readType, int[] fallbackMapping) {
+ List<DataField> fields = new ArrayList<>(readType.getFields());
+ for (int i = 0; i < fallbackMapping.length; i++) {
+ int sourceIndex = fallbackMapping[i];
+ if (sourceIndex < 0) {
+ continue;
+ }
+ DataField sourceField = fields.get(sourceIndex);
+ DataType metadataType = readType.getFields().get(i).type();
+ if (sourceField.type().getTypeRoot() != metadataType.getTypeRoot()) {
+ continue;
+ }
+ fields.set(
+ sourceIndex,
+ sourceField.newType(metadataType.copy(sourceField.type().isNullable())));
+ }
+ return NestedProjectedRow.create(readType, new RowType(readType.isNullable(), fields));
+ }
+
+ /**
+ * Returns the type of {@code fullType} pruned to the nested children requested by either {@code
+ * left} or {@code right}, keeping the declared child order of {@code fullType}.
+ */
+ private static DataType unionPrunedType(DataType fullType, DataType left, DataType right) {
+ // An engine may read a column as a different kind of type, for example a selected-key MAP
+ // pushdown reads a ROW. Such a type cannot be merged with the metadata field's type.
+ if (left.getTypeRoot() != fullType.getTypeRoot()
+ || right.getTypeRoot() != fullType.getTypeRoot()) {
+ return left;
Review Comment:
[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.
##########
paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java:
##########
@@ -67,6 +82,15 @@ public FileRecordIterator<KeyValue> readBatch() throws IOException {
return null;
}
KeyValue kv = serializer.fromRow(internalRow).setLevel(level);
+ if (metadataFallbackRow != null) {
Review Comment:
[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.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala:
##########
@@ -48,6 +49,16 @@ abstract class PaimonSparkTableBase(val table: Table)
lazy val coreOptions = new CoreOptions(table.options())
+ override lazy val schema: org.apache.spark.sql.types.StructType = {
+ val baseRowType = table.rowType()
+ val extendedRowType = table match {
+ case fst: FileStoreTable if !coreOptions.changelogEventMetadataFields().isEmpty =>
+ ChangelogEventMetadataTable.computeExtendedRowType(fst, baseRowType)
Review Comment:
[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.
##########
paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java:
##########
@@ -76,6 +77,9 @@ public SplitRead<InternalRow> withIOManager(@Nullable IOManager ioManager) {
@Override
public SplitRead<InternalRow> withReadType(RowType readType) {
this.readType = readType;
+ // Let the merge read add its internal dependencies, such as sequence fields and the
+ // source fields of event metadata, so the diff is computed on a consistent value layout.
+ mergeRead.withReadType(readType);
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5860400336
@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 after the table has snapshots. 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on code in PR #9822:
URL: https://github.com/apache/paimon/pull/9822#discussion_r4037352492
##########
docs/docs/primary-key-table/changelog-producer.md:
##########
@@ -109,6 +109,48 @@ checkpoint completion waits for compaction. Tune it with checkpoint duration and
`lookup` is incompatible with `full-compaction.delta-commits`. For periodic full compaction with
changelog generation, use `full-compaction` instead.
+By setting `'changelog-producer.expose-field-as-metadata'` to a comma-separated list of column names,
Review Comment:
Can be made less verbose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "junmuz (via GitHub)" <gi...@apache.org>.
junmuz commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5813508895
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][flink][spark] Expose changelog event metadata as separate columns [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9822:
URL: https://github.com/apache/paimon/pull/9822#issuecomment-5954440402
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 sen
d 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 a
m 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org