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 dc02d6ff90 [core] Rebuild the default value row when the write type
changes (#10208)
dc02d6ff90 is described below
commit dc02d6ff90633aa09aa8415f359fa8813ab94c19
Author: YangJie <[email protected]>
AuthorDate: Mon Oct 5 10:11:49 2026 -0400
[core] Rebuild the default value row when the write type changes (#10208)
---
.../org/apache/paimon/casting/DefaultValueRow.java | 33 ++++++++++++++-
.../apache/paimon/data/DefaultValueRowTest.java | 23 +++++++++++
.../apache/paimon/table/sink/TableWriteImpl.java | 5 ++-
.../NestedSubfieldDataEvolutionTableTest.java | 48 ++++++++++++++++++++++
.../NestedSubfieldMergeIntoActionITCase.java | 22 ++++++++++
.../apache/paimon/spark/sql/DefaultValueTest.scala | 12 ++++++
6 files changed, 140 insertions(+), 3 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/casting/DefaultValueRow.java
b/paimon-common/src/main/java/org/apache/paimon/casting/DefaultValueRow.java
index 555f065c70..0e73c35074 100644
--- a/paimon-common/src/main/java/org/apache/paimon/casting/DefaultValueRow.java
+++ b/paimon-common/src/main/java/org/apache/paimon/casting/DefaultValueRow.java
@@ -27,10 +27,13 @@ import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalVector;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.variant.Variant;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.NestedProjectedRow;
import javax.annotation.Nullable;
@@ -223,6 +226,16 @@ public class DefaultValueRow implements InternalRow {
@Nullable
public static DefaultValueRow create(RowType rowType) {
+ return create(rowType, rowType);
+ }
+
+ /**
+ * Creates the default value row of {@code rowType}, which may keep only
some sub-fields of a
+ * ROW field of {@code fullRowType}. The default of such a field is
converted with the full ROW
+ * type and then projected to the kept sub-fields.
+ */
+ @Nullable
+ public static DefaultValueRow create(RowType rowType, RowType fullRowType)
{
List<DataField> fields = rowType.getFields();
GenericRow row = new GenericRow(fields.size());
boolean containsDefaultValue = false;
@@ -234,8 +247,7 @@ public class DefaultValueRow implements InternalRow {
}
containsDefaultValue = true;
- Object defaultValue = convertDefaultValue(dataField.type(),
defaultValueStr);
- row.setField(i, defaultValue);
+ row.setField(i, defaultValueOf(dataField, fullRowType,
defaultValueStr));
}
if (!containsDefaultValue) {
@@ -244,4 +256,21 @@ public class DefaultValueRow implements InternalRow {
return DefaultValueRow.from(row);
}
+
+ private static Object defaultValueOf(
+ DataField field, RowType fullRowType, String defaultValueStr) {
+ DataType type = field.type();
+ if (type instanceof RowType && fullRowType.containsField(field.id())) {
+ DataType fullType = fullRowType.getField(field.id()).type();
+ if (fullType instanceof RowType && !fullType.equals(type)) {
+ InternalRow fullDefault =
+ (InternalRow) convertDefaultValue(fullType,
defaultValueStr);
+ NestedProjectedRow projected =
+ NestedProjectedRow.create((RowType) fullType,
(RowType) type);
+ return new InternalRowSerializer((RowType) type)
+ .copy(projected.replaceRow(fullDefault));
+ }
+ }
+ return convertDefaultValue(type, defaultValueStr);
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/DefaultValueRowTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/DefaultValueRowTest.java
index 43fccf10d1..4531078493 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/DefaultValueRowTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/DefaultValueRowTest.java
@@ -25,6 +25,8 @@ import org.apache.paimon.types.RowType;
import org.junit.Test;
+import java.util.Collections;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -123,6 +125,27 @@ public class DefaultValueRowTest {
assertThat(propertiesValue).isNull();
}
+ @Test
+ public void testDefaultValueRowProjectsRowDefaultToWriteType() {
+ RowType fullType =
+ RowType.of(
+ new DataField(
+ 0,
+ "nest",
+ RowType.of(
+ new DataField(1, "a", DataTypes.INT()),
+ new DataField(2, "b",
DataTypes.STRING())),
+ null,
+ "{42, z}"));
+ RowType writeType =
fullType.projectByPaths(Collections.singletonList("nest.b"));
+
+ DefaultValueRow defaultValueRow = DefaultValueRow.create(writeType,
fullType);
+
+ InternalRow nest = defaultValueRow.replaceRow(GenericRow.of((Object)
null)).getRow(0, 1);
+ assertThat(nest.getFieldCount()).isEqualTo(1);
+ assertThat(nest.getString(0).toString()).isEqualTo("z");
+ }
+
@Test
public void testDefaultValueRowWithRowType() {
// Test with Row default value
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
index 3d728aeed7..7f9add7879 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
@@ -63,11 +63,12 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
@Nullable private final RowKindFilter rowKindFilter;
private boolean batchCommitted = false;
+ private final RowType rowType;
private RowType writeType;
private int[] notNullFieldIndex;
private int[] deleteNotNullFieldIndex;
- private final @Nullable DefaultValueRow defaultValueRow;
+ private @Nullable DefaultValueRow defaultValueRow;
private final @Nullable Set<String> deleteNotNullFieldNames;
public TableWriteImpl(
@@ -95,6 +96,7 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
@Nullable RowKindGenerator rowKindGenerator,
@Nullable RowKindFilter rowKindFilter,
@Nullable List<String> deleteNotNullFieldNames) {
+ this.rowType = rowType;
this.writeType = rowType;
this.write = write;
this.keyAndBucketExtractor = keyAndBucketExtractor;
@@ -134,6 +136,7 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
write.withWriteType(writeType);
this.writeType = writeType;
updateNotNullFieldIndexes();
+ this.defaultValueRow = DefaultValueRow.create(writeType, rowType);
return this;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/NestedSubfieldDataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/NestedSubfieldDataEvolutionTableTest.java
index 3adea522b0..0ff1edbf04 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/NestedSubfieldDataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/NestedSubfieldDataEvolutionTableTest.java
@@ -150,6 +150,54 @@ public class NestedSubfieldDataEvolutionTableTest extends
DataEvolutionTestBase
}
}
+ @Test
+ public void testUpdateSubFieldOfRowWithDefault() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "nest",
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "a",
DataTypes.INT()),
+ DataTypes.FIELD(1, "b",
DataTypes.STRING())),
+ null,
+ "{42, z}")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+
.option(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true")
+ .build();
+ catalog.createTable(identifier(), schema, false);
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite()) {
+ write.write(GenericRow.of(1, GenericRow.of(10,
BinaryString.fromString("x"))));
+ write.write(GenericRow.of(2, GenericRow.of(20,
BinaryString.fromString("y"))));
+ commit(builder, write.prepareCommit());
+ }
+
+ RowType partialType =
table.rowType().projectByPaths(Collections.singletonList("nest.a"));
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(partialType)) {
+ write.write(GenericRow.of(GenericRow.of(100)));
+ // a null nest takes the default {42, z} projected to nest.a
+ write.write(GenericRow.of((Object) null));
+ List<CommitMessage> messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ commit(builder, messages);
+ }
+
+ List<InternalRow> rows = read(table);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(1);
+ InternalRow nest = rows.get(0).getRow(1, 2);
+ assertThat(nest.getInt(0)).isEqualTo(100);
+ assertThat(nest.getString(1).toString()).isEqualTo("x");
+ assertThat(rows.get(1).getInt(0)).isEqualTo(2);
+ nest = rows.get(1).getRow(1, 2);
+ assertThat(nest.isNullAt(0)).isFalse();
+ assertThat(nest.getInt(0)).isEqualTo(42);
+ assertThat(nest.getString(1).toString()).isEqualTo("y");
+ }
+
@Test
public void testDisabledOptionAllowsTopLevelColumnContainingDot() throws
Exception {
Schema disabledSchema =
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/NestedSubfieldMergeIntoActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/NestedSubfieldMergeIntoActionITCase.java
index a5722ff6ce..a2a83592f4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/NestedSubfieldMergeIntoActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/NestedSubfieldMergeIntoActionITCase.java
@@ -119,6 +119,28 @@ public class NestedSubfieldMergeIntoActionITCase extends
ActionITCaseBase {
assertThat(deltaWriteCols("T")).contains(Collections.singletonList("nest.a"));
}
+ @Test
+ public void testUpdateSubFieldOfRowWithDefault() throws Exception {
+ prepareNestedTarget(true);
+ sEnv.executeSql("CALL sys.alter_column_default_value('default.T',
'nest', '{42, z}')")
+ .await();
+
assertThat(getFileStoreTable("T").rowType().getField("nest").defaultValue())
+ .isEqualTo("{42, z}");
+ prepareSubFieldSource();
+
+ builder(warehouse, database, "T")
+ .withMergeCondition("T.id=S.id")
+ .withMatchedUpdateSet("T.nest.a=S.newa")
+ .withSourceTable("S")
+ .withSinkParallelism(2)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, nest.a, nest.b FROM T",
+ Arrays.asList(changelogRow("+I", 1, 100, "x"),
changelogRow("+I", 2, 20, "y")));
+ }
+
@Test
public void testUpdateSubFieldDisabledThrows() throws Exception {
// data-evolution.nested-field.enabled left at its default (false)
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DefaultValueTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DefaultValueTest.scala
index 22edb8f61a..493b1e5f2c 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DefaultValueTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DefaultValueTest.scala
@@ -92,6 +92,18 @@ class DefaultValueTest extends PaimonSparkTestBase {
}
}
+ test("Default Value: update a row-tracking table") {
+ withTable("t") {
+ sql("""CREATE TABLE t (id INT, data INT, c INT DEFAULT 5)
+ |TBLPROPERTIES ('row-tracking.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 1, 10), (2, 2, 20)")
+
+ sql("UPDATE t SET data = 22 WHERE id = 2")
+ checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(1, 1, 10),
Row(2, 22, 20)))
+ }
+ }
+
test("Default Value: unsupported default value") {
withTimeZone("Asia/Shanghai") {
withTable("t") {
