[core] Fix query results being scrambled after querying $audit_log ba…

paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java

@@ -97,17 +97,20 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
     protected final List<DataField> specialFields;
 
     public AuditLogTable(FileStoreTable wrapped) {
-        this.wrapped = wrapped;
         this.specialFields = new ArrayList<>();
         specialFields.add(SpecialFields.ROW_KIND);
 
         boolean includeSequenceNumber =
                 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. */

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

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