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 4ef55adc67 [core] Reject merge-schema type widening for primary-key
and partition columns (#10077)
4ef55adc67 is described below
commit 4ef55adc67214cb01cab7530a7382ac8d7e7ccea
Author: Arvin <[email protected]>
AuthorDate: Tue Oct 13 13:45:07 2026 +0800
[core] Reject merge-schema type widening for primary-key and partition
columns (#10077)
---
.../apache/paimon/schema/SchemaMergingUtils.java | 50 +++++++++++++
.../paimon/schema/SchemaMergingUtilsTest.java | 76 +++++++++++++++++++
.../org/apache/paimon/spark/PaimonSinkTest.scala | 13 +++-
.../org/apache/paimon/spark/PaimonSinkTest.scala | 13 +++-
.../paimon/spark/sql/DataFrameWriteTestBase.scala | 87 ++++++++++++++--------
5 files changed, 198 insertions(+), 41 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java
index bb9f42fc79..88c5eb2f5c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java
@@ -68,6 +68,17 @@ public class SchemaMergingUtils {
return currentTableSchema;
}
+ // The automatic merge path must not relax the invariants enforced by
the explicit schema
+ // change path (see SchemaManagerUtils#assertNotUpdatingPartitionKeys /
+ // #assertNotUpdatingPrimaryKeys). Widening a key column changes its
value encoding, which
+ // silently rehashes it into another bucket, so the same logical key
would land in two
+ // different buckets and bypass deduplication. Reject such merges
instead of committing
+ // them.
+ assertNotUpdatingKeyColumn(
+ currentTableSchema, currentType, newRowType, "update",
"partition column");
+ assertNotUpdatingKeyColumn(
+ currentTableSchema, currentType, newRowType, "update",
"primary key");
+
return new TableSchema(
currentTableSchema.id() + 1,
newRowType.getFields(),
@@ -270,6 +281,45 @@ public class SchemaMergingUtils {
return DataTypeCasts.supportsCast(sourceType, targetType,
allowExplicitCast);
}
+ /**
+ * Rejects a merge that changes the type of a primary-key or partition
column, mirroring the
+ * explicit schema change path's guards. Only top-level columns are
considered, as key columns
+ * are required to be non-nested.
+ */
+ private static void assertNotUpdatingKeyColumn(
+ TableSchema currentTableSchema,
+ RowType currentType,
+ RowType mergedType,
+ String operation,
+ String keyDescription) {
+ List<String> keyColumns =
+ "primary key".equals(keyDescription)
+ ? currentTableSchema.primaryKeys()
+ : currentTableSchema.partitionKeys();
+ if (keyColumns.isEmpty()) {
+ return;
+ }
+
+ Map<String, DataField> mergedFieldMap =
+ buildFieldMap(mergedType.getFields(), /* caseSensitive */
true);
+ for (DataField currentField : currentType.getFields()) {
+ if (!keyColumns.contains(currentField.name())) {
+ continue;
+ }
+ DataField mergedField = mergedFieldMap.get(currentField.name());
+ if (mergedField != null &&
!mergedField.type().equals(currentField.type())) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Cannot %s %s type from %s to %s: [%s].",
+ operation,
+ keyDescription,
+ currentField.type(),
+ mergedField.type(),
+ currentField.name()));
+ }
+ }
+ }
+
private static DataField assignIdForNewField(DataField field,
AtomicInteger highestFieldId) {
DataType dataType = ReassignFieldId.reassign(field.type(),
highestFieldId);
return new DataField(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
index b5befc4158..8e3acb4047 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
@@ -144,6 +144,82 @@ public class SchemaMergingUtilsTest {
assertThat(fields.get(2).type()).isEqualTo(new IntType());
}
+ @Test
+ public void testRejectTypeWideningOnPrimaryKeyColumn() {
+ // An INT primary key must not be widened to BIGINT by the automatic
merge path: INT
+ // values hash into different buckets than their BIGINT counterparts,
so the same logical
+ // key would end up in two buckets and bypass intra-bucket
deduplication.
+ DataField a = new DataField(0, "a", new IntType());
+ DataField b = new DataField(1, "b", new
VarCharType(VarCharType.MAX_LENGTH));
+ TableSchema current =
+ new TableSchema(
+ 0,
+ Lists.newArrayList(a, b),
+ 1,
+ new ArrayList<>(),
+ Lists.newArrayList("a"),
+ new HashMap<>(),
+ "");
+
+ DataField aWidened = new DataField(-1, "a", new BigIntType());
+ RowType t = new RowType(Lists.newArrayList(aWidened, b));
+
+ assertThatThrownBy(() -> SchemaMergingUtils.mergeSchemas(current, t,
true, false, true))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("Cannot update primary key type");
+ }
+
+ @Test
+ public void testRejectTypeWideningOnPartitionColumn() {
+ // Partition columns are equally encoded into the partition path;
widening them would
+ // silently create a second partition layout for the same logical key.
This uses a plain
+ // INT -> BIGINT widening with allowExplicitCast = false, so the
rejection is exercised by
+ // the normal type-widening path and is independent of explicit-cast
behavior.
+ DataField a = new DataField(0, "a", new IntType());
+ DataField b = new DataField(1, "b", new IntType());
+ TableSchema current =
+ new TableSchema(
+ 0,
+ Lists.newArrayList(a, b),
+ 1,
+ Lists.newArrayList("b"),
+ new ArrayList<>(),
+ new HashMap<>(),
+ "");
+
+ DataField bWidened = new DataField(-1, "b", new BigIntType());
+ RowType t = new RowType(Lists.newArrayList(a, bWidened));
+
+ assertThatThrownBy(() -> SchemaMergingUtils.mergeSchemas(current, t,
true, false, true))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("Cannot update partition column type");
+ }
+
+ @Test
+ public void testAllowTypeWideningOnNonKeyColumn() {
+ // Widening a non-key column is unaffected by the key-column guard.
+ DataField a = new DataField(0, "a", new IntType());
+ DataField b = new DataField(1, "b", new IntType());
+ TableSchema current =
+ new TableSchema(
+ 0,
+ Lists.newArrayList(a, b),
+ 1,
+ new ArrayList<>(),
+ Lists.newArrayList("a"),
+ new HashMap<>(),
+ "");
+
+ DataField bWidened = new DataField(-1, "b", new BigIntType());
+ RowType t = new RowType(Lists.newArrayList(a, bWidened));
+
+ TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, true,
false, true);
+ assertThat(merged.id()).isEqualTo(1);
+ assertThat(merged.primaryKeys()).containsExactly("a");
+ assertThat(merged.fields().get(0).type()).isEqualTo(new IntType());
+ assertThat(merged.fields().get(1).type()).isEqualTo(new BigIntType());
+ }
+
@Test
public void testMergeSchemas() {
// This will test both `mergeSchemas` and `merge` methods.
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
index d213f257c7..045529d716 100644
---
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
+++
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
@@ -253,7 +253,12 @@ class PaimonSinkTest extends PaimonSparkTestBase with
StreamTest {
spark.sql("SELECT * FROM T ORDER BY a, b"),
Row(1, "2023-08-09") :: Row(2, "2023-08-09") :: Nil)
- val inputData = MemoryStream[(Long, Date, Int)]
+ // The primary key column `a` deliberately keeps its INT type:
widening a
+ // primary-key column changes its bucket hash and is rejected by the
automatic
+ // merge path (see
SchemaMergingUtilsTest#testRejectTypeWideningOnPrimaryKeyColumn).
+ // This test focuses on the schema evolution that is allowed —
widening a non-key
+ // column (`b`: STRING -> DATE) and adding a new column (`c`).
+ val inputData = MemoryStream[(Int, Date, Int)]
val stream = inputData
.toDS()
.toDF("a", "b", "c")
@@ -268,13 +273,13 @@ class PaimonSinkTest extends PaimonSparkTestBase with
StreamTest {
val query = () => spark.sql("SELECT * FROM T ORDER BY a")
try {
- inputData.addData((1L, date, 123), (3L, date, 456))
+ inputData.addData((1, date, 123), (3, date, 456))
stream.processAllAvailable()
checkAnswer(
query(),
- Row(1L, date, 123) :: Row(2L, Date.valueOf("2023-08-09"), null)
:: Row(
- 3L,
+ Row(1, date, 123) :: Row(2, Date.valueOf("2023-08-09"), null) ::
Row(
+ 3,
date,
456) :: Nil)
} finally {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
index f22dab4b9c..206c2c0bcf 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
@@ -256,7 +256,12 @@ class PaimonSinkTest extends PaimonSparkTestBase with
StreamTest {
spark.sql("SELECT * FROM T ORDER BY a, b"),
Row(1, "2023-08-09") :: Row(2, "2023-08-09") :: Nil)
- val inputData = MemoryStream[(Long, Date, Int)]
+ // The primary key column `a` deliberately keeps its INT type:
widening a
+ // primary-key column changes its bucket hash and is rejected by the
automatic
+ // merge path (see
SchemaMergingUtilsTest#testRejectTypeWideningOnPrimaryKeyColumn).
+ // This test focuses on the schema evolution that is allowed —
widening a non-key
+ // column (`b`: STRING -> DATE) and adding a new column (`c`).
+ val inputData = MemoryStream[(Int, Date, Int)]
val stream = inputData
.toDS()
.toDF("a", "b", "c")
@@ -271,13 +276,13 @@ class PaimonSinkTest extends PaimonSparkTestBase with
StreamTest {
val query = () => spark.sql("SELECT * FROM T ORDER BY a")
try {
- inputData.addData((1L, date, 123), (3L, date, 456))
+ inputData.addData((1, date, 123), (3, date, 456))
stream.processAllAvailable()
checkAnswer(
query(),
- Row(1L, date, 123) :: Row(2L, Date.valueOf("2023-08-09"), null)
:: Row(
- 3L,
+ Row(1, date, 123) :: Row(2, Date.valueOf("2023-08-09"), null) ::
Row(
+ 3,
date,
456) :: Nil)
} finally {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
index e4b91730ed..504ecfce7c 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
@@ -723,10 +723,14 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
}
checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"),
expected2)
- // Case 2: two fields with the evolved types: Int -> Long,
Long -> Decimal
+ // Case 2: a non-key column evolves its type (Long ->
Decimal). The primary key
+ // `a` deliberately keeps its INT type: widening a primary-key
or partition column
+ // changes its bucket hash and is rejected by the merge path
(see
+ //
SchemaMergingUtilsTest#testRejectTypeWideningOnPrimaryKeyColumn). Key widening
is
+ // exercised as a rejection case in its own test below.
val df3 = Seq(
- (2L, "b2", BigDecimal.decimal(234), Map("k" -> 22.2)),
- (4L, "d", BigDecimal.decimal(456), Map("k" ->
44.4))).toDF("a", "b", "c", "d")
+ (2, "b2", BigDecimal.decimal(234), Map("k" -> 22.2)),
+ (4, "d", BigDecimal.decimal(456), Map("k" ->
44.4))).toDF("a", "b", "c", "d")
df3.write
.format("paimon")
.mode("append")
@@ -734,42 +738,62 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
.option("write.merge-schema.type-widening", "true")
.save(location)
val expected3 = if (hasPk) {
- Row(1L, "a2", BigDecimal.decimal(123), Map("k" -> 11.1)) ::
Row(
- 2L,
+ Row(1, "a2", BigDecimal.decimal(123), Map("k" -> 11.1)) ::
Row(
+ 2,
"b2",
BigDecimal.decimal(234),
Map("k" -> 22.2)) :: Row(
- 3L,
+ 3,
"c",
BigDecimal.decimal(345),
Map("k" -> 33.3)) :: Row(
- 4L,
+ 4,
"d",
BigDecimal.decimal(456),
Map("k" -> 44.4)) :: Nil
} else {
- Row(1L, "a", null, null) :: Row(
- 1L,
+ Row(1, "a", null, null) :: Row(
+ 1,
"a2",
BigDecimal.decimal(123),
- Map("k" -> 11.1)) :: Row(2L, "b", null, null) :: Row(
- 2L,
+ Map("k" -> 11.1)) :: Row(2, "b", null, null) :: Row(
+ 2,
"b2",
BigDecimal.decimal(234),
Map("k" -> 22.2)) :: Row(
- 3L,
+ 3,
"c",
BigDecimal.decimal(345),
Map("k" -> 33.3)) :: Row(
- 4L,
+ 4,
"d",
BigDecimal.decimal(456),
Map("k" -> 44.4)) :: Nil
}
checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"),
expected3)
- // Case 3: insert Decimal(20,18) to Decimal(38,18)
- val df4 = Seq((99L, "df4", BigDecimal.decimal(4.0), Map("4" ->
4.1)))
+ // Case 2b: widening a primary-key column must be rejected and
leave the schema
+ // and the previously committed rows unchanged.
+ if (hasPk) {
+ val schemaIdBefore = loadTable("T").schema().id()
+ val dfKeyWiden = Seq((2L, "b2"), (4L, "d")).toDF("a", "b")
+ val widMsg = intercept[Exception] {
+ dfKeyWiden.write
+ .format("paimon")
+ .mode("append")
+ .option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
+ .save(location)
+ }.getMessage
+ assert(widMsg.contains("Cannot update primary key type"))
+ // The rejected merge must not have persisted a schema
change or new rows.
+ assert(loadTable("T").schema().id() == schemaIdBefore)
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"),
expected3)
+ }
+
+ // Case 3: insert Decimal(20,18) to Decimal(38,18). The
primary key `a` stays INT so
+ // the key-type guard does not fire; only the non-key column
`c` is widened.
+ val df4 = Seq((99, "df4", BigDecimal.decimal(4.0), Map("4" ->
4.1)))
.toDF("a", "b", "c", "d")
.selectExpr("a", "b", "cast(c as decimal(20,18)) as c", "d")
df4.write
@@ -779,7 +803,7 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
.option("write.merge-schema.type-widening", "true")
.save(location)
val expected4 =
- expected3 ++ Seq(Row(99L, "df4", BigDecimal.decimal(4.0),
Map("4" -> 4.1)))
+ expected3 ++ Seq(Row(99, "df4", BigDecimal.decimal(4.0),
Map("4" -> 4.1)))
checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"),
expected4)
val decimalType =
spark.table("T").schema.apply(2).dataType.asInstanceOf[DecimalType]
@@ -840,9 +864,10 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
}
checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"), expected2)
- // Case 2: a: Int -> Long, b: String -> Date, c: Long -> Int, d:
Map -> String
+ // Case 2: a is kept INT (widening a primary-key column is
rejected), b: String -> Date,
+ // c: Long -> Int, d: Map -> String are the evolved non-key
columns.
val date = java.sql.Date.valueOf("2023-07-31")
- val df3 = Seq((2L, date, 234, null), (4L, date, 456, "2023-08-01
11:00:00.0")).toDF(
+ val df3 = Seq((2, date, 234, null), (4, date, 456, "2023-08-01
11:00:00.0")).toDF(
"a",
"b",
"c",
@@ -866,26 +891,22 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
.option("write.merge-schema.explicit-cast", "true")
.save(location)
val expected3 = if (hasPk) {
- Row(1L, Date.valueOf("2023-08-01"), 12, ts.toString) :: Row(
- 2L,
- date,
- 234,
- null) :: Row(3L, Date.valueOf("2023-08-03"), 34, ts.toString)
:: Row(
- 4L,
- date,
- 456,
- "2023-08-01 11:00:00.0") :: Nil
+ Row(1, Date.valueOf("2023-08-01"), 12, ts.toString) :: Row(2,
date, 234, null) :: Row(
+ 3,
+ Date.valueOf("2023-08-03"),
+ 34,
+ ts.toString) :: Row(4, date, 456, "2023-08-01 11:00:00.0") ::
Nil
} else {
- Row(1L, Date.valueOf("2023-08-01"), null, null) :: Row(
- 1L,
+ Row(1, Date.valueOf("2023-08-01"), null, null) :: Row(
+ 1,
Date.valueOf("2023-08-01"),
12,
- ts.toString) :: Row(2L, date, 234, null) :: Row(
- 2L,
+ ts.toString) :: Row(2, date, 234, null) :: Row(
+ 2,
Date.valueOf("2023-08-02"),
null,
- null) :: Row(3L, Date.valueOf("2023-08-03"), 34, ts.toString)
:: Row(
- 4L,
+ null) :: Row(3, Date.valueOf("2023-08-03"), 34, ts.toString)
:: Row(
+ 4,
date,
456,
"2023-08-01 11:00:00.0") :: Nil
