fileFieldIdsForEntry(ManifestEntry entry) {
return fileFieldIdsCache.computeIfAbsent(
Pair.of(entry.file().schemaId(), entry.file().writeCols()),
- pair -> fileFieldIds(this::scanTableSchema, entry.file()));
+ pair -> {
+ TableSchema fileSchema = scanTableSchema(entry.file().schemaId());
+ boolean nestedFieldEnabled =
+ new CoreOptions(fileSchema.options()).dataEvolutionNestedFieldEnabled();
+ return fileFieldIds(fileSchema.fields(), entry.file(), nestedFieldEnabled);
+ });
}
@VisibleForTesting
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionReadPlanner.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionReadPlanner.java
index 70e50fc38169..50746b831577 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionReadPlanner.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionReadPlanner.java
@@ -33,33 +33,86 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
- * Pure (no-IO) planner for sub-field-level data evolution reads. Given the requested read row type
- * and, for each column-group file ("bunch"), the row type it physically provides (its written
- * columns, already wrapped with row-tracking fields), it decides for every read field whether it is
- * taken whole from a single file or composed sub-field by sub-field across several files (latest
- * file wins per leaf), and produces the offset maps and the per-field {@link
- * DataEvolutionRow.NestedField} assembly plans.
+ * Pure (no-IO) planner for data evolution reads. Given the requested read row type and, for each
+ * column-group file ("bunch"), the row type it physically provides (its written columns, already
+ * wrapped with row-tracking fields), it produces the source offsets and per-bunch physical read
+ * fields.
*
- * Separating this from {@link DataEvolutionSplitRead} keeps the reader-building (IO) thin and
- * lets the layout logic be unit-tested directly. Only one level of nested composition is supported;
- * deeper or cross-file splits of a sub-struct throw {@link UnsupportedOperationException}.
+ *
With nested-field evolution disabled, fields are matched only by top-level id. With it
+ * enabled, the planner selects the latest provider per leaf and composes nested fields split across
+ * bunches. Only one level of nested composition is supported; deeper or cross-file splits of a
+ * sub-struct throw {@link UnsupportedOperationException}.
+ *
+ *
Separating this from {@link DataEvolutionSplitRead} keeps schema resolution and reader
+ * creation out of the layout logic and lets both planning modes be unit-tested directly.
*/
class DataEvolutionReadPlanner {
private final RowType readRowType;
// for each bunch, the (row-tracked) row type it physically provides
private final List bunchAvailTypes;
+ private final boolean nestedFieldEnabled;
- DataEvolutionReadPlanner(RowType readRowType, List bunchAvailTypes) {
+ DataEvolutionReadPlanner(
+ RowType readRowType, List bunchAvailTypes, boolean nestedFieldEnabled) {
this.readRowType = readRowType;
this.bunchAvailTypes = bunchAvailTypes;
+ this.nestedFieldEnabled = nestedFieldEnabled;
}
DataEvolutionReadPlan plan() {
+ DataEvolutionReadPlan plan = nestedFieldEnabled ? planNested() : planTopLevel();
+ List readFields = readRowType.getFields();
+ for (int i = 0; i < readFields.size(); i++) {
+ if (plan.rowOffsets[i] == -1 && plan.nested[i] == null) {
+ checkArgument(
+ readFields.get(i).type().isNullable(),
+ "Field %s is not null but can't find any file contains it.",
+ readFields.get(i));
+ }
+ }
+ return plan;
+ }
+
+ private DataEvolutionReadPlan planTopLevel() {
+ List allReadFields = readRowType.getFields();
+ int numFields = allReadFields.size();
+ int[] readFieldIds = allReadFields.stream().mapToInt(DataField::id).toArray();
+ int[] rowOffsets = new int[numFields];
+ int[] fieldOffsets = new int[numFields];
+ Arrays.fill(rowOffsets, -1);
+ Arrays.fill(fieldOffsets, -1);
+
+ List> bunchReadFields = new ArrayList<>();
+ for (int i = 0; i < bunchAvailTypes.size(); i++) {
+ Set availableFieldIds =
+ bunchAvailTypes.get(i).getFields().stream()
+ .map(DataField::id)
+ .collect(Collectors.toSet());
+ List readFields = new ArrayList<>();
+ for (int j = 0; j < readFieldIds.length; j++) {
+ if (rowOffsets[j] == -1 && availableFieldIds.contains(readFieldIds[j])) {
+ rowOffsets[j] = i;
+ fieldOffsets[j] = readFields.size();
+ readFields.add(allReadFields.get(j));
+ }
+ }
+ bunchReadFields.add(readFields);
+ }
+
+ return new DataEvolutionReadPlan(
+ rowOffsets,
+ fieldOffsets,
+ new DataEvolutionRow.NestedField[numFields],
+ bunchReadFields);
+ }
+
+ private DataEvolutionReadPlan planNested() {
List allReadFields = readRowType.getFields();
int numFields = allReadFields.size();
int numBunches = bunchAvailTypes.size();
@@ -89,12 +142,16 @@ DataEvolutionReadPlan plan() {
boolean[] composite = new boolean[numFields];
int[] wholeBunch = new int[numFields];
Arrays.fill(wholeBunch, -1);
+ List> nullnessAnchors = new ArrayList<>();
+ for (int i = 0; i < numFields; i++) {
+ nullnessAnchors.add(new LinkedHashSet<>());
+ }
for (int j = 0; j < numFields; j++) {
DataField rf = allReadFields.get(j);
List leaves = leafIdsOf(rf);
Map leafProvider = new HashMap<>();
- Set providers = new HashSet<>();
+ Set providers = new LinkedHashSet<>();
for (int leaf : leaves) {
int p = providerOf(leaf, bunchLeaves);
if (p >= 0) {
@@ -102,33 +159,52 @@ DataEvolutionReadPlan plan() {
providers.add(p);
}
}
- if (providers.isEmpty()) {
- // no file provides this field; it stays null (nullability checked below)
+
+ if (!(rf.type() instanceof RowType)) {
+ if (!providers.isEmpty()) {
+ int b = providers.iterator().next();
+ bunchSelection.get(b).put(rf.id(), null);
+ wholeBunch[j] = b;
+ }
+ continue;
+ }
+
+ // A ROW's nullness is determined by all of its latest sibling providers, including
+ // siblings omitted by the projection. Otherwise projecting only nest.a could return a
+ // null nest from a later nest.a file even though nest.b in another winning file keeps
+ // the merged nest non-null.
+ Set parentProviders =
+ topFieldProvidersOf(rf.id(), bunchAvailTypes, bunchLeaves);
+ if (parentProviders.isEmpty()) {
+ // The whole top-level ROW is absent. Leave it unplanned so the caller can either
+ // null-fill a nullable field or reject a missing non-null field.
continue;
}
- // Only read a field whole from a single file when that file covers ALL of its leaves.
- // If a single file provides only some leaves of a struct, go through the composite
- // plan so the selection is made per direct sub-field: the sub-fields it does provide
- // are read from it, and the ones absent everywhere stay null.
boolean allLeavesCovered = leafProvider.size() == leaves.size();
- if (providers.size() == 1 && allLeavesCovered) {
+ if (providers.size() == 1 && allLeavesCovered && parentProviders.equals(providers)) {
int b = providers.iterator().next();
bunchSelection.get(b).put(rf.id(), null);
wholeBunch[j] = b;
} else {
- checkArgument(
- rf.type() instanceof RowType,
- "Field %s is split across files but is not a struct.",
- rf.name());
composite[j] = true;
+ nullnessAnchors.get(j).addAll(parentProviders);
for (DataField sub : ((RowType) rf.type()).getFields()) {
- Set subProviders = new HashSet<>();
+ Set subProviders = new LinkedHashSet<>();
for (int leaf : leafIdsOf(sub)) {
int p = leafProvider.getOrDefault(leaf, -1);
if (p >= 0) {
subProviders.add(p);
}
}
+ if (subProviders.isEmpty()) {
+ // Every requested leaf may have been added after the files were written.
+ // Find the latest provider of older siblings under this direct sub-field;
+ // reading that sub-field preserves its ROW nullness while schema evolution
+ // null-fills the requested leaves.
+ subProviders =
+ subFieldProvidersOf(
+ rf.id(), sub.id(), bunchAvailTypes, bunchLeaves);
+ }
if (subProviders.size() > 1) {
throw new UnsupportedOperationException(
"Sub-field-level data evolution does not yet support splitting a "
@@ -152,6 +228,14 @@ DataEvolutionReadPlan plan() {
}
// else: sub-field absent everywhere -> stays null
}
+ for (int parentProvider : parentProviders) {
+ Map> selection = bunchSelection.get(parentProvider);
+ if (!selection.containsKey(rf.id())) {
+ // This provider only contributes an unprojected sibling. Read the projected
+ // shape as a hidden anchor so it still participates in parent nullness.
+ selection.put(rf.id(), null);
+ }
+ }
}
}
@@ -204,6 +288,16 @@ DataEvolutionReadPlan plan() {
Arrays.fill(subFieldOffsets, -1);
Map bunchToPartial = new LinkedHashMap<>();
List partials = new ArrayList<>();
+ for (int b : nullnessAnchors.get(j)) {
+ Map subOffsets = bunchSubOffset.get(b).get(rf.id());
+ bunchToPartial.put(b, partials.size());
+ partials.add(
+ new int[] {
+ b,
+ bunchTopOffset.get(b).get(rf.id()),
+ subOffsets == null ? subFields.size() : subOffsets.size()
+ });
+ }
for (int s = 0; s < subCount; s++) {
int subId = subFields.get(s).id();
int b = findSubProvider(rf.id(), subId, bunchSubOffset);
@@ -275,6 +369,53 @@ private static int providerOf(int leafId, List> bunchLeaves) {
return -1;
}
+ private static Set topFieldProvidersOf(
+ int fieldId, List bunchTypes, List> bunchLeaves) {
+ Set siblingLeaves = new LinkedHashSet<>();
+ for (RowType bunchType : bunchTypes) {
+ if (bunchType.containsField(fieldId)) {
+ collectLeafIds(
+ Collections.singletonList(bunchType.getField(fieldId)), siblingLeaves);
+ }
+ }
+ Set providers = new LinkedHashSet<>();
+ for (int leaf : siblingLeaves) {
+ int provider = providerOf(leaf, bunchLeaves);
+ if (provider >= 0) {
+ providers.add(provider);
+ }
+ }
+ return providers;
+ }
+
+ private static Set subFieldProvidersOf(
+ int topFieldId,
+ int subFieldId,
+ List bunchTypes,
+ List> bunchLeaves) {
+ Set siblingLeaves = new LinkedHashSet<>();
+ for (RowType bunchType : bunchTypes) {
+ if (!bunchType.containsField(topFieldId)) {
+ continue;
+ }
+ DataField topField = bunchType.getField(topFieldId);
+ if (topField.type() instanceof RowType
+ && ((RowType) topField.type()).containsField(subFieldId)) {
+ collectLeafIds(
+ Collections.singletonList(((RowType) topField.type()).getField(subFieldId)),
+ siblingLeaves);
+ }
+ }
+ Set providers = new LinkedHashSet<>();
+ for (int leaf : siblingLeaves) {
+ int provider = providerOf(leaf, bunchLeaves);
+ if (provider >= 0) {
+ providers.add(provider);
+ }
+ }
+ return providers;
+ }
+
private static int findSubProvider(
int topId, int subId, List>> bunchSubOffset) {
for (int b = 0; b < bunchSubOffset.size(); b++) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
index 4a0a2b5b7d40..56ab8d18854f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
@@ -82,7 +82,6 @@
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
-import static java.lang.String.format;
import static java.util.Collections.reverseOrder;
import static java.util.Comparator.comparingLong;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
@@ -226,9 +225,6 @@ private RecordReader createReader(
pathFactory.createDataFilePathFactory(partition, dataSplit.bucket());
List> suppliers = new ArrayList<>();
- // the merge path builds its readers with filter push down disabled, so the shared builder
- // carries no filters, the single file path creates its own with per file filters
- Builder formatBuilder = formatBuilder(readRowType, null);
// the suppliers below run lazily, so take the filters now, the same way the read type is
// already taken by the caller
List filters = readTypeFilters(this.filters, readRowType);
@@ -263,7 +259,6 @@ private RecordReader createReader(
needMergeFiles,
partition,
dataFilePathFactory,
- formatBuilder,
rowRanges,
readRowType,
deletionVector);
@@ -287,7 +282,6 @@ private DataEvolutionFileReader createUnionReader(
List needMergeFiles,
BinaryRow partition,
DataFilePathFactory dataFilePathFactory,
- Builder formatBuilder,
List rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector)
@@ -318,10 +312,11 @@ private DataEvolutionFileReader createUnionReader(
}
}
- // Init all we need to create a compound reader: resolve each bunch's physically-provided
- // (row-tracked) row type, then delegate the no-IO layout planning (leaf-level matching and
- // nested sub-field assembly) to DataEvolutionReadPlanner; this class only builds readers.
- List allReadFields = readRowType.getFields();
+ boolean nestedFieldEnabled = nestedFieldEnabledFor(needMergeFiles);
+ Builder formatBuilder = formatBuilder(readRowType, null, nestedFieldEnabled);
+ // Resolve each bunch's physically-provided (row-tracked) row type, then delegate all no-IO
+ // layout planning to DataEvolutionReadPlanner; this class only resolves schemas and builds
+ // readers.
int numBunches = fieldsFiles.size();
RecordReader[] fileRecordReaders = new RecordReader[numBunches];
@@ -333,7 +328,8 @@ private DataEvolutionFileReader createUnionReader(
bunchAvailTypes.add(rowTypeWithRowTracking(bunchDataSchemas[i].logicalRowType()));
}
DataEvolutionReadPlanner.DataEvolutionReadPlan plan =
- new DataEvolutionReadPlanner(readRowType, bunchAvailTypes).plan();
+ new DataEvolutionReadPlanner(readRowType, bunchAvailTypes, nestedFieldEnabled)
+ .plan();
// Build the per-bunch readers from the planned partial read row types.
for (int i = 0; i < numBunches; i++) {
@@ -349,9 +345,13 @@ private DataEvolutionFileReader createUnionReader(
long schemaId = firstFile.schemaId();
TableSchema dataSchema = bunchDataSchemas[i];
RowType partialReadRowType = new RowType(readFields);
+ List cacheKey =
+ nestedFieldEnabled
+ ? readerCacheKey(readFields, dataSchema.fields(), true)
+ : readFields.stream().map(DataField::name).collect(Collectors.toList());
FormatReaderMapping formatReaderMapping =
formatReaderMappings.computeIfAbsent(
- new FormatKey(schemaId, formatIdentifier, readerCacheKey(readFields)),
+ new FormatKey(schemaId, formatIdentifier, cacheKey),
key ->
formatBuilder.build(
formatIdentifier,
@@ -371,39 +371,54 @@ private DataEvolutionFileReader createUnionReader(
deletionVector));
}
- for (int j = 0; j < allReadFields.size(); j++) {
- if (plan.rowOffsets[j] == -1 && plan.nested[j] == null) {
- checkArgument(
- allReadFields.get(j).type().isNullable(),
- format(
- "Field %s is not null but can't find any file contains it.",
- allReadFields.get(j)));
+ return nestedFieldEnabled
+ ? new DataEvolutionFileReader(
+ plan.rowOffsets, plan.fieldOffsets, fileRecordReaders, plan.nested)
+ : new DataEvolutionFileReader(
+ plan.rowOffsets, plan.fieldOffsets, fileRecordReaders);
+ }
+
+ private boolean nestedFieldEnabledFor(List files) {
+ if (coreOptions.dataEvolutionNestedFieldEnabled()) {
+ return true;
+ }
+ for (DataFileMeta file : files) {
+ TableSchema fileSchema = schemaFetcher.apply(file.schemaId());
+ if (new CoreOptions(fileSchema.options()).dataEvolutionNestedFieldEnabled()) {
+ return true;
}
}
-
- return new DataEvolutionFileReader(
- plan.rowOffsets, plan.fieldOffsets, fileRecordReaders, plan.nested);
+ return false;
}
/**
- * A cache key describing the exact (possibly partially nested) fields read from one bunch. It
- * encodes field ids and nesting structure rather than names, so two bunches reading different
- * sub-fields of the same struct (e.g. {@code nest.a} vs {@code nest.b}) never collide.
+ * A cache key describing both the exact (possibly partially nested) fields requested from one
+ * bunch and the projected data schema it physically provides. Both affect schema-evolution
+ * casts, so two bunches reading the same projected shape from different sibling files must not
+ * share a mapping.
*
- * Deliberately not {@link RowType#leafPaths(RowType)}: that describes a written type
+ *
Deliberately not {@link RowType#collectLeafPaths(RowType)}: that describes a written type
* relative to the schema it was written against and therefore enforces the write-side
* restrictions (at most one level of partial nesting, no dotted names). A read type is not
* bound by those — it may be pruned arbitrarily deep by the engine, and it may be *wider* than
* the file's own schema after a nested {@code ADD COLUMN}.
*/
- private static List readerCacheKey(List readFields) {
- List key = new ArrayList<>(readFields.size());
- for (DataField field : readFields) {
+ private static List readerCacheKey(
+ List readFields, List dataFields, boolean nestedFieldEnabled) {
+ List key = new ArrayList<>(readFields.size() + dataFields.size() + 3);
+ key.add("nested=" + nestedFieldEnabled);
+ appendFieldsKey("read", readFields, key);
+ appendFieldsKey("data", dataFields, key);
+ return key;
+ }
+
+ private static void appendFieldsKey(String prefix, List fields, List key) {
+ key.add(prefix);
+ for (DataField field : fields) {
StringBuilder builder = new StringBuilder();
appendFieldKey(field, builder);
key.add(builder.toString());
}
- return key;
}
private static void appendFieldKey(DataField field, StringBuilder builder) {
@@ -523,6 +538,7 @@ private FileRecordReader createFileReader(
String formatIdentifier = readTarget.formatIdentifier;
long schemaId = file.schemaId();
TableSchema dataSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId);
+ boolean nestedFieldEnabled = nestedFieldEnabledFor(Collections.singletonList(file));
// no column merge here, so the filters this file can answer reach both the file index and
// the format reader
@@ -530,9 +546,13 @@ private FileRecordReader createFileReader(
FormatReaderMapping formatReaderMapping =
singleFileReaderMappings.computeIfAbsent(
new SingleFileKey(
- schemaId, formatIdentifier, file.writeCols(), readRowType),
+ schemaId,
+ formatIdentifier,
+ file.writeCols(),
+ readRowType,
+ nestedFieldEnabled),
key ->
- formatBuilder(readRowType, fileFilters)
+ formatBuilder(readRowType, fileFilters, nestedFieldEnabled)
.build(formatIdentifier, schema, dataSchema));
FileIndexResult fileIndexResult = null;
@@ -730,7 +750,8 @@ private boolean skipByFileIndex(
return false;
}
- private Builder formatBuilder(RowType readRowType, @Nullable List filters) {
+ private Builder formatBuilder(
+ RowType readRowType, @Nullable List filters, boolean nestedFieldEnabled) {
return new Builder(
formatDiscover,
readRowType.getFields(),
@@ -738,7 +759,8 @@ private Builder formatBuilder(RowType readRowType, @Nullable List fil
schema -> rowTypeWithRowTracking(schema.logicalRowType(), true, true).getFields(),
filters,
null,
- null);
+ null,
+ nestedFieldEnabled);
}
/**
@@ -945,7 +967,8 @@ private FileReadTarget(String formatIdentifier, Path path, long fileSize) {
* pushes down depend on the read type, which is not the same for every split: {@link
* IndexedSplitRecordReader#readInfo} adds a row id to the read type when the split carries
* scores. The columns the file wrote are part of the key as well, they decide which filters the
- * file can answer.
+ * file can answer. The effective nested mode can change for a retained reader after a persisted
+ * false-to-true table-option update, and it changes the schema-evolution mapping.
*/
private static class SingleFileKey {
@@ -953,16 +976,19 @@ private static class SingleFileKey {
private final String formatIdentifier;
@Nullable private final List writeCols;
private final RowType readRowType;
+ private final boolean nestedFieldEnabled;
private SingleFileKey(
long schemaId,
String formatIdentifier,
@Nullable List writeCols,
- RowType readRowType) {
+ RowType readRowType,
+ boolean nestedFieldEnabled) {
this.schemaId = schemaId;
this.formatIdentifier = formatIdentifier;
this.writeCols = writeCols;
this.readRowType = readRowType;
+ this.nestedFieldEnabled = nestedFieldEnabled;
}
@Override
@@ -975,6 +1001,7 @@ public boolean equals(Object o) {
}
SingleFileKey that = (SingleFileKey) o;
return schemaId == that.schemaId
+ && nestedFieldEnabled == that.nestedFieldEnabled
&& Objects.equals(formatIdentifier, that.formatIdentifier)
&& Objects.equals(writeCols, that.writeCols)
&& Objects.equals(readRowType, that.readRowType);
@@ -982,7 +1009,8 @@ public boolean equals(Object o) {
@Override
public int hashCode() {
- return Objects.hash(schemaId, formatIdentifier, writeCols, readRowType);
+ return Objects.hash(
+ schemaId, formatIdentifier, writeCols, readRowType, nestedFieldEnabled);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
index f5fe777b2972..0c79a11db288 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -85,6 +85,7 @@ public class RawFileSplitRead implements SplitRead {
private final Map formatReaderMappings;
private final boolean fileIndexReadEnabled;
private final boolean rowTrackingEnabled;
+ private final boolean nestedFieldEnabled;
private final boolean ignoreCorruptFiles;
private final boolean ignoreLostFiles;
@@ -112,6 +113,7 @@ public RawFileSplitRead(
this.ignoreCorruptFiles = coreOptions.scanIgnoreCorruptFile();
this.ignoreLostFiles = coreOptions.scanIgnoreLostFile();
this.rowTrackingEnabled = coreOptions.rowTrackingEnabled();
+ this.nestedFieldEnabled = coreOptions.dataEvolutionNestedFieldEnabled();
this.readRowType = rowType;
}
@@ -271,7 +273,8 @@ private Builder createFormatReaderMappingBuilder(
},
filters,
pushDownTopN,
- pushDownLimit);
+ pushDownLimit,
+ nestedFieldEnabled);
}
private ReaderSupplier createFileReader(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index b660d50ad481..1a202ac95eea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -118,6 +118,7 @@ public static ConflictDetection create(
BucketMode bucketMode,
boolean deletionVectorsEnabled,
boolean dataEvolutionEnabled,
+ boolean dataEvolutionNestedFieldEnabled,
boolean pkClusteringOverride,
IndexFileHandler indexFileHandler,
SnapshotManager snapshotManager,
@@ -130,6 +131,7 @@ public static ConflictDetection create(
pathFactory,
bucketMode,
deletionVectorsEnabled,
+ dataEvolutionNestedFieldEnabled,
indexFileHandler,
snapshotManager,
commitScanner);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
index bb6e429138c9..056c6b8dca86 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
@@ -72,6 +72,7 @@ public class DataEvolutionConflictDetection extends ConflictDetection {
private final String tableName;
private final String commitUser;
private final SnapshotManager snapshotManager;
+ private final boolean nestedFieldEnabled;
private @Nullable Long rowIdCheckFromSnapshot;
private @Nullable RowIdConflictCheckStrategy rowIdConflictCheckStrategy;
@@ -83,6 +84,7 @@ public DataEvolutionConflictDetection(
FileStorePathFactory pathFactory,
BucketMode bucketMode,
boolean deletionVectorsEnabled,
+ boolean nestedFieldEnabled,
IndexFileHandler indexFileHandler,
SnapshotManager snapshotManager,
CommitScanner commitScanner) {
@@ -98,6 +100,7 @@ public DataEvolutionConflictDetection(
this.tableName = tableName;
this.commitUser = commitUser;
this.snapshotManager = snapshotManager;
+ this.nestedFieldEnabled = nestedFieldEnabled;
}
@Override
@@ -132,7 +135,8 @@ public RowIdConflictChecker createRowIdConflictChecker(
if (!shouldCheckRowIdFromSnapshot(commitKind)) {
return null;
}
- return rowIdConflictCheckStrategy().createChecker(schemaManager, deltaFiles);
+ return rowIdConflictCheckStrategy()
+ .createChecker(schemaManager, deltaFiles, nestedFieldEnabled);
}
private RowIdConflictCheckStrategy rowIdConflictCheckStrategy() {
@@ -402,7 +406,9 @@ private interface RowIdConflictCheckStrategy {
boolean appliesTo(CommitKind commitKind);
RowIdConflictChecker createChecker(
- SchemaManager schemaManager, List deltaFiles);
+ SchemaManager schemaManager,
+ List deltaFiles,
+ boolean nestedFieldEnabled);
boolean shouldCheckHistoricalEntry(FileKind kind);
}
@@ -419,10 +425,13 @@ public boolean appliesTo(CommitKind commitKind) {
@Override
public RowIdConflictChecker createChecker(
- SchemaManager schemaManager, List deltaFiles) {
+ SchemaManager schemaManager,
+ List deltaFiles,
+ boolean nestedFieldEnabled) {
return RowIdColumnConflictChecker.fromDataFiles(
schemaManager,
- deltaFiles.stream().map(ManifestEntry::file).collect(Collectors.toList()));
+ deltaFiles.stream().map(ManifestEntry::file).collect(Collectors.toList()),
+ nestedFieldEnabled);
}
@Override
@@ -443,7 +452,9 @@ public boolean appliesTo(CommitKind commitKind) {
@Override
public RowIdConflictChecker createChecker(
- SchemaManager schemaManager, List deltaFiles) {
+ SchemaManager schemaManager,
+ List deltaFiles,
+ boolean nestedFieldEnabled) {
// Materializing deletion vectors rewrites complete row ranges. A concurrent ADD in a
// deleted normal-file range can otherwise restore logically deleted rows.
List deletedNormalFiles =
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
index 272413dc8f05..ce32b5dabbb0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
@@ -49,18 +49,25 @@
*/
public class RowIdColumnConflictChecker implements RowIdConflictChecker {
- private final SchemaManager schemaManager;
+ private final WriteFieldIdResolver fieldIdResolver;
private final List writeRanges;
- private final Map rowTypeCache = new HashMap<>();
- private RowIdColumnConflictChecker(SchemaManager schemaManager, List deltaFiles) {
- this.schemaManager = schemaManager;
+ private RowIdColumnConflictChecker(
+ SchemaManager schemaManager,
+ List deltaFiles,
+ boolean nestedFieldEnabled) {
+ this.fieldIdResolver =
+ nestedFieldEnabled
+ ? new NestedFieldIdResolver(schemaManager)
+ : new TopLevelFieldIdResolver(schemaManager);
this.writeRanges = buildWriteRanges(deltaFiles);
}
public static RowIdColumnConflictChecker fromDataFiles(
- SchemaManager schemaManager, List deltaFiles) {
- return new RowIdColumnConflictChecker(schemaManager, deltaFiles);
+ SchemaManager schemaManager,
+ List deltaFiles,
+ boolean nestedFieldEnabled) {
+ return new RowIdColumnConflictChecker(schemaManager, deltaFiles, nestedFieldEnabled);
}
private List buildWriteRanges(List deltaFiles) {
@@ -97,13 +104,12 @@ private List buildWriteRanges(List deltaFiles) {
private void addWriteFieldIds(Set fieldIds, DataFileMeta file) {
List writeCols = file.writeCols();
if (writeCols == null) {
- // full-schema write touches every leaf field
- collectLeafIds(rowType(file.schemaId()).getFields(), fieldIds);
+ fieldIdResolver.addAllFieldIds(file.schemaId(), fieldIds);
return;
}
for (String writeCol : writeCols) {
- fieldIds.addAll(leafFieldIds(file.schemaId(), writeCol));
+ fieldIds.addAll(writeFieldIds(file.schemaId(), writeCol));
}
}
@@ -183,7 +189,7 @@ private boolean containsAnyWriteField(Set fieldIds, DataFileMeta file)
}
for (String writeCol : writeCols) {
- for (Integer fieldId : leafFieldIds(file.schemaId(), writeCol)) {
+ for (Integer fieldId : writeFieldIds(file.schemaId(), writeCol)) {
if (fieldIds.contains(fieldId)) {
return true;
}
@@ -198,24 +204,11 @@ private boolean containsAnyWriteField(Set fieldIds, DataFileMeta file)
* its leaf ids, so a whole-struct write and a sub-field write of the same struct still
* conflict.
*/
- private List leafFieldIds(long schemaId, String writeCol) {
+ private List writeFieldIds(long schemaId, String writeCol) {
if (SpecialFields.isSystemField(writeCol)) {
return Collections.emptyList();
}
- // projectByPaths handles both plain top-level names and dotted nested paths, and throws if
- // the path does not exist in the schema
- RowType projected;
- try {
- projected = rowType(schemaId).projectByPaths(Collections.singletonList(writeCol));
- } catch (IllegalArgumentException e) {
- throw new RuntimeException(
- String.format(
- "Cannot find write column '%s' in schema %s.", writeCol, schemaId),
- e);
- }
- List ids = new ArrayList<>();
- collectLeafIds(projected.getFields(), ids);
- return ids;
+ return fieldIdResolver.resolve(schemaId, writeCol);
}
private static void collectLeafIds(List fields, java.util.Collection out) {
@@ -228,9 +221,87 @@ private static void collectLeafIds(List fields, java.util.Collection<
}
}
- private RowType rowType(long schemaId) {
- return rowTypeCache.computeIfAbsent(
- schemaId, id -> schemaManager.schema(id).logicalRowType());
+ private static RuntimeException unknownWriteColumn(
+ long schemaId, String writeCol, Throwable cause) {
+ return new RuntimeException(
+ String.format("Cannot find write column '%s' in schema %s.", writeCol, schemaId),
+ cause);
+ }
+
+ private interface WriteFieldIdResolver {
+
+ void addAllFieldIds(long schemaId, Set fieldIds);
+
+ List resolve(long schemaId, String writeCol);
+ }
+
+ private static class TopLevelFieldIdResolver implements WriteFieldIdResolver {
+
+ private final SchemaManager schemaManager;
+ private final Map> fieldIdByNameCache = new HashMap<>();
+
+ private TopLevelFieldIdResolver(SchemaManager schemaManager) {
+ this.schemaManager = schemaManager;
+ }
+
+ @Override
+ public void addAllFieldIds(long schemaId, Set fieldIds) {
+ fieldIds.addAll(fieldIdByName(schemaId).values());
+ }
+
+ @Override
+ public List resolve(long schemaId, String writeCol) {
+ Integer fieldId = fieldIdByName(schemaId).get(writeCol);
+ if (fieldId == null) {
+ throw unknownWriteColumn(schemaId, writeCol, null);
+ }
+ return Collections.singletonList(fieldId);
+ }
+
+ private Map fieldIdByName(long schemaId) {
+ return fieldIdByNameCache.computeIfAbsent(
+ schemaId,
+ id ->
+ schemaManager.schema(id).fields().stream()
+ .collect(Collectors.toMap(DataField::name, DataField::id)));
+ }
+ }
+
+ private static class NestedFieldIdResolver implements WriteFieldIdResolver {
+
+ private final SchemaManager schemaManager;
+ private final Map rowTypeCache = new HashMap<>();
+
+ private NestedFieldIdResolver(SchemaManager schemaManager) {
+ this.schemaManager = schemaManager;
+ }
+
+ @Override
+ public void addAllFieldIds(long schemaId, Set fieldIds) {
+ // A full-schema write touches every leaf field when nested data evolution is in use, so
+ // it conflicts with any partial sub-field write.
+ collectLeafIds(rowType(schemaId).getFields(), fieldIds);
+ }
+
+ @Override
+ public List resolve(long schemaId, String writeCol) {
+ // projectByPaths handles both plain top-level names and dotted nested paths, and throws
+ // if the path does not exist in the schema.
+ RowType projected;
+ try {
+ projected = rowType(schemaId).projectByPaths(Collections.singletonList(writeCol));
+ } catch (IllegalArgumentException e) {
+ throw unknownWriteColumn(schemaId, writeCol, e);
+ }
+ List ids = new ArrayList<>();
+ collectLeafIds(projected.getFields(), ids);
+ return ids;
+ }
+
+ private RowType rowType(long schemaId) {
+ return rowTypeCache.computeIfAbsent(
+ schemaId, id -> schemaManager.schema(id).logicalRowType());
+ }
}
/** Range and field id Set. */
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 1a76668448b7..61c9c03a0ea3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -1382,6 +1382,18 @@ public static void checkAlterTableOption(
String.format("Change '%s' is not supported yet.", key));
}
+ if (CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key().equals(key)) {
+ boolean oldEnabled =
+ oldValue == null
+ ? CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.defaultValue()
+ : Boolean.parseBoolean(oldValue);
+ boolean newEnabled = Boolean.parseBoolean(newValue);
+ if (oldEnabled && !newEnabled) {
+ throw new UnsupportedOperationException(
+ String.format("Cannot disable table option '%s'.", key));
+ }
+ }
+
if (CoreOptions.BUCKET.key().equals(key)) {
int oldBucket =
oldValue == null
@@ -1474,6 +1486,14 @@ public static void checkResetTableOption(Map options, String key
DELETION_VECTORS_ENABLED.defaultValue().toString());
}
+ if (CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key().equals(key)) {
+ checkAlterTableOption(
+ options,
+ key,
+ options.get(key),
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.defaultValue().toString());
+ }
+
if (IGNORE_DELETE.key().equals(key)) {
checkAlterTableOption(
options, key, options.get(key), IGNORE_DELETE.defaultValue().toString());
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index e370e674aebe..996fceaa97bf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -1592,6 +1592,14 @@ private static void validateRowTracking(TableSchema schema, CoreOptions options)
"Data evolution config must disabled with clustering.incremental");
}
+ if (options.dataEvolutionNestedFieldEnabled()) {
+ checkArgument(
+ options.dataEvolutionEnabled(),
+ "%s requires %s=true.",
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(),
+ CoreOptions.DATA_EVOLUTION_ENABLED.key());
+ }
+
List fields = schema.fields();
List blobNames =
fields.stream()
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
index 9d665c0eacf6..e1fae54ee97f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
@@ -19,10 +19,10 @@
package org.apache.paimon.utils;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
@@ -52,9 +52,12 @@ public class DataEvolutionUtils {
* Collect exact written field ids; an empty list is exact and an empty optional is unresolved.
*/
public static Optional> collectWrittenColumnIds(
- Collection splits, Function schemaLoader) {
+ Collection splits,
+ Function> schemaFieldsLoader,
+ Function nestedFieldEnabledLoader) {
Set fieldIds = new TreeSet<>();
Map> schemaFieldsCache = new HashMap<>();
+ Map nestedFieldEnabledCache = new HashMap<>();
Map>, Set> fieldIdsCache = new HashMap<>();
try {
for (DataSplit split : splits) {
@@ -66,14 +69,19 @@ public static Optional> collectWrittenColumnIds(
schemaFieldsCache.computeIfAbsent(
file.schemaId(),
schemaId -> {
- TableSchema schema = schemaLoader.apply(schemaId);
+ List loaded =
+ schemaFieldsLoader.apply(schemaId);
checkArgument(
- schema != null,
+ loaded != null,
"Cannot find schema %s.",
schemaId);
- return schema.fields();
+ return loaded;
});
- fileFieldIds = resolveFileFieldIds(schemaFields, file, true);
+ boolean nestedFieldEnabled =
+ nestedFieldEnabledCache.computeIfAbsent(
+ file.schemaId(), nestedFieldEnabledLoader);
+ fileFieldIds =
+ resolveFileFieldIds(schemaFields, file, nestedFieldEnabled, true);
fieldIdsCache.put(cacheKey, fileFieldIds);
}
fieldIds.addAll(fileFieldIds);
@@ -89,12 +97,15 @@ public static Optional> collectWrittenColumnIds(
* Table field ids physically present in a file, resolved through the schema used to write it.
*/
public static Set fileFieldIds(
- Function scanTableSchema, DataFileMeta file) {
- return resolveFileFieldIds(scanTableSchema.apply(file.schemaId()).fields(), file, false);
+ List schemaFields, DataFileMeta file, boolean nestedFieldEnabled) {
+ return resolveFileFieldIds(schemaFields, file, nestedFieldEnabled, false);
}
private static Set resolveFileFieldIds(
- List schemaFields, DataFileMeta file, boolean strict) {
+ List schemaFields,
+ DataFileMeta file,
+ boolean nestedFieldEnabled,
+ boolean strict) {
List writeCols = file.writeCols();
Set ids = new HashSet<>();
if (writeCols == null) {
@@ -115,7 +126,7 @@ private static Set resolveFileFieldIds(
// same top-level field. Try the exact name first so a column whose own name contains a
// dot is not split, matching RowType#projectByPaths.
DataField field = byName.get(writeCol);
- if (field == null) {
+ if (field == null && nestedFieldEnabled) {
int dot = writeCol.indexOf('.');
if (dot > 0) {
field = byName.get(writeCol.substring(0, dot));
@@ -142,15 +153,32 @@ private static Set resolveFileFieldIds(
/** Table fields physically present in a file, in their physical write order. */
public static List fileFields(
- Function scanTableSchema, DataFileMeta file) {
- TableSchema schema = scanTableSchema.apply(file.schemaId());
+ List schemaFields, DataFileMeta file, boolean nestedFieldEnabled) {
List writeCols = file.writeCols();
if (writeCols == null) {
- return schema.fields();
+ return schemaFields;
+ }
+
+ if (nestedFieldEnabled) {
+ Set fieldNames =
+ schemaFields.stream().map(DataField::name).collect(Collectors.toSet());
+ List tableWriteCols =
+ writeCols.stream()
+ .filter(
+ writeCol -> {
+ if (fieldNames.contains(writeCol)) {
+ return true;
+ }
+ int dot = writeCol.indexOf('.');
+ return dot > 0
+ && fieldNames.contains(writeCol.substring(0, dot));
+ })
+ .collect(Collectors.toList());
+ return new RowType(schemaFields).projectByPaths(tableWriteCols).getFields();
}
Map fieldsByName = new HashMap<>();
- for (DataField field : schema.fields()) {
+ for (DataField field : schemaFields) {
fieldsByName.put(field.name(), field);
}
List fields = new ArrayList<>();
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java b/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
index e583f389d079..8ee5f2dd7ced 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
@@ -42,6 +42,7 @@
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -164,6 +165,7 @@ public static class Builder {
@Nullable private final List filters;
@Nullable private final TopN topN;
@Nullable private final Integer limit;
+ private final boolean nestedFieldEnabled;
public Builder(
FileFormatDiscover formatDiscover,
@@ -172,12 +174,24 @@ public Builder(
@Nullable List filters,
@Nullable TopN topN,
@Nullable Integer limit) {
+ this(formatDiscover, readFields, fieldsExtractor, filters, topN, limit, false);
+ }
+
+ public Builder(
+ FileFormatDiscover formatDiscover,
+ List readFields,
+ Function> fieldsExtractor,
+ @Nullable List filters,
+ @Nullable TopN topN,
+ @Nullable Integer limit,
+ boolean nestedFieldEnabled) {
this.formatDiscover = formatDiscover;
this.readFields = readFields;
this.fieldsExtractor = fieldsExtractor;
this.filters = filters;
this.topN = topN;
this.limit = limit;
+ this.nestedFieldEnabled = nestedFieldEnabled;
}
/**
@@ -212,7 +226,11 @@ public FormatReaderMapping build(
Set selectedKeysFieldIds = selectedKeysFieldIds(tableSchema, expectedFields);
List readDataFields =
- readDataFields(allDataFieldsInFile, expectedFields, selectedKeysFieldIds);
+ readDataFields(
+ allDataFieldsInFile,
+ expectedFields,
+ selectedKeysFieldIds,
+ nestedFieldEnabled);
IndexCastMapping indexCastMapping =
SchemaEvolutionUtil.createIndexCastMapping(expectedFields, readDataFields);
@@ -324,6 +342,14 @@ private List readDataFields(
List allDataFields,
List expectedFields,
Set selectedKeysFieldIds) {
+ return readDataFields(allDataFields, expectedFields, selectedKeysFieldIds, false);
+ }
+
+ private List readDataFields(
+ List allDataFields,
+ List expectedFields,
+ Set selectedKeysFieldIds,
+ boolean nestedFieldEnabled) {
List readDataFields = new ArrayList<>();
for (DataField dataField : allDataFields) {
expectedFields.stream()
@@ -338,7 +364,10 @@ private List readDataFields(
}
DataType prunedType =
- pruneDataType(field.type(), dataField.type());
+ pruneDataType(
+ field.type(),
+ dataField.type(),
+ nestedFieldEnabled);
if (prunedType != null) {
readDataFields.add(dataField.newType(prunedType));
}
@@ -402,7 +431,8 @@ private void validateSelectedKeyValueTypes(DataField expectedField, DataField ta
}
@Nullable
- private DataType pruneDataType(DataType readType, DataType dataType) {
+ private DataType pruneDataType(
+ DataType readType, DataType dataType, boolean nestedFieldEnabled) {
switch (readType.getTypeRoot()) {
case ROW:
RowType r = (RowType) readType;
@@ -414,7 +444,8 @@ private DataType pruneDataType(DataType readType, DataType dataType) {
for (DataField rf : r.getFields()) {
if (d.containsField(rf.id())) {
DataField df = d.getField(rf.id());
- DataType newType = pruneDataType(rf.type(), df.type());
+ DataType newType =
+ pruneDataType(rf.type(), df.type(), nestedFieldEnabled);
if (newType == null) {
continue;
}
@@ -422,19 +453,26 @@ private DataType pruneDataType(DataType readType, DataType dataType) {
}
}
if (newFields.isEmpty()) {
- // When all fields are pruned, we should not return an empty row type
- return null;
+ // Every requested child may have been added after this file was written.
+ // Keep one physical child as a hidden anchor so the format reader can
+ // preserve the ROW's nullness and row count; the schema-evolution cast
+ // projects it away and null-fills the requested children.
+ return !nestedFieldEnabled || d.getFields().isEmpty()
+ ? null
+ : d.copy(Collections.singletonList(d.getFields().get(0)));
}
return d.copy(newFields);
case MAP:
DataType keyType =
pruneDataType(
((MapType) readType).getKeyType(),
- ((MapType) dataType).getKeyType());
+ ((MapType) dataType).getKeyType(),
+ nestedFieldEnabled);
DataType valueType =
pruneDataType(
((MapType) readType).getValueType(),
- ((MapType) dataType).getValueType());
+ ((MapType) dataType).getValueType(),
+ nestedFieldEnabled);
if (keyType == null || valueType == null) {
return null;
}
@@ -443,7 +481,8 @@ private DataType pruneDataType(DataType readType, DataType dataType) {
DataType elementType =
pruneDataType(
((ArrayType) readType).getElementType(),
- ((ArrayType) dataType).getElementType());
+ ((ArrayType) dataType).getElementType(),
+ nestedFieldEnabled);
if (elementType == null) {
return null;
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
index abf036b97497..dd91e37f3a48 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
@@ -25,6 +25,7 @@
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -204,7 +205,10 @@ private DataFileMeta updateColumnsAndCompact(
}
private long columnSequence(DataFileMeta file, int fieldId) throws Exception {
- List fields = fileFields(getTableDefault().schemaManager()::schema, file);
+ TableSchema fileSchema = getTableDefault().schemaManager().schema(file.schemaId());
+ boolean nestedFieldEnabled =
+ new CoreOptions(fileSchema.options()).dataEvolutionNestedFieldEnabled();
+ List fields = fileFields(fileSchema.fields(), file, nestedFieldEnabled);
long[] sequences = file.columnMaxSequenceNumbers();
assertThat(sequences).hasSize(fields.size());
for (int i = 0; i < fields.size(); i++) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadPlannerTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadPlannerTest.java
index 4fc365c179aa..9d0a167acf11 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadPlannerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadPlannerTest.java
@@ -51,6 +51,51 @@ private static RowType nest(DataField... subFields) {
Collections.singletonList(new DataField(1, "nest", DataTypes.ROW(subFields))));
}
+ @Test
+ void testTopLevelPlanningKeepsLegacyWholeFieldSelection() {
+ RowType avail0 = nest(new DataField(2, "a", DataTypes.INT()));
+ RowType avail1 =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(
+ 1,
+ "nest",
+ DataTypes.ROW(new DataField(3, "b", DataTypes.STRING())))));
+
+ DataEvolutionReadPlan plan =
+ new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1), false)
+ .plan();
+
+ assertThat(plan.nested).containsOnlyNulls();
+ assertThat(plan.rowOffsets).containsExactly(1, 0);
+ assertThat(plan.fieldOffsets).containsExactly(0, 0);
+ assertThat(plan.bunchReadFields.get(0)).containsExactly(READ_TYPE.getField(1));
+ assertThat(plan.bunchReadFields.get(1)).containsExactly(READ_TYPE.getField(0));
+ }
+
+ @Test
+ void testMissingNonNullFieldIsRejectedInBothPlanningModes() {
+ RowType readType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "id", DataTypes.INT().notNull())));
+ RowType unrelated =
+ new RowType(Collections.singletonList(new DataField(1, "other", DataTypes.INT())));
+
+ for (boolean nestedFieldEnabled : Arrays.asList(false, true)) {
+ assertThatThrownBy(
+ () ->
+ new DataEvolutionReadPlanner(
+ readType,
+ Collections.singletonList(unrelated),
+ nestedFieldEnabled)
+ .plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("id");
+ }
+ }
+
@Test
void testStructSplitAcrossFilesIsComposed() {
// bunch0 (latest) provides nest.a; bunch1 provides id + nest.b
@@ -65,7 +110,7 @@ void testStructSplitAcrossFilesIsComposed() {
DataTypes.ROW(new DataField(3, "b", DataTypes.STRING())))));
DataEvolutionReadPlan plan =
- new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1)).plan();
+ new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1), true).plan();
// id is taken whole from bunch1
assertThat(plan.nested[0]).isNull();
@@ -89,7 +134,7 @@ void testStructWholeFromSingleFile() {
new RowType(Collections.singletonList(new DataField(0, "id", DataTypes.INT())));
DataEvolutionReadPlan plan =
- new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1)).plan();
+ new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1), true).plan();
// nest is taken whole from bunch0, not composed
assertThat(plan.nested[1]).isNull();
@@ -105,7 +150,7 @@ void testSubFieldAbsentEverywhereStaysNullWhenNullable() {
new RowType(Collections.singletonList(new DataField(0, "id", DataTypes.INT())));
DataEvolutionReadPlan plan =
- new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1)).plan();
+ new DataEvolutionReadPlanner(READ_TYPE, Arrays.asList(avail0, avail1), true).plan();
// nest still composed (only a present), no exception since b is nullable
assertThat(plan.nested[1]).isNotNull();
@@ -166,7 +211,7 @@ void testDeeperThanOneLevelSplitThrows() {
assertThatThrownBy(
() ->
new DataEvolutionReadPlanner(
- readType, Arrays.asList(avail0, avail1))
+ readType, Arrays.asList(avail0, avail1), true)
.plan())
.isInstanceOf(UnsupportedOperationException.class);
}
@@ -215,7 +260,7 @@ void testDeepAddColumnIsNullFilledInsteadOfRejected() {
DataTypes.INT())))))));
DataEvolutionReadPlan plan =
- new DataEvolutionReadPlanner(readType, Arrays.asList(avail0, avail1)).plan();
+ new DataEvolutionReadPlanner(readType, Arrays.asList(avail0, avail1), true).plan();
// id comes whole from the latest partial file
assertThat(plan.rowOffsets[0]).isEqualTo(0);
@@ -223,4 +268,140 @@ void testDeepAddColumnIsNullFilledInsteadOfRejected() {
// by schema evolution rather than rejected as an unsupported deep split.
assertThat(plan.bunchReadFields.get(1)).anySatisfy(f -> assertThat(f.id()).isEqualTo(1));
}
+
+ @Test
+ void testProjectedAddedLeafUsesParentAsReaderAnchor() {
+ // Only payload.y is projected. The old file predates y, but its payload field still has to
+ // be read so schema evolution can null-fill y and the union reader keeps row cardinality.
+ RowType readType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ 1,
+ "payload",
+ new RowType(
+ false,
+ Collections.singletonList(
+ new DataField(3, "y", DataTypes.INT()))))));
+ RowType unrelatedUpdate =
+ new RowType(Collections.singletonList(new DataField(0, "id", DataTypes.INT())));
+ RowType oldFile =
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ 1,
+ "payload",
+ new RowType(
+ false,
+ Collections.singletonList(
+ new DataField(2, "x", DataTypes.INT()))))));
+
+ DataEvolutionReadPlan plan =
+ new DataEvolutionReadPlanner(
+ readType, Arrays.asList(unrelatedUpdate, oldFile), true)
+ .plan();
+
+ assertThat(plan.rowOffsets[0]).isEqualTo(-1);
+ assertThat(plan.nested[0]).isNotNull();
+ assertThat(plan.bunchReadFields.get(0)).isEmpty();
+ assertThat(plan.bunchReadFields.get(1)).containsExactly(readType.getFields().get(0));
+ }
+
+ @Test
+ void testProjectedAddedLeafUsesAllWinningSiblingProvidersAsAnchors() {
+ RowType readType =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(5, "added", DataTypes.INT()))));
+ RowType latestX =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(2, "x", DataTypes.INT()))));
+ RowType latestZ =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(4, "z", DataTypes.INT()))));
+ RowType staleX =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(2, "x", DataTypes.INT()))));
+
+ DataEvolutionReadPlan plan =
+ new DataEvolutionReadPlanner(
+ readType, Arrays.asList(latestX, latestZ, staleX), true)
+ .plan();
+
+ assertThat(plan.rowOffsets[0]).isEqualTo(-1);
+ assertThat(plan.nested[0]).isNotNull();
+ assertThat(plan.bunchReadFields.get(0)).containsExactly(readType.getFields().get(0));
+ assertThat(plan.bunchReadFields.get(1)).containsExactly(readType.getFields().get(0));
+ assertThat(plan.bunchReadFields.get(2)).isEmpty();
+ }
+
+ @Test
+ void testProjectedExistingLeafUsesAllWinningSiblingProvidersAsAnchors() {
+ RowType readType =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(2, "x", DataTypes.INT()))));
+ RowType latestX =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(2, "x", DataTypes.INT()))));
+ RowType latestZ =
+ rowType(
+ new DataField(
+ 1, "payload", rowType(new DataField(4, "z", DataTypes.INT()))));
+
+ DataEvolutionReadPlan plan =
+ new DataEvolutionReadPlanner(readType, Arrays.asList(latestX, latestZ), true)
+ .plan();
+
+ assertThat(plan.rowOffsets[0]).isEqualTo(-1);
+ assertThat(plan.nested[0]).isNotNull();
+ assertThat(plan.bunchReadFields.get(0)).containsExactly(readType.getFields().get(0));
+ assertThat(plan.bunchReadFields.get(1)).containsExactly(readType.getFields().get(0));
+ }
+
+ @Test
+ void testProjectedDeepAddedLeafUsesSiblingUnderSameParent() {
+ DataField projectedSub =
+ new DataField(2, "sub", rowType(new DataField(4, "added", DataTypes.INT())));
+ RowType readType = rowType(new DataField(1, "payload", rowType(projectedSub)));
+ RowType existingSub =
+ rowType(
+ new DataField(
+ 1,
+ "payload",
+ rowType(
+ new DataField(
+ 2,
+ "sub",
+ rowType(
+ new DataField(
+ 3,
+ "existing",
+ DataTypes.INT()))))));
+ RowType otherSibling =
+ rowType(
+ new DataField(
+ 1,
+ "payload",
+ rowType(new DataField(5, "other", DataTypes.STRING()))));
+
+ DataEvolutionReadPlan plan =
+ new DataEvolutionReadPlanner(
+ readType, Arrays.asList(existingSub, otherSibling), true)
+ .plan();
+
+ assertThat(plan.rowOffsets[0]).isEqualTo(-1);
+ assertThat(plan.nested[0]).isNotNull();
+ RowType subProviderReadType = (RowType) plan.bunchReadFields.get(0).get(0).type();
+ assertThat(subProviderReadType.getFields()).containsExactly(projectedSub);
+ assertThat(plan.bunchReadFields.get(1)).containsExactly(readType.getFields().get(0));
+ }
+
+ private static RowType rowType(DataField field) {
+ return new RowType(Collections.singletonList(field));
+ }
}
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 f6d150d02335..1e1f0b225f82 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
@@ -2062,6 +2062,7 @@ private FileStoreCommitImpl newCommitWithSnapshotCommit(
store.bucketMode(),
options.deletionVectorsEnabled(),
dataEvolutionEnabled,
+ options.dataEvolutionNestedFieldEnabled(),
options.pkClusteringOverride(),
store.newIndexFileHandler(),
store.snapshotManager(),
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
index a0bb1459bd00..787d0761ea3e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
@@ -1582,6 +1582,7 @@ private ConflictDetection createConflictDetection(
BucketMode.HASH_FIXED,
false,
dataEvolutionEnabled,
+ false,
pkClusteringOverride,
null,
snapshotManager,
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdColumnConflictCheckerTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdColumnConflictCheckerTest.java
index 76ef065b6268..8c3649900bef 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdColumnConflictCheckerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdColumnConflictCheckerTest.java
@@ -118,7 +118,7 @@ void testFailsOnUnknownNonSystemWriteColumn() {
void testSubFieldDisjointLeavesDoNotConflict() {
// schema 2: id INT, nest ROW
RowIdColumnConflictChecker checker =
- checker(file("current", 0L, 10L, 2L, Arrays.asList("nest.a")));
+ nestedChecker(file("current", 0L, 10L, 2L, Arrays.asList("nest.a")));
assertThat(checker.conflictsWith(file("historical", 0L, 10L, 2L, Arrays.asList("nest.b"))))
.isFalse();
@@ -127,7 +127,7 @@ void testSubFieldDisjointLeavesDoNotConflict() {
@Test
void testSubFieldSameLeafConflicts() {
RowIdColumnConflictChecker checker =
- checker(file("current", 0L, 10L, 2L, Arrays.asList("nest.a")));
+ nestedChecker(file("current", 0L, 10L, 2L, Arrays.asList("nest.a")));
assertThat(checker.conflictsWith(file("historical", 0L, 10L, 2L, Arrays.asList("nest.a"))))
.isTrue();
@@ -137,7 +137,7 @@ void testSubFieldSameLeafConflicts() {
void testWholeStructConflictsWithSubField() {
// a whole-struct write expands to all of its leaves, so it conflicts with a sub-field write
RowIdColumnConflictChecker checker =
- checker(file("current", 0L, 10L, 2L, Arrays.asList("nest")));
+ nestedChecker(file("current", 0L, 10L, 2L, Arrays.asList("nest")));
assertThat(checker.conflictsWith(file("historical", 0L, 10L, 2L, Arrays.asList("nest.a"))))
.isTrue();
@@ -145,15 +145,33 @@ void testWholeStructConflictsWithSubField() {
@Test
void testFullSchemaWriteConflictsWithSubField() {
- RowIdColumnConflictChecker checker = checker(file("current", 0L, 10L, 2L, null));
+ RowIdColumnConflictChecker checker = nestedChecker(file("current", 0L, 10L, 2L, null));
assertThat(checker.conflictsWith(file("historical", 0L, 10L, 2L, Arrays.asList("nest.a"))))
.isTrue();
}
+ @Test
+ void testNestedWriteColumnIsUnknownWhenOptionDisabled() {
+ RowIdColumnConflictChecker checker =
+ checker(file("current", 0L, 10L, 2L, Arrays.asList("nest")));
+
+ assertThatThrownBy(
+ () ->
+ checker.conflictsWith(
+ file("historical", 0L, 10L, 2L, Arrays.asList("nest.a"))))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Cannot find write column 'nest.a'");
+ }
+
private RowIdColumnConflictChecker checker(DataFileMeta... files) {
return RowIdColumnConflictChecker.fromDataFiles(
- createSchemaManager(), Arrays.asList(files));
+ createSchemaManager(), Arrays.asList(files), false);
+ }
+
+ private RowIdColumnConflictChecker nestedChecker(DataFileMeta... files) {
+ return RowIdColumnConflictChecker.fromDataFiles(
+ createSchemaManager(), Arrays.asList(files), true);
}
private DataFileMeta file(
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 14ea3c37bfe2..c9588f4552fa 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -1442,6 +1442,32 @@ void testRowTrackingWithPkTable() {
.hasMessageContaining("primary-key");
}
+ @Test
+ void testNestedFieldDataEvolutionRequiresDataEvolution() {
+ Map options = new HashMap<>();
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+ options.put(BUCKET.key(), String.valueOf(-1));
+
+ List fields =
+ Arrays.asList(
+ new DataField(0, "f0", DataTypes.INT()),
+ new DataField(
+ 1,
+ "nest",
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "a", DataTypes.INT()),
+ DataTypes.FIELD(3, "b", DataTypes.STRING()))));
+ TableSchema schema = new TableSchema(1, fields, 10, emptyList(), emptyList(), options, "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key())
+ .hasMessageContaining(CoreOptions.DATA_EVOLUTION_ENABLED.key());
+
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
@Test
public void testFileIndexColumns() {
List keys =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/NestedDataEvolutionTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/NestedDataEvolutionTableTest.java
index 4fcf1d9790af..bc80b914b56a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/NestedDataEvolutionTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/NestedDataEvolutionTableTest.java
@@ -29,6 +29,7 @@
import org.apache.paimon.reader.DataEvolutionFileReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -271,6 +272,70 @@ public void testProjectionWithNested() throws Exception {
assertThat(readRowIds()).hasSize(n);
}
+ @Test
+ public void testProjectedAddedNestedLeafKeepsRowCardinality() throws Exception {
+ createTableDefault();
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.setOption(
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true"),
+ false);
+ RowType originalType = getTableDefault().rowType();
+ int n = 3;
+
+ BatchWriteBuilder builder = getTableDefault().newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite().withWriteType(originalType)) {
+ for (int i = 0; i < n; i++) {
+ write.write(
+ GenericRow.of(
+ i,
+ BinaryString.fromString("old" + i),
+ i == n - 1 ? null : nestOf(i, "n" + i),
+ arrOf(i),
+ mapOf("k" + i, i)));
+ }
+ builder.newCommit().commit(write.prepareCommit());
+ }
+
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn(new String[] {"nest", "c"}, DataTypes.INT(), null, null),
+ false);
+ FileStoreTable table = getTableDefault();
+ builder = table.newBatchWriteBuilder();
+ RowType updatedColumn = table.rowType().project(Collections.singletonList("f1"));
+ try (BatchTableWrite write = builder.newWrite().withWriteType(updatedColumn)) {
+ for (int i = 0; i < n; i++) {
+ write.write(GenericRow.of(BinaryString.fromString("new" + i)));
+ }
+ List messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ builder.newCommit().commit(messages);
+ }
+
+ RowType readType = table.rowType().projectByPaths(Collections.singletonList("nest.c"));
+ ReadBuilder readBuilder = table.newReadBuilder().withReadType(readType);
+ try (RecordReader reader =
+ readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ RecordReader.RecordIterator batch = reader.readBatch();
+ assertThat(batch).isNotNull();
+ for (int i = 0; i < n; i++) {
+ InternalRow row = batch.next();
+ assertThat(row).isNotNull();
+ if (i == n - 1) {
+ assertThat(row.isNullAt(0)).isTrue();
+ } else {
+ InternalRow nest = row.getRow(0, 1);
+ assertThat(nest).isNotNull();
+ assertThat(nest.isNullAt(0)).isTrue();
+ }
+ }
+ assertThat(batch.next()).isNull();
+ batch.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+ }
+
/**
* G5 (limitation): a sub-field of a nested ROW cannot be addressed as a top-level write column,
* because data evolution splits at top-level column granularity. {@code RowType.project}
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 7f8e1b7d38dc..3adea522b0f4 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
@@ -28,6 +28,7 @@
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -49,6 +50,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests for sub-field-level data evolution + row-tracking: updating a single sub-field of a
@@ -73,6 +75,7 @@ protected Schema schemaDefault() {
b.column("mp", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()));
b.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
b.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ b.option(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
return b.build();
}
@@ -92,6 +95,254 @@ private void commit(BatchWriteBuilder builder, List messages) thr
}
}
+ @Test
+ public void testNestedFieldOptionCanBePersistentlyEnabledAfterWrite() throws Exception {
+ Schema disabledSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "nest",
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "a", DataTypes.INT()),
+ DataTypes.FIELD(1, "b", DataTypes.STRING())))
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .build();
+ catalog.createTable(identifier(), disabledSchema, false);
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite().withWriteType(table.rowType())) {
+ write.write(GenericRow.of(1, GenericRow.of(10, BinaryString.fromString("old"))));
+ commit(builder, write.prepareCommit());
+ }
+ // Keep a reader created from the old table object. Long-running engines may retain it
+ // across the persisted false -> true option change.
+ ReadBuilder staleReadBuilder = table.newReadBuilder();
+
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.setOption(
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true"),
+ false);
+ table = getTableDefault();
+ RowType partialType = table.rowType().projectByPaths(Collections.singletonList("nest.a"));
+ builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite().withWriteType(partialType)) {
+ write.write(GenericRow.of(GenericRow.of(20)));
+ List messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ commit(builder, messages);
+ }
+
+ try (RecordReader reader =
+ staleReadBuilder.newRead().createReader(staleReadBuilder.newScan().plan())) {
+ RecordReader.RecordIterator batch = reader.readBatch();
+ assertThat(batch).isNotNull();
+ InternalRow row = batch.next();
+ assertThat(row).isNotNull();
+ assertThat(row.getInt(0)).isEqualTo(1);
+ InternalRow nest = row.getRow(1, 2);
+ assertThat(nest.getInt(0)).isEqualTo(20);
+ assertThat(nest.getString(1).toString()).isEqualTo("old");
+ assertThat(batch.next()).isNull();
+ batch.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+ }
+
+ @Test
+ public void testDisabledOptionAllowsTopLevelColumnContainingDot() throws Exception {
+ Schema disabledSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("literal.dot", DataTypes.INT())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .build();
+ catalog.createTable(identifier(), disabledSchema, false);
+ FileStoreTable table = getTableDefault();
+ RowType writeType =
+ table.rowType().projectByPaths(Collections.singletonList("literal.dot"));
+
+ try (BatchTableWrite ignored =
+ table.newBatchWriteBuilder().newWrite().withWriteType(writeType)) {
+ // A dot in a top-level column name is not a nested sub-field write.
+ }
+ }
+
+ @Test
+ public void testNestedFieldOptionCannotBeDisabledAfterWrite() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite().withWriteType(table.rowType())) {
+ write.write(
+ GenericRow.of(
+ 1,
+ BinaryString.fromString("v"),
+ GenericRow.of(10, BinaryString.fromString("n")),
+ arrOf(1),
+ mapOf("k", 1)));
+ commit(builder, write.prepareCommit());
+ }
+
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.setOption(
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED
+ .key(),
+ "false"),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key());
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.removeOption(
+ CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED
+ .key()),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key());
+ }
+
+ @Test
+ public void testAddedLeafPreservesNullnessAcrossSiblingGroups() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ RowType full = table.rowType();
+ RowType cgA = full.projectByPaths(Collections.singletonList("nest.a"));
+ RowType cgB = full.projectByPaths(Collections.singletonList("nest.b"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+
+ try (BatchTableWrite write = builder.newWrite().withWriteType(cgB)) {
+ write.write(GenericRow.of(GenericRow.of(BinaryString.fromString("present-in-b"))));
+ commit(builder, write.prepareCommit());
+ }
+ try (BatchTableWrite write = builder.newWrite().withWriteType(cgA)) {
+ write.write(GenericRow.of((Object) null));
+ List messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ commit(builder, messages);
+ }
+
+ RowType projectedA = table.rowType().projectByPaths(Collections.singletonList("nest.a"));
+ ReadBuilder projectedAReadBuilder = table.newReadBuilder().withReadType(projectedA);
+ try (RecordReader reader =
+ projectedAReadBuilder
+ .newRead()
+ .createReader(projectedAReadBuilder.newScan().plan())) {
+ RecordReader.RecordIterator batch = reader.readBatch();
+ assertThat(batch).isNotNull();
+ InternalRow row = batch.next();
+ assertThat(row).isNotNull();
+ assertThat(row.isNullAt(0)).isFalse();
+ assertThat(row.getRow(0, 1).isNullAt(0)).isTrue();
+ assertThat(batch.next()).isNull();
+ batch.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn(new String[] {"nest", "added"}, DataTypes.INT(), null, null),
+ false);
+ table = getTableDefault();
+ RowType readType = table.rowType().projectByPaths(Collections.singletonList("nest.added"));
+ ReadBuilder readBuilder = table.newReadBuilder().withReadType(readType);
+
+ try (RecordReader reader =
+ readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ RecordReader.RecordIterator batch = reader.readBatch();
+ assertThat(batch).isNotNull();
+ InternalRow row = batch.next();
+ assertThat(row).isNotNull();
+ assertThat(row.isNullAt(0)).isFalse();
+ assertThat(row.getRow(0, 1).isNullAt(0)).isTrue();
+ assertThat(batch.next()).isNull();
+ batch.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+ }
+
+ @Test
+ public void testProjectedDeepAddedLeafPreservesEveryParentNullness() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "nest",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 0,
+ "sub",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 1, "existing", DataTypes.INT()))),
+ DataTypes.FIELD(2, "other", DataTypes.STRING())))
+ .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();
+
+ RowType otherType = table.rowType().projectByPaths(Collections.singletonList("nest.other"));
+ try (BatchTableWrite write = builder.newWrite().withWriteType(otherType)) {
+ write.write(GenericRow.of(GenericRow.of(BinaryString.fromString("other-0"))));
+ write.write(GenericRow.of(GenericRow.of(BinaryString.fromString("other-1"))));
+ commit(builder, write.prepareCommit());
+ }
+
+ RowType existingType =
+ table.rowType().projectByPaths(Collections.singletonList("nest.sub.existing"));
+ try (BatchTableWrite write = builder.newWrite().withWriteType(existingType)) {
+ write.write(GenericRow.of(GenericRow.of(GenericRow.of(10))));
+ write.write(GenericRow.of(GenericRow.of((Object) null)));
+ List messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ commit(builder, messages);
+ }
+
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn(
+ new String[] {"nest", "sub", "added"}, DataTypes.INT(), null, null),
+ false);
+ table = getTableDefault();
+ RowType readType =
+ table.rowType().projectByPaths(Collections.singletonList("nest.sub.added"));
+ ReadBuilder readBuilder = table.newReadBuilder().withReadType(readType);
+
+ try (RecordReader reader =
+ readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ RecordReader.RecordIterator batch = reader.readBatch();
+ assertThat(batch).isNotNull();
+
+ InternalRow first = batch.next();
+ assertThat(first).isNotNull();
+ InternalRow firstNest = first.getRow(0, 1);
+ assertThat(firstNest).isNotNull();
+ InternalRow firstSub = firstNest.getRow(0, 1);
+ assertThat(firstSub).isNotNull();
+ assertThat(firstSub.isNullAt(0)).isTrue();
+
+ InternalRow second = batch.next();
+ assertThat(second).isNotNull();
+ InternalRow secondNest = second.getRow(0, 1);
+ assertThat(secondNest).isNotNull();
+ assertThat(secondNest.isNullAt(0)).isTrue();
+
+ assertThat(batch.next()).isNull();
+ batch.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+ }
+
/**
* Core: write sub-field {@code nest.a} and {@code nest.b} into two separate files (plus the
* rest of the columns in a third), then read the full struct assembled from all three.
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
index 99c4508bd661..35cfc2db7aa5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.utils;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.schema.Schema;
@@ -42,10 +43,6 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
/** Test for {@link DataEvolutionUtils}. */
public class DataEvolutionUtilsTest {
@@ -66,24 +63,26 @@ public void testFileFieldIdsIgnoresSystemFields() {
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema,
+ schema.fields(),
dataFile(
"mixed.parquet",
1,
Arrays.asList(
SpecialFields.ROW_ID.name(),
"indexed",
- SpecialFields.SEQUENCE_NUMBER.name()))))
+ SpecialFields.SEQUENCE_NUMBER.name())),
+ false))
.containsExactly(1);
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema,
+ schema.fields(),
dataFile(
"system-only.parquet",
1,
Arrays.asList(
SpecialFields.ROW_ID.name(),
- SpecialFields.SEQUENCE_NUMBER.name()))))
+ SpecialFields.SEQUENCE_NUMBER.name())),
+ false))
.isEmpty();
}
@@ -103,31 +102,78 @@ public void testFileFieldIdsHandlesFullEmptyAndUnrelatedWrites() {
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema, dataFile("full.parquet", 1, null)))
+ schema.fields(), dataFile("full.parquet", 1, null), false))
.containsExactlyInAnyOrder(1, 2);
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema,
- dataFile("empty.parquet", 1, Collections.emptyList())))
+ schema.fields(),
+ dataFile("empty.parquet", 1, Collections.emptyList()),
+ false))
.isEmpty();
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema,
+ schema.fields(),
dataFile(
- "unrelated.parquet",
- 1,
- Collections.singletonList("other"))))
+ "unrelated.parquet", 1, Collections.singletonList("other")),
+ false))
.containsExactly(2);
assertThat(
DataEvolutionUtils.fileFieldIds(
- ignored -> schema,
+ schema.fields(),
dataFile(
- "unknown.parquet",
- 1,
- Collections.singletonList("unknown"))))
+ "unknown.parquet", 1, Collections.singletonList("unknown")),
+ false))
.isEmpty();
}
+ @Test
+ public void testNestedWriteColumnResolutionRequiresEnabledOption() {
+ DataField nested =
+ new DataField(
+ 1,
+ "nest",
+ DataTypes.ROW(
+ new DataField(2, "a", DataTypes.INT()),
+ new DataField(3, "b", DataTypes.INT())));
+ TableSchema disabled = tableSchema(1L, Collections.emptyMap(), nested);
+ DataFileMeta nestedFile =
+ dataFile("nested.parquet", 1L, Collections.singletonList("nest.a"));
+
+ assertThat(DataEvolutionUtils.fileFieldIds(disabled.fields(), nestedFile, false)).isEmpty();
+ assertThat(collectWrittenColumnIds(ignored -> disabled, nestedFile)).isEmpty();
+
+ Map enabledOptions = new HashMap<>();
+ enabledOptions.put(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+ TableSchema enabled = tableSchema(1L, enabledOptions, nested);
+ assertThat(DataEvolutionUtils.fileFieldIds(enabled.fields(), nestedFile, true))
+ .containsExactly(1);
+ assertThat(collectWrittenColumnIds(ignored -> enabled, nestedFile))
+ .hasValue(Collections.singletonList(1));
+ }
+
+ @Test
+ public void testDottedTopLevelWriteColumnWinsOverNestedPath() {
+ TableSchema schema =
+ tableSchema(
+ 1L,
+ Collections.emptyMap(),
+ new DataField(1, "nest.a", DataTypes.INT()),
+ new DataField(
+ 2, "nest", DataTypes.ROW(new DataField(3, "a", DataTypes.INT()))));
+ DataFileMeta file = dataFile("dotted.parquet", 1L, Collections.singletonList("nest.a"));
+
+ assertThat(DataEvolutionUtils.fileFieldIds(schema.fields(), file, false))
+ .containsExactly(1);
+ assertThat(DataEvolutionUtils.fileFields(schema.fields(), file, false))
+ .extracting(DataField::id)
+ .containsExactly(1);
+ assertThat(DataEvolutionUtils.fileFields(schema.fields(), file, true))
+ .extracting(DataField::id)
+ .containsExactly(1);
+ assertThat(collectWrittenColumnIds(ignored -> schema, file))
+ .hasValue(Collections.singletonList(1));
+ }
+
@Test
public void testCollectWrittenColumnIdsAcrossSchemas() {
Map schemas = new HashMap<>();
@@ -151,6 +197,30 @@ public void testCollectWrittenColumnIdsAcrossSchemas() {
.hasValue(Arrays.asList(1, 2, 3));
}
+ @Test
+ public void testCollectWrittenColumnIdsUsesNestedOptionOfEachSchema() {
+ Map schemas = new HashMap<>();
+ schemas.put(
+ 0L,
+ tableSchema(
+ 0L, Collections.emptyMap(), new DataField(1, "nest.a", DataTypes.INT())));
+ Map nestedOptions = new HashMap<>();
+ nestedOptions.put(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+ schemas.put(
+ 1L,
+ tableSchema(
+ 1L,
+ nestedOptions,
+ new DataField(
+ 2, "nest", DataTypes.ROW(new DataField(3, "b", DataTypes.INT())))));
+
+ DataFileMeta dottedTopLevelFile = dataFile(0L, Collections.singletonList("nest.a"));
+ DataFileMeta nestedFile = dataFile(1L, Collections.singletonList("nest.b"));
+
+ assertThat(collectWrittenColumnIds(schemas::get, dottedTopLevelFile, nestedFile))
+ .hasValue(Arrays.asList(1, 2));
+ }
+
@Test
public void testCollectWrittenColumnIdsFallsBackWhenResolutionFails() {
DataFileMeta unknownSchemaFile = dataFile(99L, Collections.singletonList("a"));
@@ -203,30 +273,31 @@ public void testCollectWrittenColumnIdsIgnoresSystemFields() {
@Test
public void testCollectWrittenColumnIdsCachesSchemaAcrossProjections() {
TableSchema schema =
- spy(
- tableSchema(
- 1L,
- new DataField(1, "a", DataTypes.INT()),
- new DataField(2, "b", DataTypes.STRING())));
+ tableSchema(
+ 1L,
+ new DataField(1, "a", DataTypes.INT()),
+ new DataField(2, "b", DataTypes.STRING()));
DataFileMeta first = dataFile(1L, Collections.singletonList("a"));
DataFileMeta second = dataFile(1L, Collections.singletonList("b"));
DataFileMeta repeated = dataFile(1L, Collections.singletonList("a"));
- AtomicInteger schemaLoads = new AtomicInteger();
+ AtomicInteger schemaFieldLoads = new AtomicInteger();
+ AtomicInteger nestedOptionLoads = new AtomicInteger();
Optional> result =
- collectWrittenColumnIds(
+ DataEvolutionUtils.collectWrittenColumnIds(
+ Collections.singletonList(dataSplit(first, second, repeated)),
ignored -> {
- schemaLoads.incrementAndGet();
- return schema;
+ schemaFieldLoads.incrementAndGet();
+ return schema.fields();
},
- first,
- second,
- repeated);
+ ignored -> {
+ nestedOptionLoads.incrementAndGet();
+ return false;
+ });
assertThat(result.get()).containsExactly(1, 2);
- assertThat(schemaLoads).hasValue(1);
- verify(schema).fields();
- verify(repeated).writeCols();
+ assertThat(schemaFieldLoads).hasValue(1);
+ assertThat(nestedOptionLoads).hasValue(1);
}
@Test
@@ -258,16 +329,40 @@ public void testFileFieldsFollowWriteColsOrderAndIgnoreSystemFields() {
assertThat(
DataEvolutionUtils.fileFields(
- ignored -> schema,
+ schema.fields(),
dataFile(
"reordered.parquet",
1,
Arrays.asList(
- "other", SpecialFields.ROW_ID.name(), "indexed"))))
+ "other", SpecialFields.ROW_ID.name(), "indexed")),
+ false))
.extracting(DataField::id)
.containsExactly(2, 1);
}
+ @Test
+ public void testFileFieldsProjectsNestedPathsOnlyWhenEnabled() {
+ DataField nested =
+ new DataField(
+ 1,
+ "nest",
+ DataTypes.ROW(
+ new DataField(2, "a", DataTypes.INT()),
+ new DataField(3, "b", DataTypes.INT())));
+ DataFileMeta file = dataFile("nested.parquet", 1, Arrays.asList("nest.b", "nest.a"));
+
+ TableSchema disabled = tableSchema(1L, Collections.emptyMap(), nested);
+ assertThat(DataEvolutionUtils.fileFields(disabled.fields(), file, false)).isEmpty();
+
+ Map enabledOptions = new HashMap<>();
+ enabledOptions.put(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+ TableSchema enabled = tableSchema(1L, enabledOptions, nested);
+ List fields = DataEvolutionUtils.fileFields(enabled.fields(), file, true);
+ assertThat(fields).extracting(DataField::name).containsExactly("nest");
+ assertThat(((org.apache.paimon.types.RowType) fields.get(0).type()).getFieldNames())
+ .containsExactly("b", "a");
+ }
+
@Test
public void testFieldMaxSequenceNumberFallsBackForMissingOrMalformedArray() {
DataFileMeta legacy = dataFile("legacy.parquet", 10, null);
@@ -357,10 +452,21 @@ private static DataFileMeta dataFile(
}
private static DataFileMeta dataFile(long schemaId, java.util.List writeCols) {
- DataFileMeta file = mock(DataFileMeta.class);
- when(file.schemaId()).thenReturn(schemaId);
- when(file.writeCols()).thenReturn(writeCols);
- return file;
+ return DataFileMeta.forAppend(
+ "schema-" + schemaId + ".parquet",
+ 1L,
+ 1L,
+ SimpleStats.EMPTY_STATS,
+ 1L,
+ 1L,
+ schemaId,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ 0L,
+ writeCols);
}
private static DataSplit dataSplit(DataFileMeta... files) {
@@ -375,18 +481,33 @@ private static DataSplit dataSplit(DataFileMeta... files) {
private static Optional> collectWrittenColumnIds(
Function schemaLoader, DataFileMeta... files) {
+ Map schemaCache = new HashMap<>();
+ Function cachedSchemaLoader =
+ schemaId -> schemaCache.computeIfAbsent(schemaId, schemaLoader);
return DataEvolutionUtils.collectWrittenColumnIds(
- Collections.singletonList(dataSplit(files)), schemaLoader);
+ Collections.singletonList(dataSplit(files)),
+ schemaId -> {
+ TableSchema schema = cachedSchemaLoader.apply(schemaId);
+ return schema == null ? null : schema.fields();
+ },
+ schemaId ->
+ new CoreOptions(cachedSchemaLoader.apply(schemaId).options())
+ .dataEvolutionNestedFieldEnabled());
}
private static TableSchema tableSchema(long id, DataField... fields) {
+ return tableSchema(id, Collections.emptyMap(), fields);
+ }
+
+ private static TableSchema tableSchema(
+ long id, Map options, DataField... fields) {
return TableSchema.create(
id,
new Schema(
Arrays.asList(fields),
Collections.emptyList(),
Collections.emptyList(),
- Collections.emptyMap(),
+ options,
null));
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
index 88f8d94366a3..70c1710151db 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
@@ -301,6 +301,98 @@ public void testNormalRowWithSelectedKeysMetadataComment() throws Exception {
.containsExactly(rowField);
}
+ @Test
+ public void testAddedNestedLeafReadsSiblingAsNullabilityAnchor() throws Exception {
+ DataField dataField =
+ DataTypes.FIELD(
+ 1, "payload", DataTypes.ROW(DataTypes.FIELD(2, "x", DataTypes.INT())));
+ DataField expectedField =
+ DataTypes.FIELD(
+ 1, "payload", DataTypes.ROW(DataTypes.FIELD(3, "y", DataTypes.INT())));
+
+ List readFields =
+ invokeNestedDataFields(
+ Collections.singletonList(dataField),
+ Collections.singletonList(expectedField),
+ Collections.emptySet());
+
+ Assertions.assertThat(readFields).containsExactly(dataField);
+ IndexCastMapping mapping =
+ SchemaEvolutionUtil.createIndexCastMapping(
+ Collections.singletonList(expectedField), readFields);
+ InternalRow evolved =
+ mapping.getCastMapping()[0].getFieldOrNull(GenericRow.of(GenericRow.of(10)));
+ Assertions.assertThat(evolved).isNotNull();
+ Assertions.assertThat(evolved.isNullAt(0)).isTrue();
+ Object nullParent =
+ mapping.getCastMapping()[0].getFieldOrNull(GenericRow.of((Object) null));
+ Assertions.assertThat(nullParent).isNull();
+ }
+
+ @Test
+ public void testDeepAddedNestedLeafReadsSiblingAsNullabilityAnchor() throws Exception {
+ DataField dataField =
+ DataTypes.FIELD(
+ 1,
+ "payload",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 2,
+ "sub",
+ DataTypes.ROW(
+ DataTypes.FIELD(3, "existing", DataTypes.INT())))));
+ DataField expectedField =
+ DataTypes.FIELD(
+ 1,
+ "payload",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 2,
+ "sub",
+ DataTypes.ROW(
+ DataTypes.FIELD(4, "added", DataTypes.INT())))));
+
+ List readFields =
+ invokeNestedDataFields(
+ Collections.singletonList(dataField),
+ Collections.singletonList(expectedField),
+ Collections.emptySet());
+
+ Assertions.assertThat(readFields).containsExactly(dataField);
+ IndexCastMapping mapping =
+ SchemaEvolutionUtil.createIndexCastMapping(
+ Collections.singletonList(expectedField), readFields);
+ InternalRow evolved =
+ mapping.getCastMapping()[0].getFieldOrNull(
+ GenericRow.of(GenericRow.of(GenericRow.of(10))));
+ Assertions.assertThat(evolved).isNotNull();
+ InternalRow evolvedSub = evolved.getRow(0, 1);
+ Assertions.assertThat(evolvedSub).isNotNull();
+ Assertions.assertThat(evolvedSub.isNullAt(0)).isTrue();
+ InternalRow nullSub =
+ mapping.getCastMapping()[0].getFieldOrNull(
+ GenericRow.of(GenericRow.of((Object) null)));
+ Assertions.assertThat(nullSub).isNotNull();
+ Assertions.assertThat(nullSub.isNullAt(0)).isTrue();
+ }
+
+ @Test
+ public void testAddedNestedLeafDoesNotReadSiblingWhenNestedFieldDisabled() throws Exception {
+ DataField dataField =
+ DataTypes.FIELD(
+ 1, "payload", DataTypes.ROW(DataTypes.FIELD(2, "x", DataTypes.INT())));
+ DataField expectedField =
+ DataTypes.FIELD(
+ 1, "payload", DataTypes.ROW(DataTypes.FIELD(3, "y", DataTypes.INT())));
+
+ Assertions.assertThat(
+ invokeDataFields(
+ Collections.singletonList(dataField),
+ Collections.singletonList(expectedField),
+ Collections.emptySet()))
+ .isEmpty();
+ }
+
@Test
public void testRejectSelectedKeysDataFieldWithNonMapType() {
DataField dataField = DataTypes.FIELD(2, "attrs", DataTypes.BIGINT());
@@ -338,6 +430,23 @@ private static List invokeDataFields(
method.invoke(builder, allDataFields, expectedFields, selectedKeysFieldIds);
}
+ @SuppressWarnings("unchecked")
+ private static List invokeNestedDataFields(
+ List allDataFields,
+ List expectedFields,
+ Set selectedKeysFieldIds)
+ throws Exception {
+ FormatReaderMapping.Builder builder =
+ new FormatReaderMapping.Builder(
+ null, Collections.emptyList(), null, null, null, null);
+ Method method =
+ FormatReaderMapping.Builder.class.getDeclaredMethod(
+ "readDataFields", List.class, List.class, Set.class, boolean.class);
+ method.setAccessible(true);
+ return (List)
+ method.invoke(builder, allDataFields, expectedFields, selectedKeysFieldIds, true);
+ }
+
@SuppressWarnings("unchecked")
private static Set invokeSelectedKeysFieldIds(
TableSchema tableSchema, List expectedFields) throws Exception {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
index 1e771c3ec377..4407ca3a8581 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
@@ -105,6 +105,7 @@ public class DataEvolutionMergeIntoAction extends TableActionBase {
private static final Logger LOG = LoggerFactory.getLogger(DataEvolutionMergeIntoAction.class);
private final CoreOptions coreOptions;
+ private final boolean nestedFieldEnabled;
// field names of target table
private final List targetFieldNames;
@@ -165,6 +166,7 @@ public DataEvolutionMergeIntoAction(
latestSnapshotId.toString()));
this.coreOptions = ((FileStoreTable) table).coreOptions();
+ this.nestedFieldEnabled = coreOptions.dataEvolutionNestedFieldEnabled();
if (!coreOptions.dataEvolutionEnabled()) {
throw new UnsupportedOperationException(
@@ -315,7 +317,7 @@ public Tuple2, RowType> buildSource() {
checkSchema(source);
RowType sourceType;
- if (updateAll) {
+ if (updateAll || !nestedFieldEnabled) {
List columnNames = source.getResolvedSchema().getColumnNames();
sourceType = SpecialFields.rowTypeWithRowId(table.rowType()).project(columnNames);
writePaths =
@@ -345,6 +347,10 @@ public Tuple2, RowType> buildSource() {
* Also sets {@link #writePaths}.
*/
private List buildExplicitProject() {
+ if (!nestedFieldEnabled) {
+ return buildTopLevelExplicitProject();
+ }
+
checkNoDuplicateSetTargets();
Map changes = parseCommaSeparatedKeyValues(matchedUpdateSet);
@@ -374,7 +380,7 @@ private List buildExplicitProject() {
wholeCols.put(topCol, entry.getValue());
} else {
// nested sub-field update
- if (!coreOptions.dataEvolutionNestedFieldEnabled()) {
+ if (!nestedFieldEnabled) {
throw new UnsupportedOperationException(
"Updating a nested sub-field ('"
+ entry.getKey()
@@ -461,12 +467,63 @@ private List buildExplicitProject() {
return project;
}
+ /** Build the legacy top-level projection when nested-field data evolution is disabled. */
+ private List buildTopLevelExplicitProject() {
+ Map changes = parseCommaSeparatedKeyValues(matchedUpdateSet);
+ List project = new ArrayList<>();
+ writePaths = new ArrayList<>();
+ for (Map.Entry entry : changes.entrySet()) {
+ String fieldName = topLevelTarget(entry.getKey());
+ if (!targetFieldNames.contains(fieldName)) {
+ throw new RuntimeException(
+ String.format(
+ "Invalid column reference '%s' of table '%s' at matched-upsert action.",
+ entry.getKey(), identifier.getFullName()));
+ }
+ writePaths.add(fieldName);
+ project.add(String.format("%s AS `%s`", entry.getValue(), fieldName));
+ }
+ return project;
+ }
+
+ /** Resolve a top-level SET target without interpreting dots inside an actual column name. */
+ private String topLevelTarget(String target) {
+ if (targetFieldNames.contains(target)) {
+ return target;
+ }
+
+ String qualifier = targetTableName() + ".";
+ if (target.startsWith(qualifier)) {
+ String unqualified = target.substring(qualifier.length());
+ if (targetFieldNames.contains(unqualified)) {
+ return unqualified;
+ }
+ }
+
+ List path = new ArrayList<>(Arrays.asList(target.split("\\.")));
+ if (path.size() > 1 && path.get(0).equals(targetTableName())) {
+ path = new ArrayList<>(path.subList(1, path.size()));
+ }
+ if (path.size() > 1 && resolvesAgainstTarget(path)) {
+ throw new UnsupportedOperationException(
+ "Updating a nested sub-field ('"
+ + target
+ + "') requires '"
+ + CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key()
+ + "=true'.");
+ }
+
+ // Keep the historical qualifier handling for ordinary top-level columns.
+ return path.get(path.size() - 1);
+ }
+
/** The first-seen order of top-level columns present in the (dotted) write paths. */
private List explicitTopColumnOrder(List paths) {
List order = new ArrayList<>();
for (String path : paths) {
int dot = path.indexOf('.');
- String topCol = dot < 0 ? path : path.substring(0, dot);
+ String topCol =
+ dot < 0 || targetFieldNames.contains(path) ? path : path.substring(0, dot);
if (!order.contains(topCol)) {
order.add(topCol);
}
@@ -486,7 +543,20 @@ private List explicitTopColumnOrder(List paths) {
* silently updating one of them.
*/
private List parseTargetPath(String target) {
+ // Prefer exact top-level names before interpreting dots as path separators. This mirrors
+ // RowType#projectByPaths and keeps columns whose names contain '.' addressable.
+ if (targetFieldNames.contains(target)) {
+ return Collections.singletonList(target);
+ }
List segs = new ArrayList<>(Arrays.asList(target.split("\\.")));
+ String qualifier = targetTableName() + ".";
+ if (target.startsWith(qualifier)) {
+ String unqualifiedTarget = target.substring(qualifier.length());
+ if (targetFieldNames.contains(unqualifiedTarget) && !resolvesAgainstTarget(segs)) {
+ return Collections.singletonList(unqualifiedTarget);
+ }
+ }
+
if (segs.size() > 1 && segs.get(0).equals(targetTableName())) {
List unqualified = new ArrayList<>(segs.subList(1, segs.size()));
boolean asQualified = resolvesAgainstTarget(unqualified);
@@ -796,7 +866,9 @@ private void checkSchema(Table source) {
// source must fully cover the target struct, so a narrower source is rejected
// instead of being written as an incomplete whole-struct file.
boolean structCompatible = false;
- if (paimonType instanceof RowType && targetField.type() instanceof RowType) {
+ if (nestedFieldEnabled
+ && paimonType instanceof RowType
+ && targetField.type() instanceof RowType) {
RowType sourceStruct = (RowType) paimonType;
RowType targetStruct = (RowType) targetField.type();
structCompatible =
@@ -878,7 +950,7 @@ private boolean isFullyCompatibleStruct(RowType source, RowType target) {
* Whether the given top-level column is written through dotted sub-field paths (e.g. nest.a).
*/
private boolean isSubFieldWrite(String topColumn) {
- if (writePaths == null) {
+ if (!nestedFieldEnabled || writePaths == null) {
return false;
}
String prefix = topColumn + ".";
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
index 29dbbde36f8a..d7b2a2eabb24 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
@@ -70,6 +70,7 @@ public class DataEvolutionPartialWriteOperator
private final FileStoreTable table;
private final Long baseSnapshotId;
+ private final boolean nestedFieldEnabled;
// dataType
private final RowType dataType;
@@ -106,9 +107,11 @@ public DataEvolutionPartialWriteOperator(
Long baseSnapshotId) {
this.table = table.copy(dataEvolutionWriteOptions());
this.baseSnapshotId = baseSnapshotId;
- // writePaths may carry nested dotted paths (e.g. "nest.a") for sub-field-level data
- // evolution; projectByPaths handles both plain top-level names and nested paths.
- this.writeType = table.rowType().projectByPaths(writePaths);
+ this.nestedFieldEnabled = this.table.coreOptions().dataEvolutionNestedFieldEnabled();
+ this.writeType =
+ nestedFieldEnabled
+ ? table.rowType().projectByPaths(writePaths)
+ : table.rowType().project(writePaths);
// sourceType is already pruned to the written columns (with partial nested structs) and
// carries the table's field ids, so it is used directly as the read/data type.
this.dataType = sourceType;
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 02e1fd3c2f2c..a5722ff6ceec 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
@@ -137,6 +137,57 @@ public void testUpdateSubFieldDisabledThrows() throws Exception {
.hasMessageContaining(DATA_EVOLUTION_NESTED_FIELD_ENABLED.key());
}
+ @Test
+ public void testDisabledOptionTreatsDotAsPartOfTopLevelColumnName() throws Exception {
+ testDottedTopLevelColumn(false);
+ }
+
+ @Test
+ public void testEnabledOptionTreatsDotAsPartOfTopLevelColumnName() throws Exception {
+ testDottedTopLevelColumn(true);
+ }
+
+ private void testDottedTopLevelColumn(boolean nestedFieldEnabled) throws Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "T",
+ Arrays.asList("id INT", "`literal.dot` INT"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ if (nestedFieldEnabled) {
+ put(DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+ }
+ }
+ }));
+ insertInto("T", "(1, 10)");
+
+ sEnv.executeSql(
+ buildDdl(
+ "S",
+ Arrays.asList("id INT", "new_value INT"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("S", "(1, 100)");
+
+ builder(warehouse, database, "T")
+ .withMergeCondition("T.id=S.id")
+ .withMatchedUpdateSet("T.literal.dot=S.new_value")
+ .withSourceTable("S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, `literal.dot` FROM T",
+ Collections.singletonList(changelogRow("+I", 1, 100)));
+ assertThat(deltaWriteCols("T")).contains(Collections.singletonList("literal.dot"));
+ }
+
@Test
public void testUpdateWholeStructStillWorks() throws Exception {
prepareNestedTarget(true);
diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 0678bfa1a8b4..88deb9817c99 100644
--- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -938,7 +938,9 @@ case class MergeIntoPaimonDataEvolutionTable(
case None => Seq(attr.name)
}
}
- val writeType = table.rowType().projectByPaths(writePaths.asJava)
+ val writeType =
+ if (nestedFieldEnabled) table.rowType().projectByPaths(writePaths.asJava)
+ else table.rowType().project(writePaths.asJava)
val writePartialFields = updateColumnsSorted.nonEmpty
val shouldPersistOutput = writePartialFields && hasMatchedDeleteActions
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionCompactMergeConflictRewriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionCompactMergeConflictRewriter.scala
index 9f5401d3278a..be2d9f8c1f0f 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionCompactMergeConflictRewriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionCompactMergeConflictRewriter.scala
@@ -54,6 +54,9 @@ class DataEvolutionCompactMergeConflictRewriter(
import DataEvolutionCompactMergeConflictRewriter._
+ private val partialColumns = new DataEvolutionPartialColumns(table)
+ private val nestedFieldEnabled = table.coreOptions().dataEvolutionNestedFieldEnabled()
+
def rewrite(
sparkSession: SparkSession,
baseSnapshot: Snapshot,
@@ -163,11 +166,17 @@ class DataEvolutionCompactMergeConflictRewriter(
* are grouped by it and it becomes the physical layout of the rebased file.
*/
private def updatedWritePaths(writePaths: Set[String]): Seq[String] = {
- val fieldNames = table.rowType().getFieldNames.asScala.toSet
+ if (!nestedFieldEnabled) {
+ return table
+ .rowType()
+ .getFieldNames
+ .asScala
+ .filter(writePaths.contains)
+ .toSeq
+ }
table.rowType().getFields.asScala.toSeq.flatMap {
field =>
- val forField = writePaths.filter(
- path => DataEvolutionPartialColumns.topLevelOf(path, fieldNames) == field.name)
+ val forField = writePaths.filter(path => partialColumns.topLevelOf(path) == field.name)
if (forField.isEmpty) {
Seq.empty[String]
} else if (forField.contains(field.name)) {
@@ -267,9 +276,8 @@ class DataEvolutionCompactMergeConflictRewriter(
// updatedFields are write paths and may address a single leaf of a struct (e.g. "nest.a"); the
// scan is in terms of top-level columns and the projection prunes each partially written
// struct, so the rebased file carries exactly the leaves the staged MERGE files carried.
- val fieldNames = table.rowType().getFieldNames.asScala.toSet
- val topColumns = DataEvolutionPartialColumns.topLevelColumns(updatedFields, fieldNames)
- val projections = DataEvolutionPartialColumns.projections(table, updatedFields)
+ val topColumns = partialColumns.topLevelColumns(updatedFields)
+ val projections = partialColumns.projections(updatedFields)
val readOutput = topColumns.map(attribute) :+ rowIdAttribute
val relation = createNewScanPlan(relevantSplits, targetRelation)
val readPlan =
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
index 2eb889c678ed..f11d661432c6 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
@@ -46,6 +46,9 @@ case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits: Se
paimonTable.copy(writeOptions.asJava)
}
+ private val dataEvolutionNestedFieldEnabled =
+ table.coreOptions().dataEvolutionNestedFieldEnabled()
+
// Whole top-level column write (kept for callers that only update full columns).
def writePartialFields(
data: DataFrame,
@@ -53,8 +56,13 @@ case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits: Se
rawBlobPlaceholderMarkerColumns: Map[String, String] = Map.empty): Seq[CommitMessage] = {
writePartialFields(
data,
- table.rowType().projectByPaths(columnNames.asJava),
- rawBlobPlaceholderMarkerColumns)
+ if (dataEvolutionNestedFieldEnabled) {
+ table.rowType().projectByPaths(columnNames.asJava)
+ } else {
+ table.rowType().project(columnNames.asJava)
+ },
+ rawBlobPlaceholderMarkerColumns
+ )
}
// Sub-field-aware write: writeType is already pruned to the written top-level columns and
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPartialColumns.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPartialColumns.scala
index e3def62ac0e6..c2c56abc9225 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPartialColumns.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPartialColumns.scala
@@ -33,32 +33,47 @@ import scala.collection.JavaConverters._
* columns may be dotted paths addressing a single leaf of a struct (e.g. `Seq("value", "nest.a")`)
* rather than plain top-level names.
*/
-private[spark] object DataEvolutionPartialColumns {
+private[spark] class DataEvolutionPartialColumns(table: FileStoreTable) {
+
+ private val dataEvolutionNestedFieldEnabled =
+ table.coreOptions().dataEvolutionNestedFieldEnabled()
+
+ private lazy val fieldNames = table.rowType().getFieldNames.asScala.toSet
/**
* The top-level column a write path addresses. A path that names a field exactly is that field
* even when its own name contains a dot, mirroring `RowType#projectByPaths`.
*/
- def topLevelOf(path: String, fieldNames: Set[String]): String = {
+ def topLevelOf(path: String): String = {
+ if (!dataEvolutionNestedFieldEnabled) {
+ return path
+ }
val dot = path.indexOf('.')
if (dot < 0 || fieldNames.contains(path)) path else path.substring(0, dot)
}
/** Distinct top-level column names addressed by `paths`, in first-seen order. */
- def topLevelColumns(paths: Seq[String], fieldNames: Set[String]): Seq[String] =
- paths.map(path => topLevelOf(path, fieldNames)).distinct
+ def topLevelColumns(paths: Seq[String]): Seq[String] =
+ paths.map(topLevelOf).distinct
/** Whether any path addresses a sub-field rather than a whole top-level column. */
- def hasNestedPaths(paths: Seq[String], fieldNames: Set[String]): Boolean =
- paths.exists(path => topLevelOf(path, fieldNames) != path)
+ def hasNestedPaths(paths: Seq[String]): Boolean =
+ paths.exists(path => topLevelOf(path) != path)
/**
* The Spark type of the row a file with these write paths physically holds, i.e. the Spark view
* of `table.rowType().projectByPaths(paths)`. Its fields are in write-path order, which is the
* order [[DataEvolutionPaimonWriter.writePartialFields]] expects the data frame to be in.
*/
- def writeStructType(table: FileStoreTable, paths: Seq[String]): StructType =
- SparkTypeUtils.fromPaimonRowType(table.rowType().projectByPaths(paths.asJava))
+ def writeStructType(paths: Seq[String]): StructType = {
+ val writeType =
+ if (dataEvolutionNestedFieldEnabled) {
+ table.rowType().projectByPaths(paths.asJava)
+ } else {
+ table.rowType().project(paths.asJava)
+ }
+ SparkTypeUtils.fromPaimonRowType(writeType)
+ }
/**
* One projection per top-level column, in write-path order and aliased to the column's name. A
@@ -66,10 +81,9 @@ private[spark] object DataEvolutionPartialColumns {
* leaves, so re-writing the file leaves its untouched siblings alone instead of overwriting them
* with nulls.
*/
- def projections(table: FileStoreTable, paths: Seq[String]): Seq[Column] = {
- val fieldNames = table.rowType().getFieldNames.asScala.toSet
- val topCols = topLevelColumns(paths, fieldNames)
- val writeType = writeStructType(table, paths)
+ def projections(paths: Seq[String]): Seq[Column] = {
+ val topCols = topLevelColumns(paths)
+ val writeType = writeStructType(paths)
topCols.zip(writeType.fields).map {
case (name, field) =>
val column = quotedColumn(name)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
index 881e61758d04..3ac2b53a8434 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
@@ -50,6 +50,8 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
import DataEvolutionRowIdConflictRewriter._
+ private val partialColumns = new DataEvolutionPartialColumns(table)
+
def rewrite(
sparkSession: SparkSession,
latestSnapshot: Snapshot,
@@ -172,9 +174,8 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
// columnNames are write paths and may address a single leaf of a struct (e.g. "nest.a"); the
// scan is always in terms of the top-level columns, and a projection then prunes each
// partially written struct down to the leaves the file actually holds.
- val fieldNames = table.rowType().getFieldNames.asScala.toSet
- val topColumns = DataEvolutionPartialColumns.topLevelColumns(columnNames, fieldNames)
- val projections = DataEvolutionPartialColumns.projections(table, columnNames)
+ val topColumns = partialColumns.topLevelColumns(columnNames)
+ val projections = partialColumns.projections(columnNames)
val readOutput = topColumns.map(attribute) :+ rowIdAttribute
def readRows(splits: Seq[DataSplit]) = {
val relation = createNewScanPlan(splits, targetRelation)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 11a508399b80..7bbc1487b6a7 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -937,7 +937,9 @@ case class MergeIntoPaimonDataEvolutionTable(
case None => Seq(attr.name)
}
}
- val writeType = table.rowType().projectByPaths(writePaths.asJava)
+ val writeType =
+ if (nestedFieldEnabled) table.rowType().projectByPaths(writePaths.asJava)
+ else table.rowType().project(writePaths.asJava)
val writePartialFields = updateColumnsSorted.nonEmpty
val shouldPersistOutput = writePartialFields && hasMatchedDeleteActions
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
index 7fc7548c5eab..62d787bdea53 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
@@ -201,7 +201,14 @@ class PaimonMicroBatchStream(
startOffset.json(),
endOffset.json(),
admittedSplits.length,
- () => DataEvolutionUtils.collectWrittenColumnIds(admittedSplitSnapshot, schemaLoader)
+ () =>
+ DataEvolutionUtils.collectWrittenColumnIds(
+ admittedSplitSnapshot,
+ schemaId => schemaLoader.apply(schemaId).fields(),
+ schemaId =>
+ new CoreOptions(schemaLoader.apply(schemaId).options())
+ .dataEvolutionNestedFieldEnabled()
+ )
)
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/NestedSubfieldMergeIntoTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/NestedSubfieldMergeIntoTest.scala
index 369013d75b6f..6524440678b7 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/NestedSubfieldMergeIntoTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/NestedSubfieldMergeIntoTest.scala
@@ -78,7 +78,7 @@ class NestedSubfieldMergeIntoTest extends PaimonSparkTestBase {
}
}
- // Guards the read path: DataEvolutionSplitRead calls leafPaths() on the planned read type,
+ // Guards the read path: DataEvolutionSplitRead calls collectLeafPaths() on the planned read type,
// which now requires recursive field order to match the schema. A reversed nested projection
// must still read back correctly rather than tripping that check.
test("Sub-field data evolution: reversed nested projection reads correctly") {
@@ -224,6 +224,28 @@ class NestedSubfieldMergeIntoTest extends PaimonSparkTestBase {
}
}
+ test("Nested-field evolution disabled: a dot remains part of a top-level column name") {
+ withTable("s", "t") {
+ sql(s"""
+ |CREATE TABLE t (id INT, `literal.dot` INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 10)")
+
+ Seq((1, 100)).toDF("id", "new_value").createOrReplaceTempView("s")
+ sql(s"""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.`literal.dot` = s.new_value
+ |""".stripMargin).collect()
+
+ checkAnswer(sql("SELECT id, `literal.dot` FROM t"), Seq(Row(1, 100)))
+ assert(latestDeltaWriteCols("t").exists(_ == Seq("literal.dot")))
+ }
+ }
+
test(
"Sub-field data evolution: sub-fields touched by separate WHEN MATCHED clauses in reverse " +
"schema order are not swapped (regression for #8334 review)") {