[flink] changelog read support for pk table without pushdown optimizations #2347

MehulBatra · 2026-01-11T20:09:29Z

Purpose

Linked issue: close #2333

Brief change log

File: ChangelogFlinkTableSource.java
Purpose: Flink table source for $changelog virtual tables
────────────────────────────────────────
File: ChangelogDeserializationSchema.java
Purpose: Deserializes LogRecord → RowData with metadata
────────────────────────────────────────
File: ChangelogRowConverter.java
Purpose: Converts records, adds _change_type, _log_offset,
commit_timestamp
────────────────────────────────────────
File: RecordToFlinkRowConverter.java
Purpose: Interface for record converters
────────────────────────────────────────
File: FlinkTableFactory.java
Change: Routes to ChangelogFlinkTableSource for changelog
tables
────────────────────────────────────────
File: TableDescriptor.java
Change: Added CHANGELOG* column name constants
────────────────────────────────────────
File: TableDescriptorValidation.java
Change: Validates reserved column names
Test Coverage
Test: testChangelogVirtualTableSchema
Type: Schema
What it verifies: DESCRIBE shows correct columns
────────────────────────────────────────
File: FlinkCatalog.java
Change: Detects $changelog suffix, builds virtual table
schema

Access via SELECT * FROM table$changelog
Metadata columns: _change_type (VARCHAR), _log_offset (BIGINT), _commit_timestamp (TIMESTAMP)
All records output as INSERT (change type in column)
Supports PK tables with change types: +I, -U, +U, -D

Tests

File: ChangelogVirtualTableITCase.java
Purpose: Integration tests (7 tests)
────────────────────────────────────────
File: ChangelogRowConverterTest.java
Purpose: Unit tests for converter
────────────────────────────────────────

API and Format

Documentation

wuchong · 2026-01-12T02:03:48Z

Great! Thank you @MehulBatra , I will review it.

wuchong

Thanks @MehulBatra for the contribution. I left some comments.

wuchong · 2026-01-18T08:09:34Z

.gitignore

 website/versioned_sidebars
 website/versions.json
 website/pnpm-lock.yaml
+/.claude/


Should it be .claude/?

nit: And we can move it near the .vscode/

wuchong · 2026-01-18T08:29:05Z

...ss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java

+        assertThat(schemaRows.get(0)).contains("_change_type");
+        assertThat(schemaRows.get(0)).contains("STRING");
+        // Flink DESCRIBE shows nullability as 'false' for NOT NULL columns
+        assertThat(schemaRows.get(0)).contains("false");
+
+        assertThat(schemaRows.get(1)).contains("_log_offset");
+        assertThat(schemaRows.get(1)).contains("BIGINT");
+        assertThat(schemaRows.get(1)).contains("false");


The DESCRIBE result is deterministic, so we should assert the entire output string rather than using contains.

Using contains makes the test intent unclear to readers and risks missing regressions in other parts of the output that should also be validated.

wuchong · 2026-01-18T08:39:07Z

...ss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java

+        assertThat(updateResults.get(1)).contains("+U");
+        assertThat(updateResults.get(1)).contains("Item-1-Updated");
+        assertThat(updateResults.get(1)).contains("150");


Simplify to assertThat(updateResults.get(0)).contains("-U", "Item-1", "100");

Update others like this.

wuchong · 2026-01-18T08:40:56Z

...ss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java

+        // Collect delete record
+        List<String> deleteResult = collectRowsWithTimeout(rowIter, 1, true);
+        assertThat(deleteResult).hasSize(1);
+        // Verify the delete record contains the row data (the change type may be -D or -U)


If this is a delete, the change type must be -D, so we should assert the change type here.

wuchong · 2026-01-18T08:48:17Z

...ss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java

+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Integration test for $changelog virtual table functionality. */
+abstract class ChangelogVirtualTableITCase extends AbstractTestBase {


This is a thorough test. However, we are facing limited testing resource (the Flink test is reaching to the 1 hour limit). And IT cases are most time-consuming. I think many tests in this IT case are doing duplicated things and we can improve this.

testChangelogVirtualTableWithPrimaryKeyTable and testBasicChangelogScanWithMetadataValidation do the same thing, but testBasicChangelogScanWithMetadataValidation verified more things (metadata). So I think we can remove testChangelogVirtualTableWithPrimaryKeyTable or merge it into testBasicChangelogScanWithMetadataValidation.

testAllChangeTypes() and testBasicChangelogScanWithMetadataValidation both verified all change types. So I think we can change testAllChangeTypes to testProjectionOnChangelogTable to test projections. We can add one more column to the source table, and select _change_type and other 2 columns.

testChangelogVirtualTableConcurrentChanges can be removed as it doesn't test concurrent updates.

testChangelogVirtualTableWithComplexSchema can be merged into testChangelogVirtualTableSchemaIntrospection.

Besides, please add tests to verify $changelog also works when scan.startup.mode option is applied, and also works for partitioned table. Take org.apache.fluss.flink.source.FlinkTableSourceITCase#testReadLogTableWithDifferentScanStartupMode as an example.

wuchong · 2026-01-18T10:42:26Z

...fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/RecordToFlinkRowConverter.java

+import java.io.Serializable;
+
+/** Interface for converting Fluss {@link LogRecord} to Flink {@link RowData}. */
+public interface RecordToFlinkRowConverter extends Serializable {


Since we will remove PlainRowConverter, there is only one implementation of this interface. Then this interface abstraction looks useless. I think we can remove this interface class. (add Serializable for ChangelogRowConverter if we decide to remove RecordToFlinkRowConverter).

wuchong · 2026-01-18T10:44:09Z

fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java

+                                    BUCKET_COLUMN_NAME,
+                                    CHANGELOG_CHANGE_TYPE_COLUMN,
+                                    CHANGELOG_LOG_OFFSET_COLUMN,
+                                    CHANGELOG_COMMIT_TIMESTAMP_COLUMN)));


Add tests for the columns in org.apache.fluss.client.admin.FlussAdminITCase#testSystemsColumns

wuchong · 2026-01-18T11:34:57Z

...nk-1.20/src/test/java/org/apache/fluss/flink/source/Flink120ChangelogVirtualTableITCase.java

+package org.apache.fluss.flink.source;
+
+/** IT case for {@link ChangelogVirtualTableITCase} in Flink 1.20. */
+public class Flink120ChangelogVirtualTableITCase extends ChangelogVirtualTableITCase {}


Create ITCases for all Flink versions to test compatibilites?

wuchong · 2026-01-18T11:37:56Z

...ss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java

+            // Parse the row to validate structure
+            String[] parts = result.substring(3, result.length() - 1).split(", ", 6);
+
+            // Validate change type column
+            assertThat(parts[0]).isEqualTo("+I");


We can set ManualClock to the FlussClusterExtension to control the commit timestamp by setting the current time CLOCK.advanceTime(xxxxx, TimeUnit.MILLISECONDS), see the usage in FlinkTableSourceITCase.

In this way, we can assert the changelog result by comparing the whole row string. (set bucket to 1 to have consistent log_offset numbers)

wuchong · 2026-01-18T11:39:41Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

+    /**
+     * Creates a virtual $changelog table by modifying the base table's to include metadata columns.
+     */
+    private CatalogBaseTable getVirtualChangelogTable(ObjectPath objectPath)


Please add tests in FlinkCatalogITCase to assert the CatalogTable got from a $changelog table.

changelog support for pk table

86029a1

MehulBatra requested a review from wuchong January 11, 2026 20:13

wuchong reviewed Jan 18, 2026

View reviewed changes

[flink] changelog read support for pk table without pushdown optimizations #2347

Are you sure you want to change the base?

[flink] changelog read support for pk table without pushdown optimizations #2347

Uh oh!

Conversation

MehulBatra commented Jan 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

wuchong commented Jan 12, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

MehulBatra commented Jan 11, 2026 •

edited

Loading