This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e06c538944 [Python] Preserve data evolution column updates across
renames (#10661)
e06c538944 is described below
commit e06c538944f15bc332a15fbce48e13659f5a0208
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 20:17:49 2026 +0800
[Python] Preserve data evolution column updates across renames (#10661)
---
.../paimon/table/DataEvolutionFileIndexTest.java | 233 +++++++++++++++++++++
paimon-python/pypaimon/read/split_read.py | 14 +-
.../tests/variant_row_sidecar_index_test.py | 162 ++++++++++++++
3 files changed, 404 insertions(+), 5 deletions(-)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
index 1e7d086679..c060daf7b7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
@@ -27,6 +27,8 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.VariantMetadataUtils;
import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
@@ -38,6 +40,9 @@ import
org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory;
import org.apache.paimon.fs.Path;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.BundleRecords;
import org.apache.paimon.io.CompactIncrement;
@@ -118,6 +123,234 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
/** Inside the min/max range of f2 ("b000".."b099"), but not written. */
private static final String MISSING_F2 = "b050x";
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "file-bitmap-single",
+ "file-bitmap-merged",
+ "bitmap-single",
+ "bitmap-merged",
+ "btree-single",
+ "btree-merged"
+ })
+ public void testVariantIndexQueryReadsRowSidecar(String mode) throws
Exception {
+ boolean merged = mode.endsWith("merged");
+ boolean fileIndex = mode.startsWith("file-");
+ String indexType = mode.substring(0, mode.indexOf('-'));
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.VARIANT())
+ .column(
+ "nested",
+ DataTypes.ROW(
+ DataTypes.FIELD(3, "label",
DataTypes.STRING()),
+ DataTypes.FIELD(4, "detail",
DataTypes.VARIANT())))
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+
.option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true")
+ .option(
+ CoreOptions.GLOBAL_INDEX_ENABLED.key(),
+ Boolean.toString(!fileIndex))
+
.option(CoreOptions.GLOBAL_INDEX_QUERY_IN_READER_ENABLED.key(), "false");
+ Identifier identifier = identifier("variant_sidecar_" + mode);
+ catalog.createTable(identifier, schemaBuilder.build(), false);
+ FileStoreTable table = getTable(identifier);
+
+ // A preceding group makes the selected global row ids differ from
physical positions.
+ write(table, variantSidecarRow(-2), variantSidecarRow(-1));
+ RowType firstWriteType =
+ merged ? table.rowType().project("id", "payload") :
table.rowType();
+ BatchWriteBuilder writeBuilder =
+ table.copy(fileIndex ? bitmapOptions("id") :
Collections.emptyMap())
+ .newBatchWriteBuilder();
+ try (BatchTableWrite write =
writeBuilder.newWrite().withWriteType(firstWriteType);
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ GenericRow row = variantSidecarRow(i);
+ write.write(merged ? GenericRow.of(i, row.getVariant(1)) :
row);
+ }
+ commit.commit(write.prepareCommit());
+ }
+ if (merged) {
+ table = getTable(identifier);
+ writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ writeBuilder
+ .newWrite()
+
.withWriteType(table.rowType().project("nested"));
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ write.write(GenericRow.of(variantSidecarRow(i).getRow(2,
2)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ setSingleFileFirstRowId(messages, 2L);
+ commit.commit(messages);
+ }
+ table = getTable(identifier);
+ writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ writeBuilder
+ .newWrite()
+
.withWriteType(table.rowType().project("payload"));
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ write.write(GenericRow.of(variantSidecarRow(i +
1000).getVariant(1)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ setSingleFileFirstRowId(messages, 2L);
+ commit.commit(messages);
+ }
+ }
+ table = getTable(identifier);
+ if (!fileIndex) {
+ ScanResult<DataSplit> indexScan =
+ new SortedGlobalIndexScanner(table, indexType)
+ .withIndexFields(Collections.singletonList("id"))
+ .scan()
+ .get();
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataSplit split : indexScan.entries()) {
+ messages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, indexType, "id", split,
indexScan.scanSnapshotId()));
+ }
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+ catalog.alterTable(
+ identifier, SchemaChange.renameColumn("payload",
"renamed_payload"), false);
+ FileStoreTable latest = getTable(identifier);
+ Predicate filter =
+ new PredicateBuilder(latest.rowType()).in(0, Arrays.asList(50,
51, 52, 53, 54));
+ RowType fullType =
+ rowTypeWithRowId(latest.rowType())
+ .project("nested", "renamed_payload", "id",
SpecialFields.ROW_ID.name());
+ ReadBuilder fullRead =
latest.newReadBuilder().withReadType(fullType).withFilter(filter);
+ TableScan.Plan plan = fullRead.newScan().plan();
+ assertThat(plan.splits()).hasSize(1);
+ Split selected = plan.splits().get(0);
+ if (!fileIndex) {
+ assertThat(selected).isInstanceOf(IndexedSplit.class);
+ assertThat(((IndexedSplit)
selected).rowRanges()).containsExactly(new Range(52L, 56L));
+ }
+ DataSplit split =
+ selected instanceof IndexedSplit
+ ? ((IndexedSplit) selected).dataSplit()
+ : (DataSplit) selected;
+ assertThat(split.dataFiles()).hasSize(merged ? 3 : 1);
+ DataFilePathFactory paths =
+ latest.store()
+ .pathFactory()
+ .createDataFilePathFactory(split.partition(),
split.bucket());
+ for (DataFileMeta file : split.dataFiles()) {
+ assertThat(file.fileName()).endsWith(".parquet");
+ List<String> sidecars =
+ file.extraFiles().stream()
+ .filter(name -> name.endsWith(".row"))
+ .collect(Collectors.toList());
+ assertThat(sidecars).hasSize(1);
+
assertThat(latest.fileIO().exists(paths.toAlignedPath(sidecars.get(0),
file))).isTrue();
+ // Any main-file fallback or full scan now fails; only the sparse
ROW path can work.
+ assertThat(latest.fileIO().delete(paths.toPath(file),
false)).isTrue();
+ }
+ InternalRowSerializer fullSerializer = new
InternalRowSerializer(fullType);
+ List<InternalRow> expected = new ArrayList<>();
+ for (int i = 50; i <= 54; i++) {
+ GenericRow row = variantSidecarRow(i);
+ expected.add(
+ fullSerializer
+ .toBinaryRow(
+ GenericRow.of(
+ row.getRow(2, 2),
+ merged
+ ? variantSidecarRow(i +
1000).getVariant(1)
+ : row.getVariant(1),
+ i,
+ i + 2L))
+ .copy());
+ }
+ assertThat(collect(fullRead.newRead().executeFilter(), plan, fullType))
+ .containsExactlyInAnyOrderElementsOf(expected);
+
+ RowType extraction =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.INT(), "$.x", false, "UTC")
+ .field(DataTypes.VARIANT(), "$")
+ .build();
+ RowType nestedType = (RowType)
latest.rowType().getField("nested").type();
+ RowType extractedType =
+ DataTypes.ROW(
+ latest.rowType()
+ .getField("nested")
+ .newType(
+ DataTypes.ROW(
+
nestedType.getField("detail").newType(extraction))),
+
latest.rowType().getField("renamed_payload").newType(extraction),
+ latest.rowType().getField("id"));
+ ReadBuilder extractedRead =
+
latest.newReadBuilder().withReadType(extractedType).withFilter(filter);
+ InternalRowSerializer extractedSerializer = new
InternalRowSerializer(extractedType);
+ expected.clear();
+ for (int i = 50; i <= 54; i++) {
+ GenericRow row = variantSidecarRow(i);
+ InternalRow nested = row.getRow(2, 2);
+ Object nestedValue = null;
+ if (nested != null) {
+ nestedValue =
+ GenericRow.of(
+ nested.isNullAt(1)
+ ? null
+ : GenericRow.of(
+ i % 5 == 1 ? null : -i,
nested.getVariant(1)));
+ }
+ expected.add(
+ extractedSerializer
+ .toBinaryRow(
+ GenericRow.of(
+ nestedValue,
+ row.isNullAt(1)
+ ? null
+ : GenericRow.of(
+ i % 5 == 0
+ ? i +
(merged ? 1000 : 0)
+ : null,
+ merged
+ ?
variantSidecarRow(i + 1000)
+
.getVariant(1)
+ :
row.getVariant(1)),
+ i))
+ .copy());
+ }
+ assertThat(
+ collect(
+ extractedRead.newRead().executeFilter(),
+ extractedRead.newScan().plan(),
+ extractedType))
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+
+ private static GenericRow variantSidecarRow(int id) {
+ int kind = Math.floorMod(id, 5);
+ GenericVariant payload =
+ kind == 1
+ ? null
+ : GenericVariant.fromJson(
+ kind == 0
+ ? "{\"x\":" + id + "}"
+ : kind == 2 ? "null" : kind == 3 ?
"\"text\"" : "[1,null]");
+ GenericVariant detail =
+ kind == 2
+ ? null
+ : GenericVariant.fromJson(kind == 1 ? "null" :
"{\"x\":" + -id + "}");
+ return GenericRow.of(
+ id,
+ payload,
+ kind == 3 ? null :
GenericRow.of(BinaryString.fromString("label"), detail));
+ }
+
@Test
public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws
Exception {
Map<String, String> options = new HashMap<>();
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index bc8d30206f..7a42e74ff4 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1678,9 +1678,9 @@ class DataEvolutionSplitRead(SplitRead):
if DataFileMeta.is_blob_file(first_file.file_name):
field_ids = [self._get_field_id_from_write_cols(first_file)]
elif DataFileMeta.is_vector_file(first_file.file_name):
- field_ids =
self._get_field_ids_from_write_cols(first_file.write_cols)
+ field_ids = self._get_field_ids_from_write_cols(first_file)
elif first_file.write_cols:
- field_ids =
self._get_field_ids_from_write_cols(first_file.write_cols)
+ field_ids = self._get_field_ids_from_write_cols(first_file)
else:
# For regular files without write_cols, derive field IDs from
# the file's schema version, not the current table schema.
@@ -1863,10 +1863,14 @@ class DataEvolutionSplitRead(SplitRead):
return field.id
raise ValueError(f"Field {field_name} not found in file schema
{file.schema_id}")
- def _get_field_ids_from_write_cols(self, write_cols: List[str]) ->
List[int]:
+ def _get_field_ids_from_write_cols(self, file: DataFileMeta) -> List[int]:
+ # write_cols records names from the schema that wrote this file.
Resolve
+ # them there so a rename cannot hide a newer column version, or rebind
a
+ # dropped/re-added name to a different field ID.
+ file_fields = self._resolve_schema(file.schema_id).fields
field_ids = []
- for field_name in write_cols:
- for field in self.table.fields:
+ for field_name in file.write_cols:
+ for field in file_fields:
if field.name == field_name:
field_ids.append(field.id)
field_ids.append(SpecialFields.ROW_ID.id)
diff --git a/paimon-python/pypaimon/tests/variant_row_sidecar_index_test.py
b/paimon-python/pypaimon/tests/variant_row_sidecar_index_test.py
new file mode 100644
index 0000000000..e099f2ab1e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/variant_row_sidecar_index_test.py
@@ -0,0 +1,162 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Global indexes must select ROW sidecars without losing Variant column
versions."""
+
+import os
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.globalindex.data_evolution_global_index_scanner import
DataEvolutionGlobalIndexScanner
+from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.utils.range import Range
+
+pytestmark = [pytest.mark.python_plan, pytest.mark.python_read,
+ pytest.mark.python_write, pytest.mark.python_commit]
+
+
+def _commit(builder, messages):
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+
+def _append(table, data):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(data)
+ messages = writer.prepare_commit()
+ _commit(builder, messages)
+ return messages
+ finally:
+ writer.close()
+
+
+def _data(schema, ids):
+ roots, nested = [], []
+ for row in ids:
+ value = ({'score': row} if row not in [6, 7, 8, 9]
+ else {6: None, 7: None, 8: 'scalar', 9: [9, None]}[row])
+ roots.append(None if row == 6 else GenericVariant.from_python(value))
+ child = (None if row == 7 else GenericVariant.from_python(
+ None if row == 6 else {'score': row * 10}))
+ nested.append(None if row == 8 else {
+ 'detail': None if child is None else {'value': child.value(),
'metadata': child.metadata()},
+ 'tag': 'nested-%d' % row})
+ return pa.table({
+ 'id': list(ids),
+ 'category': ['match' if row in range(5, 10) else 'other' for row in
ids],
+ 'payload': GenericVariant.to_arrow_array(roots),
+ 'nested': nested,
+ }, schema=schema)
+
+
+def _indexed_read(table):
+ predicate =
table.new_read_builder().new_predicate_builder().equal('category', 'match')
+ builder = table.new_read_builder().with_filter(predicate).with_projection([
+ 'renamed_payload', 'id', 'nested', '_ROW_ID'])
+ splits = builder.new_scan().plan().splits()
+ assert splits and all(isinstance(split, IndexedSplit) for split in splits)
+ assert [r for split in splits for r in split.row_ranges()] == [Range(25,
29)]
+ return builder.new_read().to_arrow(splits)
+
+
[email protected]('index_type', ['bitmap', 'btree'])
[email protected]('column_update', [False, True])
+def test_global_index_reads_variant_sidecars_after_historical_rename(tmp_path,
index_type, column_update):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ variant = GenericVariant.to_arrow_array([]).type
+ schema = pa.schema([
+ ('id', pa.int32()), ('category', pa.string()), ('payload', variant),
+ ('nested', pa.struct([('detail', variant), ('tag', pa.string())])),
+ ])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'file.format': 'parquet', 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
'data-evolution.row-sidecar.enabled': 'true',
+ 'global-index.enabled': 'true', 'scan.native-plan.enabled': 'false',
+ 'read.native.enabled': 'false', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ # The second physical file starts at global row ID 20, so global index
+ # positions 25..29 must be converted into local ROW positions 5..9.
+ messages = _append(table, _data(schema, range(-20, 0)))
+ data = _data(schema, range(100))
+ messages += _append(table, data)
+ assert table.create_global_index('category', index_type=index_type) > 0
+ predicate =
table.new_read_builder().new_predicate_builder().equal('category', 'match')
+ with DataEvolutionGlobalIndexScanner.create(table, predicate=predicate) as
scanner:
+ result = scanner.scan(predicate)
+ assert result.results().to_range_list() == [Range(25, 29)]
+ expected = data.slice(5, 5).to_pylist()
+ if column_update:
+ values = GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'score': 500}),
GenericVariant.from_python(None),
+ None, GenericVariant.from_python('updated'),
GenericVariant.from_python([99, None]),
+ ])
+ builder = table.new_batch_write_builder()
+ updates = builder.new_update().new_update_by_row_id().update_columns(
+ pa.table({'_ROW_ID': list(range(25, 30)), 'payload': values}),
['payload'])
+ _commit(builder, updates)
+ updated_files = [file for message in updates for file in
message.new_files]
+ assert len(updated_files) == 1
+ assert updated_files[0].write_cols == ['payload']
+ assert updated_files[0].row_count == 100
+ assert updated_files[0].first_row_id == 20
+ messages += updates
+ for row, value in zip(expected, values.to_pylist()):
+ row['payload'] = value
+
+ payload_id = table.field_dict['payload'].id
+ catalog.alter_table('default.t', [
+ SchemaChange.rename_column('payload', 'renamed_payload'),
+ SchemaChange.rename_column(['nested', 'detail'], 'renamed_detail'),
+ ], False)
+ table = catalog.get_table('default.t')
+ assert table.field_dict['renamed_payload'].id == payload_id
+ expected = [{
+ 'renamed_payload': row['payload'], 'id': row['id'],
+ 'nested': (None if row['nested'] is None else {
+ 'renamed_detail': row['nested']['detail'], 'tag':
row['nested']['tag']}),
+ '_ROW_ID': row['id'] + 20,
+ } for row in expected]
+
+ # This assertion also guards against silently choosing the old full file
+ # instead of the latest column-only file after its write_cols was renamed.
+ assert _indexed_read(table).to_pylist() == expected
+ for message in messages:
+ for file in message.new_files:
+ sidecars = [path for path in file.collect_files() if
path.endswith('.row')]
+ assert len(sidecars) == 1 and os.path.exists(sidecars[0])
+ os.remove(file.external_path or file.file_path)
+ # A correct result now requires every contributing column bunch to use ROW.
+ assert _indexed_read(table).to_pylist() == expected
+
+ # Without the real global index there is no row selection and the deleted
+ # primary must be opened. This rules out an accidental full-scan success.
+ unindexed = table.copy({'global-index.enabled': 'false'})
+ builder = unindexed.new_read_builder().with_filter(predicate)
+ with pytest.raises(OSError):
+ builder.new_read().to_arrow(builder.new_scan().plan().splits())
