This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d7f3d04c5a [core][spark] Fix RTAS reads after incompatible nested type
replace (#8469)
d7f3d04c5a is described below
commit d7f3d04c5a149d57de2224431f9cb80c653c0a29
Author: Zouxxyy <[email protected]>
AuthorDate: Tue Jul 7 15:25:50 2026 +0800
[core][spark] Fix RTAS reads after incompatible nested type replace (#8469)
Fix `CREATE OR REPLACE TABLE ... AS SELECT` after an incompatible nested
type replace.
The issue happens in the two-snapshot RTAS flow:
1. a truncate snapshot clears the old visible files
2. the following RTAS append snapshot writes the new result
Without extra handling, the RTAS append snapshot can still inherit stale
snapshot state from the truncate snapshot. This is especially
problematic for incompatible schema replacement, where latest-table
reads may observe old metadata/state that should not survive the
replace.
This change:
- adds a Spark SQL regression in `DDLTestBase` that reproduces the
incompatible nested-type RTAS read failure
- resets inherited manifest state for the RTAS append-after-truncate
path
- also clears inherited index manifest and statistics metadata in that
path
- adds a focused core unit test to verify the index/statistics reset
behavior
---
.../paimon/operation/FileStoreCommitImpl.java | 50 ++++++++++++-------
.../paimon/operation/FileStoreCommitTest.java | 45 +++++++++++++++++
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 57 ++++++++++++++++++++++
3 files changed, 135 insertions(+), 17 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index d5807e33d8..cb2357eabe 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -849,6 +849,16 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
null);
}
+ private boolean isRtasAfterTruncate(@Nullable Snapshot latestSnapshot,
CommitKind commitKind) {
+ if (latestSnapshot == null || operation == null) {
+ return false;
+ }
+
+ return latestSnapshot.operation() == Snapshot.Operation.TRUNCATE
+ && (operation == Snapshot.Operation.REPLACE_TABLE_AS_SELECT
+ || operation ==
Snapshot.Operation.CREATE_OR_REPLACE_TABLE_AS_SELECT);
+ }
+
@VisibleForTesting
CommitResult tryCommitOnce(
@Nullable RetryCommitResult retryResult,
@@ -1016,23 +1026,29 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
oldIndexManifest = latestSnapshot.indexManifest();
}
- // try to merge old manifest files to create base manifest list
- ManifestMergeReuse manifestMergeReuse =
- tryReuseManifestMergeResult(retryResult,
mergeBeforeManifests);
- skipManifestMergeOnRetry = manifestMergeReuse == null &&
retryResult != null;
- if (manifestMergeReuse != null) {
- mergeBeforeManifests = manifestMergeReuse.preservedManifests;
- mergeAfterManifests = manifestMergeReuse.mergeAfterManifests;
- } else if (skipManifestMergeOnRetry) {
- mergeAfterManifests = mergeBeforeManifests;
+ boolean resetSnapshotStateForRtas =
isRtasAfterTruncate(latestSnapshot, commitKind);
+ if (resetSnapshotStateForRtas) {
+ mergeBeforeManifests = emptyList();
+ mergeAfterManifests = emptyList();
+ oldIndexManifest = null;
} else {
- mergeAfterManifests =
- ManifestFileMerger.merge(
- mergeBeforeManifests,
- manifestFile,
- partitionType,
- options,
- ioManager);
+ ManifestMergeReuse manifestMergeReuse =
+ tryReuseManifestMergeResult(retryResult,
mergeBeforeManifests);
+ skipManifestMergeOnRetry = manifestMergeReuse == null &&
retryResult != null;
+ if (manifestMergeReuse != null) {
+ mergeBeforeManifests =
manifestMergeReuse.preservedManifests;
+ mergeAfterManifests =
manifestMergeReuse.mergeAfterManifests;
+ } else if (skipManifestMergeOnRetry) {
+ mergeAfterManifests = mergeBeforeManifests;
+ } else {
+ mergeAfterManifests =
+ ManifestFileMerger.merge(
+ mergeBeforeManifests,
+ manifestFile,
+ partitionType,
+ options,
+ ioManager);
+ }
}
baseManifestList = manifestList.write(mergeAfterManifests);
@@ -1081,7 +1097,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
String statsFileName = null;
if (newStatsFileName != null) {
statsFileName = newStatsFileName;
- } else if (latestSnapshot != null) {
+ } else if (latestSnapshot != null && !resetSnapshotStateForRtas) {
Optional<Statistics> previousStatistic =
statsFileHandler.readStats(latestSnapshot);
if (previousStatistic.isPresent()) {
if (previousStatistic.get().schemaId() != latestSchemaId) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 10bce28e94..e3903b5077 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1132,6 +1132,51 @@ public class FileStoreCommitTest {
.isEqualTo(0);
}
+ @Test
+ public void testRtasAppendAfterTruncateResetsInheritedIndexAndStats()
throws Exception {
+ TestFileStore store = createStore(false, 1,
CoreOptions.ChangelogProducer.NONE);
+ BinaryRow partition = gen.getPartition(gen.next());
+
+ store.commitData(generateDataList(1), s -> partition, kv -> 0);
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.commit(indexCommittable(partition, "stale-index", 0, 0),
false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ HashMap<String, ColStats<?>> fakeColStatsMap = new HashMap<>();
+ fakeColStatsMap.put("orderId", ColStats.newColStats(3, 1L, 1L, 1L, 0L,
8L, 8L));
+ Statistics fakeStats =
+ new Statistics(
+ latestSnapshot.id(), latestSnapshot.schemaId(), 1L,
100L, fakeColStatsMap);
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.commitStatistics(fakeStats, Long.MAX_VALUE);
+ }
+
+ try (FileStoreCommitImpl truncateCommit = store.newCommit()) {
+ truncateCommit.withOperation(Snapshot.Operation.TRUNCATE);
+ truncateCommit.truncateTable(1L);
+ }
+
+ List<KeyValue> replacement = generateDataList(1);
+ store.commitDataImpl(
+ replacement,
+ s -> partition,
+ kv -> 0,
+ false,
+ null,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> {
+
commit.withOperation(Snapshot.Operation.REPLACE_TABLE_AS_SELECT);
+ commit.commit(committable, false);
+ });
+
+ Snapshot rtasSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+
assertThat(rtasSnapshot.operation()).isEqualTo(Snapshot.Operation.REPLACE_TABLE_AS_SELECT);
+ assertThat(rtasSnapshot.indexManifest()).isNull();
+ assertThat(rtasSnapshot.statistics()).isNull();
+ }
+
@Test
public void testDropStatsForOverwrite() throws Exception {
TestFileStore store = createStore(false);
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
index 2c2405c366..9a63cab33d 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
@@ -406,6 +406,63 @@ abstract class DDLTestBase extends PaimonSparkTestBase {
}
}
+ test(
+ "Paimon DDL: CREATE OR REPLACE TABLE AS SELECT reads latest rows after
incompatible nested type replace") {
+ assume(gteqSpark3_4)
+ withTable("src", "t") {
+ sql("""
+ |CREATE TABLE src (
+ | id BIGINT,
+ | payload DOUBLE,
+ | name_a STRING,
+ | name_b STRING
+ |)
+ |USING paimon
+ |TBLPROPERTIES ('bucket' = '-1')
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO src VALUES
+ | (1, 1.1D, 'a', 'x'),
+ | (2, 2.2D, 'b', 'y')
+ |""".stripMargin)
+
+ sql("""
+ |CREATE TABLE t
+ |USING paimon
+ |TBLPROPERTIES ('bucket' = '-1')
+ |AS SELECT * FROM src
+ |""".stripMargin)
+
+ sql("""
+ |CREATE OR REPLACE TABLE t
+ |USING paimon
+ |TBLPROPERTIES ('bucket' = '-1')
+ |AS
+ |SELECT
+ | id,
+ | named_struct(
+ | 'items_before', array(name_a),
+ | 'items_after', array(name_b)
+ | ) AS payload,
+ | name_a,
+ | name_b
+ |FROM src
+ |""".stripMargin)
+
+ Assertions.assertEquals("struct",
spark.table("t").schema("payload").dataType.typeName)
+
+ checkAnswer(
+ sql("""
+ |SELECT id, payload.items_before, payload.items_after, name_a,
name_b
+ |FROM t
+ |WHERE name_a = 'a'
+ |LIMIT 1
+ |""".stripMargin),
+ Row(1L, Seq("a"), Seq("x"), "a", "x") :: Nil
+ )
+ }
+ }
+
test("Paimon DDL: REPLACE TABLE supports incompatible schema and preserves
old snapshots") {
assume(gteqSpark3_4)
withTable("t") {
