From 976b6a45adc281a4043003e32878e65315a19225 Mon Sep 17 00:00:00 2001 From: luops Date: Wed, 22 Jul 2026 17:47:14 +0800 Subject: [PATCH] [core] Fix query results being scrambled after querying $audit_log base table --- .../paimon/table/system/AuditLogTable.java | 7 ++-- .../table/system/AuditLogTableTest.java | 36 +++++++++++++++++++ .../paimon/flink/BatchFileStoreITCase.java | 19 ++++++++++ 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java index 0725bc1db477..e13e32ce60e0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java @@ -97,7 +97,6 @@ public class AuditLogTable implements DataTable, ReadonlyTable { protected final List specialFields; public AuditLogTable(FileStoreTable wrapped) { - this.wrapped = wrapped; this.specialFields = new ArrayList<>(); specialFields.add(SpecialFields.ROW_KIND); @@ -105,9 +104,13 @@ public AuditLogTable(FileStoreTable wrapped) { CoreOptions.fromMap(wrapped.options()).tableReadSequenceNumberEnabled(); if (includeSequenceNumber) { - this.wrapped.options().put(CoreOptions.KEY_VALUE_SEQUENCE_NUMBER_ENABLED.key(), "true"); + wrapped = + wrapped.copyWithoutTimeTravel( + Collections.singletonMap( + CoreOptions.KEY_VALUE_SEQUENCE_NUMBER_ENABLED.key(), "true")); specialFields.add(SpecialFields.SEQUENCE_NUMBER); } + this.wrapped = wrapped; } /** Creates a PredicateReplaceVisitor that adjusts field indices by systemFieldCount. */ diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java index 93c7933b0b72..00b42c2841b1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java @@ -213,6 +213,42 @@ public void testReadSequenceNumberWithTableOption() throws Exception { assertThat(result).containsExactlyInAnyOrderElementsOf(expectRow); } + @Test + public void testReadBaseTableAfterAuditLogDoesNotIncludeSequenceNumber() throws Exception { + String tableName = "audit_table_share_check"; + Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse, database, tableName)); + FileIO fileIO = LocalFileIO.create(); + + TableSchema tableSchema = + SchemaUtils.forceCommit( + new SchemaManager(fileIO, tablePath), + Schema.newBuilder() + .column("pk", DataTypes.INT()) + .column("pt", DataTypes.INT()) + .column("col1", DataTypes.INT()) + .partitionKeys("pt") + .primaryKey("pk", "pt") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "input") + .option("bucket", "1") + .option( + CoreOptions.TABLE_READ_SEQUENCE_NUMBER_ENABLED.key(), + "true") + .build()); + FileStoreTable baseTable = + FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema); + writeTestData(baseTable); + + // Wrap in AuditLogTable, this should NOT mutate the shared underlying options map. + AuditLogTable auditLogTable = new AuditLogTable(baseTable); + assertThat(auditLogTable.rowType().getFieldNames()) + .containsExactly("rowkind", "_SEQUENCE_NUMBER", "pk", "pt", "col1"); + + // Base table row type must remain unchanged - no leaked `_SEQUENCE_NUMBER` field. + assertThat(baseTable.rowType().getFieldNames()).containsExactly("pk", "pt", "col1"); + assertThat(baseTable.options()) + .doesNotContainKey(CoreOptions.KEY_VALUE_SEQUENCE_NUMBER_ENABLED.key()); + } + @Test public void testReadSequenceNumberWithAlterTable() throws Exception { String tableName = "audit_table_alter_seq"; diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java index a3cbc264b96c..6ce5f53387f3 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java @@ -1309,6 +1309,25 @@ public void testAuditLogTableWithSequenceNumberEnabled() { .containsExactlyInAnyOrder(Row.of(0L, "+I"), Row.of(1L, "+I")); } + @Test + public void testReadBaseTableAfterAuditLogWithSequenceNumberEnabled() { + // Regression test: reading `t$audit_log` must not leak the internal + // KEY_VALUE_SEQUENCE_NUMBER_ENABLED option into subsequent reads of the base table. + sql( + "CREATE TABLE test_table_reuse (a int PRIMARY KEY NOT ENFORCED, b int, c AS a + b) " + + "WITH ('table-read.sequence-number.enabled'='true');"); + sql("INSERT INTO test_table_reuse VALUES (1, 2)"); + sql("INSERT INTO test_table_reuse VALUES (3, 4)"); + + // First read the audit log + assertThat(sql("SELECT * FROM `test_table_reuse$audit_log`")) + .containsExactlyInAnyOrder(Row.of("+I", 0L, 1, 2, 3), Row.of("+I", 1L, 3, 4, 7)); + + // Then read the base table - must not include _SEQUENCE_NUMBER column. + assertThat(sql("SELECT * FROM `test_table_reuse`")) + .containsExactlyInAnyOrder(Row.of(1, 2, 3), Row.of(3, 4, 7)); + } + @Test public void testAuditLogTableWithSequenceNumberAlterTable() { // Create primary key table without sequence-number option