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 10cf6ebc7c [java][python] Read Python-planned splits with the native
reader (#10017)
10cf6ebc7c is described below
commit 10cf6ebc7c4f0ba6066d32d8aed2b0b85e0b8349
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 20 21:11:17 2026 +0800
[java][python] Read Python-planned splits with the native reader (#10017)
---
.../DataEvolutionNormalCompactTask.java | 15 +-
.../DataEvolutionGlobalIndexRefreshPlanner.java | 4 +-
.../java/org/apache/paimon/io/DataFileMeta.java | 8 +-
.../apache/paimon/io/DataFileMetaSerializer.java | 6 +-
.../org/apache/paimon/io/PojoDataFileMeta.java | 41 +-
.../apache/paimon/io/ProjectedDataFileMeta.java | 6 +-
.../apache/paimon/utils/DataEvolutionUtils.java | 6 +-
.../DataEvolutionNormalCompactTaskTest.java | 29 +-
...DataEvolutionGlobalIndexRefreshPlannerTest.java | 24 +-
.../paimon/io/DataFileMetaSerializerTest.java | 40 +-
.../paimon/io/ProjectedDataFileMetaTest.java | 8 +-
...festCommittableSerializerCompatibilityTest.java | 2 +-
.../manifest/ManifestEntrySerializerTest.java | 9 +-
.../apache/paimon/manifest/ManifestFileTest.java | 6 +-
.../table/sink/CommitMessageSerializerTest.java | 2 +-
.../table/source/DataSplitCompatibleTest.java | 4 +-
.../paimon/utils/DataEvolutionUtilsTest.java | 12 +-
.../ChangelogCompactTaskSerializerTest.java | 2 +-
.../PendingSplitsCheckpointSerializerTest.java | 18 +-
.../pypaimon/manifest/manifest_file_manager.py | 4 +
.../pypaimon/manifest/schema/data_file_meta.py | 11 +
paimon-python/pypaimon/read/native_plan.py | 22 +-
.../pypaimon/read/reader/auth_masking_reader.py | 84 ++--
.../read/scanner/chunk_shuffle_split_generator.py | 7 +-
.../read/scanner/data_evolution_split_generator.py | 8 +-
.../pypaimon/read/scanner/file_scanner.py | 11 +-
.../scanner/primary_key_table_split_generator.py | 7 +-
.../pypaimon/read/scanner/split_generator.py | 7 +-
paimon-python/pypaimon/read/sliced_split.py | 4 +
paimon-python/pypaimon/read/split.py | 12 +-
paimon-python/pypaimon/read/split_serializer.py | 444 +++++++++++++++++++--
.../pypaimon/read/streaming_table_scan.py | 15 +-
paimon-python/pypaimon/read/table_read.py | 22 +-
paimon-python/pypaimon/read/table_scan.py | 4 +-
.../table/source/primary_key_scored_result.py | 9 +-
.../tests/data_evolution_split_generator_test.py | 4 +-
.../pypaimon/tests/interval_partition_test.py | 4 +-
.../tests/manifest/manifest_manager_test.py | 3 +
.../tests/native_plan_chunk_shuffle_test.py | 2 +
.../pypaimon/tests/native_plan_incremental_test.py | 4 +-
.../pypaimon/tests/native_plan_integration_test.py | 113 +++++-
paimon-python/pypaimon/tests/native_read_test.py | 52 +++
.../tests/primary_key_index_definitions_test.py | 1 +
.../scanner/chunk_shuffle_split_generator_test.py | 8 +-
.../pypaimon/tests/split_serializer_test.py | 89 ++++-
.../paimon/spark/copy/CopyFilesUtilTest.java | 6 +-
46 files changed, 971 insertions(+), 228 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
index 6a46a4425f..7065a37bc1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
@@ -177,9 +177,9 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
long minSequenceNumber = minSequenceId(compactBefore);
long maxSequenceNumber = maxSequenceId(compactBefore);
long nextRowId = firstRowId;
- long[] columnMaxSequenceNumbers =
+ long[] writeColsSequences =
options.ignoreIndexColumnUpdate() && !writeResult.isEmpty()
- ? compactedColumnMaxSequenceNumbers(
+ ? compactedWriteColsSequences(
table,
writeResult
.get(0)
@@ -189,8 +189,8 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
DataFileMeta dataFileMeta =
file.assignFirstRowId(nextRowId)
.assignSequenceNumber(minSequenceNumber,
maxSequenceNumber);
- if (columnMaxSequenceNumbers != null) {
- dataFileMeta =
dataFileMeta.withColumnMaxSequenceNumbers(columnMaxSequenceNumbers);
+ if (writeColsSequences != null) {
+ dataFileMeta =
dataFileMeta.withWriteColsSequences(writeColsSequences);
}
compactAfter.add(dataFileMeta);
nextRowId += dataFileMeta.rowCount();
@@ -230,8 +230,7 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
}
@Nullable
- private long[] compactedColumnMaxSequenceNumbers(
- FileStoreTable table, DataFileMeta outputFile) {
+ private long[] compactedWriteColsSequences(FileStoreTable table,
DataFileMeta outputFile) {
SchemaManager schemaManager = table.schemaManager();
Map<Long, TableSchema> schemaCache = new HashMap<>();
Function<Long, TableSchema> schemaLoader =
@@ -244,12 +243,12 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
fileFieldsCache.computeIfAbsent(
Pair.of(input.schemaId(), input.writeCols()),
key -> fileFields(schemaLoader, input));
- long[] inputColumnSequences = input.columnMaxSequenceNumbers();
+ long[] inputWriteColsSequences = input.writeColsSequences();
for (int inputPosition = 0; inputPosition < inputFields.size();
inputPosition++) {
fieldMaxSequences.merge(
inputFields.get(inputPosition).id(),
fieldMaxSequenceNumber(
- input, inputColumnSequences, inputPosition,
inputFields.size()),
+ input, inputWriteColsSequences, inputPosition,
inputFields.size()),
Math::max);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
index c1d888587a..4dc6fcaf70 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
@@ -255,7 +255,7 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
TableSchema fileSchema =
schemaLoader.apply(file.schemaId());
return fileFields(fileSchema, file);
});
- long[] columnSequences = file.columnMaxSequenceNumbers();
+ long[] writeColsSequences = file.writeColsSequences();
long indexedMaxSequence = Long.MIN_VALUE;
for (int position = 0; position < physicalFields.size(); position++) {
if (indexedFieldIds.contains(physicalFields.get(position).id())) {
@@ -263,7 +263,7 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
Math.max(
indexedMaxSequence,
fieldMaxSequenceNumber(
- file, columnSequences, position,
physicalFields.size()));
+ file, writeColsSequences, position,
physicalFields.size()));
}
}
if (indexedMaxSequence != Long.MIN_VALUE) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
index a8983256e1..955750613a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
@@ -266,7 +266,7 @@ public interface DataFileMeta {
@Nullable String externalPath,
@Nullable Long firstRowId,
@Nullable List<String> writeCols,
- @Nullable long[] columnMaxSequenceNumbers) {
+ @Nullable long[] writeColsSequences) {
return new PojoDataFileMeta(
fileName,
fileSize,
@@ -288,7 +288,7 @@ public interface DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
String fileName();
@@ -377,7 +377,7 @@ public interface DataFileMeta {
* value otherwise. A null value means that only the file-level sequence
range is available.
*/
@Nullable
- long[] columnMaxSequenceNumbers();
+ long[] writeColsSequences();
DataFileMeta upgrade(int newLevel);
@@ -387,7 +387,7 @@ public interface DataFileMeta {
DataFileMeta assignSequenceNumber(long minSequenceNumber, long
maxSequenceNumber);
- DataFileMeta withColumnMaxSequenceNumbers(long[] columnMaxSequenceNumbers);
+ DataFileMeta withWriteColsSequences(long[] writeColsSequences);
DataFileMeta assignFirstRowId(long firstRowId);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
index f148bb397e..ba49e4e547 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
@@ -43,7 +43,7 @@ public class DataFileMetaSerializer extends
ObjectSerializer<DataFileMeta> {
@Override
public InternalRow toRow(DataFileMeta meta) {
- long[] columnMaxSequenceNumbers = meta.columnMaxSequenceNumbers();
+ long[] writeColsSequences = meta.writeColsSequences();
return GenericRow.of(
BinaryString.fromString(meta.fileName()),
meta.fileSize(),
@@ -65,9 +65,7 @@ public class DataFileMetaSerializer extends
ObjectSerializer<DataFileMeta> {
meta.externalPath().map(BinaryString::fromString).orElse(null),
meta.firstRowId(),
meta.writeCols() == null ? null :
toStringArrayData(meta.writeCols()),
- columnMaxSequenceNumbers == null
- ? null
- : new GenericArray(columnMaxSequenceNumbers));
+ writeColsSequences == null ? null : new
GenericArray(writeColsSequences));
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java
index 6cf1eaa0d5..af65b0800e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java
@@ -78,7 +78,7 @@ public class PojoDataFileMeta implements DataFileMeta {
private final @Nullable List<String> writeCols;
- private final @Nullable long[] columnMaxSequenceNumbers;
+ private final @Nullable long[] writeColsSequences;
public PojoDataFileMeta(
String fileName,
@@ -101,7 +101,7 @@ public class PojoDataFileMeta implements DataFileMeta {
@Nullable String externalPath,
@Nullable Long firstRowId,
@Nullable List<String> writeCols,
- @Nullable long[] columnMaxSequenceNumbers) {
+ @Nullable long[] writeColsSequences) {
this.fileName = fileName;
this.fileSize = fileSize;
@@ -126,8 +126,7 @@ public class PojoDataFileMeta implements DataFileMeta {
this.externalPath = externalPath;
this.firstRowId = firstRowId;
this.writeCols = writeCols;
- this.columnMaxSequenceNumbers =
- columnMaxSequenceNumbers == null ? null :
columnMaxSequenceNumbers.clone();
+ this.writeColsSequences = writeColsSequences == null ? null :
writeColsSequences.clone();
}
@Override
@@ -246,8 +245,8 @@ public class PojoDataFileMeta implements DataFileMeta {
@Nullable
@Override
- public long[] columnMaxSequenceNumbers() {
- return columnMaxSequenceNumbers == null ? null :
columnMaxSequenceNumbers.clone();
+ public long[] writeColsSequences() {
+ return writeColsSequences == null ? null : writeColsSequences.clone();
}
@Override
@@ -274,7 +273,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -301,7 +300,7 @@ public class PojoDataFileMeta implements DataFileMeta {
newExternalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -327,7 +326,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -353,11 +352,11 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
- public PojoDataFileMeta withColumnMaxSequenceNumbers(long[]
columnMaxSequenceNumbers) {
+ public PojoDataFileMeta withWriteColsSequences(long[] writeColsSequences) {
return new PojoDataFileMeta(
fileName,
fileSize,
@@ -379,7 +378,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -405,7 +404,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -431,7 +430,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
newFirstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -457,7 +456,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -483,7 +482,7 @@ public class PojoDataFileMeta implements DataFileMeta {
newExternalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -509,7 +508,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- columnMaxSequenceNumbers);
+ writeColsSequences);
}
@Override
@@ -541,7 +540,7 @@ public class PojoDataFileMeta implements DataFileMeta {
&& Objects.equals(externalPath,
that.externalPath().orElse(null))
&& Objects.equals(firstRowId, that.firstRowId())
&& Objects.equals(writeCols, that.writeCols())
- && Arrays.equals(columnMaxSequenceNumbers,
that.columnMaxSequenceNumbers());
+ && Arrays.equals(writeColsSequences,
that.writeColsSequences());
}
@Override
@@ -567,7 +566,7 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- Arrays.hashCode(columnMaxSequenceNumbers));
+ Arrays.hashCode(writeColsSequences));
}
@Override
@@ -578,7 +577,7 @@ public class PojoDataFileMeta implements DataFileMeta {
+ "minSequenceNumber: %d, maxSequenceNumber: %d, "
+ "schemaId: %d, level: %d, extraFiles: %s,
creationTime: %s, "
+ "deleteRowCount: %d, fileSource: %s, valueStatsCols:
%s, externalPath: %s, "
- + "firstRowId: %s, writeCols: %s,
columnMaxSequenceNumbers: %s}",
+ + "firstRowId: %s, writeCols: %s, writeColsSequences:
%s}",
fileName,
fileSize,
rowCount,
@@ -599,6 +598,6 @@ public class PojoDataFileMeta implements DataFileMeta {
externalPath,
firstRowId,
writeCols,
- Arrays.toString(columnMaxSequenceNumbers));
+ Arrays.toString(writeColsSequences));
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java
index 70da63dba7..1eaaf9af39 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java
@@ -256,7 +256,7 @@ public final class ProjectedDataFileMeta implements
DataFileMeta {
@Nullable
@Override
- public long[] columnMaxSequenceNumbers() {
+ public long[] writeColsSequences() {
int position = requiredPosition(Fields.WRITE_COLS_SEQUENCES);
InternalRow row = currentRow();
if (row.isNullAt(position)) {
@@ -308,8 +308,8 @@ public final class ProjectedDataFileMeta implements
DataFileMeta {
}
@Override
- public DataFileMeta withColumnMaxSequenceNumbers(long[]
columnMaxSequenceNumbers) {
- throw unsupportedOperation("withColumnMaxSequenceNumbers(long[])");
+ public DataFileMeta withWriteColsSequences(long[] writeColsSequences) {
+ throw unsupportedOperation("withWriteColsSequences(long[])");
}
@Override
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 a4e702184b..f8a2112eb6 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
@@ -291,13 +291,13 @@ public class DataEvolutionUtils {
/** Returns the latest sequence known for a physical field position in the
file. */
public static long fieldMaxSequenceNumber(
DataFileMeta file,
- @Nullable long[] columnSequences,
+ @Nullable long[] writeColsSequences,
int fieldPosition,
int physicalFieldCount) {
- if (columnSequences == null || columnSequences.length !=
physicalFieldCount) {
+ if (writeColsSequences == null || writeColsSequences.length !=
physicalFieldCount) {
return file.maxSequenceNumber();
}
- return columnSequences[fieldPosition];
+ return writeColsSequences[fieldPosition];
}
/**
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 6d15a8c345..2464cbf892 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
@@ -82,7 +82,7 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
}
@Test
- public void testPropagateColumnSequencesAcrossCompactions() throws
Exception {
+ public void testPropagateWriteColsSequencesAcrossCompactions() throws
Exception {
write();
int f0Id = getTableDefault().rowType().getField("f0").id();
@@ -93,22 +93,23 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
1,
CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
- long f0Sequence = columnSequence(firstCompact, f0Id);
+ long f0Sequence = writeColSequence(firstCompact, f0Id);
assertThat(f0Sequence).isLessThan(firstCompact.maxSequenceNumber());
- assertThat(columnSequence(firstCompact,
f1Id)).isEqualTo(firstCompact.maxSequenceNumber());
+ assertThat(writeColSequence(firstCompact, f1Id))
+ .isEqualTo(firstCompact.maxSequenceNumber());
DataFileMeta secondCompact =
updateColumnsAndCompact(
Collections.singletonList("f1"),
2,
CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
- assertThat(columnSequence(secondCompact, f0Id)).isEqualTo(f0Sequence);
- assertThat(columnSequence(secondCompact, f1Id))
+ assertThat(writeColSequence(secondCompact,
f0Id)).isEqualTo(f0Sequence);
+ assertThat(writeColSequence(secondCompact, f1Id))
.isEqualTo(secondCompact.maxSequenceNumber());
}
@Test
- public void testOmitAndReconstructRedundantColumnSequences() throws
Exception {
+ public void testOmitAndReconstructRedundantWriteColsSequences() throws
Exception {
write();
DataFileMeta fullUpdate =
@@ -116,7 +117,7 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
Arrays.asList("f0", "f1"),
1,
CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
- assertThat(fullUpdate.columnMaxSequenceNumbers()).isNull();
+ assertThat(fullUpdate.writeColsSequences()).isNull();
int f0Id = getTableDefault().rowType().getField("f0").id();
DataFileMeta partialUpdate =
@@ -124,13 +125,13 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
Collections.singletonList("f1"),
2,
CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
- assertThat(columnSequence(partialUpdate, f0Id))
+ assertThat(writeColSequence(partialUpdate, f0Id))
.isEqualTo(fullUpdate.maxSequenceNumber())
.isLessThan(partialUpdate.maxSequenceNumber());
}
@Test
- public void testOmitColumnSequencesUnlessUpdatesAreIgnored() throws
Exception {
+ public void testOmitWriteColsSequencesUnlessUpdatesAreIgnored() throws
Exception {
write();
DataFileMeta compacted =
@@ -138,7 +139,7 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
Collections.singletonList("f1"),
1,
CoreOptions.GlobalIndexColumnUpdateAction.THROW_ERROR);
- assertThat(compacted.columnMaxSequenceNumbers()).isNull();
+ assertThat(compacted.writeColsSequences()).isNull();
}
@ParameterizedTest
@@ -221,9 +222,9 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
assertThat(file.minSequenceNumber()).isEqualTo(original.minSequenceNumber());
assertThat(file.maxSequenceNumber()).isEqualTo(maxSequenceNumber);
if (updateColumn) {
- assertThat(columnSequence(file,
table.rowType().getField("f1").id()))
+ assertThat(writeColSequence(file,
table.rowType().getField("f1").id()))
.isEqualTo(original.maxSequenceNumber());
- assertThat(columnSequence(file,
table.rowType().getField("f0").id()))
+ assertThat(writeColSequence(file,
table.rowType().getField("f0").id()))
.isEqualTo(maxSequenceNumber);
}
nextRowId += file.rowCount();
@@ -793,12 +794,12 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
return rowRangeFiles.get(0);
}
- private long columnSequence(DataFileMeta file, int fieldId) throws
Exception {
+ private long writeColSequence(DataFileMeta file, int fieldId) throws
Exception {
TableSchema fileSchema =
getTableDefault().schemaManager().schema(file.schemaId());
boolean nestedFieldEnabled =
new
CoreOptions(fileSchema.options()).dataEvolutionNestedFieldEnabled();
List<DataField> fields = fileFields(fileSchema.fields(), file,
nestedFieldEnabled);
- long[] sequences = file.columnMaxSequenceNumbers();
+ long[] sequences = file.writeColsSequences();
assertThat(sequences).hasSize(fields.size());
for (int i = 0; i < fields.size(); i++) {
if (fields.get(i).id() == fieldId) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
index f7bc10051c..0b8e07b39d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
@@ -155,13 +155,13 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
}
@Test
- void testUsesColumnSequenceNumbersForCompactedFullFile() {
+ void testUsesWriteColsSequencesForCompactedFullFile() {
IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
assertThat(
plan(
Collections.singletonList(
- dataWithColumnSequences(
+ dataWithWriteColsSequences(
"unrelated-compact",
0,
100,
@@ -172,7 +172,7 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
assertThat(
plan(
Collections.singletonList(
- dataWithColumnSequences(
+ dataWithWriteColsSequences(
"index-compact",
0,
100,
@@ -187,13 +187,13 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
}
@Test
- void testColumnSequenceNumbersFollowWriteColsOrder() {
+ void testWriteColsSequencesFollowWriteColsOrder() {
IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
assertThat(
plan(
Collections.singletonList(
- dataWithColumnSequences(
+ dataWithWriteColsSequences(
"reordered-compact",
0,
100,
@@ -206,13 +206,13 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
}
@Test
- void testColumnSequenceNumbersIgnoreRowTrackingFields() {
+ void testWriteColsSequencesIgnoreRowTrackingFields() {
IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
assertThat(
plan(
Collections.singletonList(
- dataWithColumnSequences(
+ dataWithWriteColsSequences(
"row-tracking-compact",
0,
100,
@@ -241,13 +241,13 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
}
@Test
- void testMalformedColumnSequenceNumbersFallBackToFileSequence() {
+ void testMalformedWriteColsSequencesFallBackToFileSequence() {
IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
assertThat(
plan(
Collections.singletonList(
- dataWithColumnSequences(
+ dataWithWriteColsSequences(
"malformed-compact",
0,
100,
@@ -599,17 +599,17 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1,
file);
}
- private ManifestEntry dataWithColumnSequences(
+ private ManifestEntry dataWithWriteColsSequences(
String fileName,
long firstRowId,
long rowCount,
long maxSequenceNumber,
- long[] columnSequences,
+ long[] writeColsSequences,
String... writeCols) {
DataFileMeta file =
data(fileName, firstRowId, rowCount, maxSequenceNumber, 1,
writeCols)
.file()
- .withColumnMaxSequenceNumbers(columnSequences);
+ .withWriteColsSequences(writeColsSequences);
return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1,
file);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaSerializerTest.java
index be87c063aa..d7f1c1bbf7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaSerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaSerializerTest.java
@@ -42,44 +42,44 @@ public class DataFileMetaSerializerTest extends
ObjectSerializerTestBase<DataFil
return gen.next()
.meta
.copy(Arrays.asList("extra1", "extra2"))
- .withColumnMaxSequenceNumbers(new long[] {3L, 42L});
+ .withWriteColsSequences(new long[] {3L, 42L});
}
@Test
- void testCopyOperationsPreserveColumnSequences() {
+ void testCopyOperationsPreserveWriteColsSequences() {
DataFileMeta file = object();
- assertColumnSequences(file.upgrade(file.level() + 1));
- assertColumnSequences(file.rename("renamed.parquet"));
- assertColumnSequences(file.copyWithoutStats());
- assertColumnSequences(file.assignSequenceNumber(1L, 2L));
- assertColumnSequences(file.assignFirstRowId(1L));
- assertColumnSequences(file.newFirstRowId(null));
- assertColumnSequences(file.copy(Collections.emptyList()));
-
assertColumnSequences(file.newExternalPath("external/renamed.parquet"));
- assertColumnSequences(file.copy(new byte[] {1}));
+ assertWriteColsSequences(file.upgrade(file.level() + 1));
+ assertWriteColsSequences(file.rename("renamed.parquet"));
+ assertWriteColsSequences(file.copyWithoutStats());
+ assertWriteColsSequences(file.assignSequenceNumber(1L, 2L));
+ assertWriteColsSequences(file.assignFirstRowId(1L));
+ assertWriteColsSequences(file.newFirstRowId(null));
+ assertWriteColsSequences(file.copy(Collections.emptyList()));
+
assertWriteColsSequences(file.newExternalPath("external/renamed.parquet"));
+ assertWriteColsSequences(file.copy(new byte[] {1}));
}
@Test
- void testLegacySerializerDropsColumnSequences() {
+ void testLegacySerializerDropsWriteColsSequences() {
DataFileMetaWriteColsLegacySerializer legacy = new
DataFileMetaWriteColsLegacySerializer();
DataFileMeta file = legacy.fromRow(legacy.toRow(object()));
- assertThat(file.columnMaxSequenceNumbers()).isNull();
+ assertThat(file.writeColsSequences()).isNull();
}
@Test
- void testColumnSequencesAreDefensivelyCopied() {
+ void testWriteColsSequencesAreDefensivelyCopied() {
long[] sequences = {3L, 42L};
- DataFileMeta file =
gen.next().meta.withColumnMaxSequenceNumbers(sequences);
+ DataFileMeta file = gen.next().meta.withWriteColsSequences(sequences);
sequences[0] = 100L;
- assertColumnSequences(file);
+ assertWriteColsSequences(file);
- long[] returned = file.columnMaxSequenceNumbers();
+ long[] returned = file.writeColsSequences();
returned[1] = 100L;
- assertColumnSequences(file);
+ assertWriteColsSequences(file);
}
- private void assertColumnSequences(DataFileMeta file) {
- assertThat(file.columnMaxSequenceNumbers()).containsExactly(3L, 42L);
+ private void assertWriteColsSequences(DataFileMeta file) {
+ assertThat(file.writeColsSequences()).containsExactly(3L, 42L);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/ProjectedDataFileMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/ProjectedDataFileMetaTest.java
index 744aa14ed0..f1d6b35bbd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/io/ProjectedDataFileMetaTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/io/ProjectedDataFileMetaTest.java
@@ -56,7 +56,7 @@ public class ProjectedDataFileMetaTest {
"external/dir/data.parquet",
10L,
Collections.singletonList("write_col"))
- .withColumnMaxSequenceNumbers(new long[] {11L});
+ .withWriteColsSequences(new long[] {11L});
ProjectedDataFileMeta actual =
ProjectedDataFileMeta.Projection.create(DataFileMeta.SCHEMA)
.createDataFile()
@@ -92,7 +92,7 @@ public class ProjectedDataFileMetaTest {
assertThat(actual.firstRowId()).isEqualTo(10L);
assertThat(actual.nonNullFirstRowId()).isEqualTo(10L);
assertThat(actual.writeCols()).containsExactly("write_col");
- assertThat(actual.columnMaxSequenceNumbers()).containsExactly(11L);
+ assertThat(actual.writeColsSequences()).containsExactly(11L);
assertThat(actual.containsWriteColumn(BinaryString.fromString("write_col"))).isTrue();
assertThat(actual.containsWriteColumn(BinaryString.fromString("other"))).isFalse();
assertThat(actual.toFileSelection(Collections.singletonList(new
Range(11L, 12L))))
@@ -104,8 +104,8 @@ public class ProjectedDataFileMetaTest {
assertUnsupported(
() -> actual.assignSequenceNumber(4L, 5L),
"assignSequenceNumber(long, long)");
assertUnsupported(
- () -> actual.withColumnMaxSequenceNumbers(new long[] {2L}),
- "withColumnMaxSequenceNumbers(long[])");
+ () -> actual.withWriteColsSequences(new long[] {2L}),
+ "withWriteColsSequences(long[])");
assertUnsupported(() -> actual.assignFirstRowId(20L),
"assignFirstRowId(long)");
assertUnsupported(() -> actual.newFirstRowId(20L),
"newFirstRowId(Long)");
assertUnsupported(() -> actual.copy(Collections.emptyList()),
"copy(List)");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java
index dbe1ebfab0..c25bee603e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java
@@ -75,7 +75,7 @@ public class ManifestCommittableSerializerCompatibilityTest {
1L,
Arrays.asList("a", "b"),
null)
- .withColumnMaxSequenceNumbers(new long[] {3L, 5L});
+ .withWriteColsSequences(new long[] {3L, 5L});
IndexFileMeta indexFile =
new IndexFileMeta(
"index-type", "index-file", 100L, 10L,
(GlobalIndexMeta) null, null);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySerializerTest.java
index 5005f04cc9..80e82dd945 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySerializerTest.java
@@ -40,21 +40,22 @@ public class ManifestEntrySerializerTest extends
ObjectSerializerTestBase<Manife
@Test
void testWriteColsLegacySerializer() throws IOException {
ManifestEntry expected = gen.next();
- ManifestEntry withColumnSequences =
+ ManifestEntry withWriteColsSequences =
ManifestEntry.create(
expected.kind(),
expected.partition(),
expected.bucket(),
expected.totalBuckets(),
- expected.file().withColumnMaxSequenceNumbers(new
long[] {3L, 42L}));
+ expected.file().withWriteColsSequences(new long[] {3L,
42L}));
ManifestEntryWriteColsLegacySerializer serializer =
new ManifestEntryWriteColsLegacySerializer();
ManifestEntry actual =
-
serializer.deserializeFromBytes(serializer.serializeToBytes(withColumnSequences));
+ serializer.deserializeFromBytes(
+ serializer.serializeToBytes(withWriteColsSequences));
assertThat(actual).isEqualTo(expected);
- assertThat(actual.file().columnMaxSequenceNumbers()).isNull();
+ assertThat(actual.file().writeColsSequences()).isNull();
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
index 24f84659d1..2eb0c1f4cb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
@@ -624,7 +624,7 @@ public class ManifestFileTest {
}
@Test
- void testLegacyAvroReaderSkipsColumnSequenceNumbers() throws Exception {
+ void testLegacyAvroReaderSkipsWriteColsSequences() throws Exception {
ManifestEntry expected = gen.next();
ManifestEntry source =
ManifestEntry.create(
@@ -632,7 +632,7 @@ public class ManifestFileTest {
expected.partition(),
expected.bucket(),
expected.totalBuckets(),
- expected.file().withColumnMaxSequenceNumbers(new
long[] {3L, 42L}));
+ expected.file().withWriteColsSequences(new long[] {3L,
42L}));
List<DataField> legacyManifestFields =
ManifestEntry.MANIFEST_ROW_TYPE.getFields().stream()
.map(
@@ -664,7 +664,7 @@ public class ManifestFileTest {
}
assertThat(actual).isEqualTo(expected);
- assertThat(actual.file().columnMaxSequenceNumbers()).isNull();
+ assertThat(actual.file().writeColsSequences()).isNull();
}
@ParameterizedTest
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java
index bc36deedca..04f40f71e5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java
@@ -47,7 +47,7 @@ public class CommitMessageSerializerTest {
dataIncrement
.newFiles()
.get(0)
- .withColumnMaxSequenceNumbers(new long[] {3L,
42L}));
+ .withWriteColsSequences(new long[] {3L, 42L}));
dataIncrement.newIndexFiles().addAll(Arrays.asList(randomIndexFile(),
randomIndexFile()));
dataIncrement
.deletedIndexFiles()
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
index e3f5403c68..7e7a76b815 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
@@ -221,7 +221,7 @@ public class DataSplitCompatibleTest {
DataFileTestDataGenerator gen =
DataFileTestDataGenerator.builder().build();
DataFileTestDataGenerator.Data data = gen.next();
List<DataFileMeta> files = new ArrayList<>();
- files.add(gen.next().meta.withColumnMaxSequenceNumbers(new long[] {3L,
42L}));
+ files.add(gen.next().meta.withWriteColsSequences(new long[] {3L,
42L}));
for (int i = 0; i < ThreadLocalRandom.current().nextInt(10); i++) {
files.add(gen.next().meta);
}
@@ -833,7 +833,7 @@ public class DataSplitCompatibleTest {
12L,
Arrays.asList("a", "b", "c", "f"),
null)
- .withColumnMaxSequenceNumbers(new long[] {15L, 100L,
150L, 200L});
+ .withWriteColsSequences(new long[] {15L, 100L, 150L,
200L});
List<DataFileMeta> dataFiles = Collections.singletonList(dataFile);
DeletionFile deletionFile = new DeletionFile("deletion_file", 100, 22,
33L);
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 4d82ecd479..028c12ee44 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
@@ -367,21 +367,19 @@ public class DataEvolutionUtilsTest {
public void
testFieldMaxSequenceNumberFallsBackForMissingOrMalformedArray() {
DataFileMeta legacy = dataFile("legacy.parquet", 10, null);
DataFileMeta malformed =
- dataFile("malformed.parquet", 10, null)
- .withColumnMaxSequenceNumbers(new long[] {5L});
+ dataFile("malformed.parquet", 10,
null).withWriteColsSequences(new long[] {5L});
DataFileMeta valid =
- dataFile("valid.parquet", 10, null)
- .withColumnMaxSequenceNumbers(new long[] {5L, 8L});
+ dataFile("valid.parquet", 10, null).withWriteColsSequences(new
long[] {5L, 8L});
assertThat(
DataEvolutionUtils.fieldMaxSequenceNumber(
- legacy, legacy.columnMaxSequenceNumbers(), 0,
2))
+ legacy, legacy.writeColsSequences(), 0, 2))
.isEqualTo(10L);
assertThat(
DataEvolutionUtils.fieldMaxSequenceNumber(
- malformed,
malformed.columnMaxSequenceNumbers(), 0, 2))
+ malformed, malformed.writeColsSequences(), 0,
2))
.isEqualTo(10L);
- long[] validSequences = valid.columnMaxSequenceNumbers();
+ long[] validSequences = valid.writeColsSequences();
assertThat(DataEvolutionUtils.fieldMaxSequenceNumber(valid,
validSequences, 0, 2))
.isEqualTo(5L);
assertThat(DataEvolutionUtils.fieldMaxSequenceNumber(valid,
validSequences, 1, 2))
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskSerializerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskSerializerTest.java
index 2dba94abd8..fe35fed0d4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskSerializerTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskSerializerTest.java
@@ -104,6 +104,6 @@ public class ChangelogCompactTaskSerializerTest {
null,
null,
null)
- .withColumnMaxSequenceNumbers(new long[] {1L});
+ .withWriteColsSequences(new long[] {1L});
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/PendingSplitsCheckpointSerializerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/PendingSplitsCheckpointSerializerTest.java
index e89ef16bb2..3a697d2d61 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/PendingSplitsCheckpointSerializerTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/PendingSplitsCheckpointSerializerTest.java
@@ -164,8 +164,8 @@ public class PendingSplitsCheckpointSerializerTest {
assertThat(incremental.afterFiles())
.extracting(DataFileMeta::fileName)
.containsExactly("after.parquet");
-
assertThat(incremental.beforeFiles().get(0).columnMaxSequenceNumbers()).isNull();
-
assertThat(incremental.afterFiles().get(0).columnMaxSequenceNumbers()).isNull();
+
assertThat(incremental.beforeFiles().get(0).writeColsSequences()).isNull();
+
assertThat(incremental.afterFiles().get(0).writeColsSequences()).isNull();
assertThat(incremental.beforeDeletionFiles())
.containsExactly(new DeletionFile("before.dv", 0L, 1L, 1L));
assertThat(incremental.afterDeletionFiles())
@@ -179,7 +179,7 @@ public class PendingSplitsCheckpointSerializerTest {
.extracting(DataFileMeta::fileName)
.containsExactly("before.parquet", "after.parquet");
assertThat(chain.dataFiles())
- .allSatisfy(file ->
assertThat(file.columnMaxSequenceNumbers()).isNull());
+ .allSatisfy(file ->
assertThat(file.writeColsSequences()).isNull());
assertThat(chain.fileBranchMapping())
.containsOnly(
org.assertj.core.data.MapEntry.entry("before.parquet",
"snapshot"),
@@ -209,7 +209,7 @@ public class PendingSplitsCheckpointSerializerTest {
}
private static DataFileMeta file(String fileName, long sequence) {
- return newFile(0).rename(fileName).withColumnMaxSequenceNumbers(new
long[] {sequence});
+ return newFile(0).rename(fileName).withWriteColsSequences(new long[]
{sequence});
}
private static void assertIncrementalSplit(
@@ -217,8 +217,8 @@ public class PendingSplitsCheckpointSerializerTest {
assertThat(sourceSplit.recordsToSkip()).isEqualTo(recordsToSkip);
assertThat(sourceSplit.split()).isInstanceOf(IncrementalSplit.class);
IncrementalSplit split = (IncrementalSplit) sourceSplit.split();
- assertColumnSequences(split.beforeFiles(), 1L);
- assertColumnSequences(split.afterFiles(), 2L);
+ assertWriteColsSequences(split.beforeFiles(), 1L);
+ assertWriteColsSequences(split.afterFiles(), 2L);
assertThat(split.beforeDeletionFiles())
.containsExactly(new DeletionFile("before.dv", 0L, 1L, 1L));
assertThat(split.afterDeletionFiles())
@@ -233,17 +233,17 @@ public class PendingSplitsCheckpointSerializerTest {
assertThat(sourceSplit.recordsToSkip()).isEqualTo(recordsToSkip);
assertThat(sourceSplit.split()).isInstanceOf(ChainSplit.class);
ChainSplit split = (ChainSplit) sourceSplit.split();
- assertColumnSequences(split.dataFiles(), 1L, 2L);
+ assertWriteColsSequences(split.dataFiles(), 1L, 2L);
assertThat(split.fileBranchMapping()).isEqualTo(branchMapping);
assertThat(split.fileBucketPathMapping()).isEqualTo(bucketPathMapping);
assertThat(split.deletionFiles())
.hasValue(Arrays.asList(null, new DeletionFile("chain.dv", 2L,
1L, 1L)));
}
- private static void assertColumnSequences(List<DataFileMeta> files,
long... sequences) {
+ private static void assertWriteColsSequences(List<DataFileMeta> files,
long... sequences) {
assertThat(files).hasSize(sequences.length);
for (int i = 0; i < sequences.length; i++) {
-
assertThat(files.get(i).columnMaxSequenceNumbers()).containsExactly(sequences[i]);
+
assertThat(files.get(i).writeColsSequences()).containsExactly(sequences[i]);
}
}
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 1d833f3d35..c5b9e22c03 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -292,6 +292,8 @@ class ManifestFileManager:
external_path=file_dict.get('_EXTERNAL_PATH'),
first_row_id=file_dict['_FIRST_ROW_ID'] if '_FIRST_ROW_ID' in
file_dict else None,
write_cols=file_dict['_WRITE_COLS'] if '_WRITE_COLS' in
file_dict else None,
+ write_cols_sequences=file_dict.get(
+ '_WRITE_COLS_SEQUENCES'),
)
if partition is None:
partition = GenericRowDeserializer.from_bytes(
@@ -432,6 +434,8 @@ class ManifestFileManager:
"_EXTERNAL_PATH": entry.file.external_path,
"_FIRST_ROW_ID": entry.file.first_row_id,
"_WRITE_COLS": entry.file.write_cols,
+ "_WRITE_COLS_SEQUENCES": (
+ entry.file.write_cols_sequences),
}
}
diff --git a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
index 5a5a70b3e7..a315c5157f 100644
--- a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
+++ b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
@@ -54,6 +54,9 @@ class DataFileMeta:
# not a schema field, just for internal usage
file_path: str = None
+ # Current DataFileMeta v9 field. Kept after the historical constructor
+ # fields so positional callers retain their existing argument mapping.
+ write_cols_sequences: Optional[List[int]] = None
def row_id_range(self) -> Optional[Range]:
if self.first_row_id is None:
@@ -108,6 +111,7 @@ class DataFileMeta:
first_row_id: Optional[int] = None,
write_cols: Optional[List[str]] = None,
file_path: Optional[str] = None,
+ write_cols_sequences: Optional[List[int]] = None,
) -> 'DataFileMeta':
if creation_time is None:
creation_time = Timestamp.now()
@@ -133,6 +137,7 @@ class DataFileMeta:
external_path=external_path,
first_row_id=first_row_id,
write_cols=write_cols,
+ write_cols_sequences=write_cols_sequences,
file_path=file_path,
)
@@ -170,6 +175,7 @@ class DataFileMeta:
external_path=self.external_path,
first_row_id=self.first_row_id,
write_cols=self.write_cols,
+ write_cols_sequences=self.write_cols_sequences,
file_path=self.file_path
)
@@ -204,6 +210,7 @@ class DataFileMeta:
external_path=self.external_path,
first_row_id=first_row_id,
write_cols=self.write_cols,
+ write_cols_sequences=self.write_cols_sequences,
file_path=self.file_path
)
@@ -230,6 +237,7 @@ class DataFileMeta:
external_path=self.external_path,
first_row_id=self.first_row_id,
write_cols=self.write_cols,
+ write_cols_sequences=self.write_cols_sequences,
file_path=self.file_path
)
@@ -266,5 +274,8 @@ DATA_FILE_META_SCHEMA = {
{"name": "_WRITE_COLS",
"type": ["null", {"type": "array", "items": "string"}],
"default": None},
+ {"name": "_WRITE_COLS_SEQUENCES",
+ "type": ["null", {"type": "array", "items": "long"}],
+ "default": None},
]
}
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 89ba3616cc..f1e4d0a870 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -32,7 +32,8 @@ from pypaimon.common.options.options_utils import OptionsUtils
from pypaimon.common.predicate import Predicate
from pypaimon.read.plan import Plan
from pypaimon.read.split import Split
-from pypaimon.read.split_serializer import deserialize_split_v1
+from pypaimon.read.split_serializer import (
+ deserialize_split_v1, serialize_split_v1)
def native_runtime_available() -> bool:
@@ -56,6 +57,25 @@ def native_reader_available() -> bool:
and native_method_available('TableRead', 'read'))
+def native_split_bridge_available() -> bool:
+ """Whether Rust accepts Java-compatible split bytes from Python plans."""
+ return native_reader_available() and native_method_available(
+ 'Split', 'deserialize')
+
+
+def native_split_from_python(split):
+ """Convert a Python-planned DataSplit/IndexedSplit for the Rust reader.
+
+ Vector scores intentionally stay on the Python IndexedSplit. Rust needs
+ only its row ranges to perform the physical read.
+ """
+ if not native_split_bridge_available():
+ return None
+ from pypaimon_rust.datafusion import Split as NativeSplit
+ return NativeSplit.deserialize(
+ serialize_split_v1(split, include_scores=False))
+
+
def native_family_search_modes_available() -> bool:
"""Whether Rust supports family-specific global-index search modes."""
return native_version_at_least(0, 4)
diff --git a/paimon-python/pypaimon/read/reader/auth_masking_reader.py
b/paimon-python/pypaimon/read/reader/auth_masking_reader.py
index aeee5e13e9..bf425581ab 100644
--- a/paimon-python/pypaimon/read/reader/auth_masking_reader.py
+++ b/paimon-python/pypaimon/read/reader/auth_masking_reader.py
@@ -153,6 +153,45 @@ class AuthFilterReader(RecordBatchReader):
self._inner.close()
+def compile_auth_masking_rules(masking_rules: Dict[str, str], read_fields:
List):
+ """Parse and validate masking transforms for batch and native readers."""
+ read_field_names = {f.name for f in read_fields}
+ parsed = {}
+ for col, transform_json in masking_rules.items():
+ if col not in read_field_names or not transform_json:
+ continue
+ transform = json.loads(transform_json)
+ if transform is not None:
+ parsed[col] = transform
+ for col_name, transform in parsed.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not
present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+ return parsed
+
+
+def apply_auth_masking(batch: pa.RecordBatch, parsed_rules: Dict[str, dict]):
+ """Apply compiled masking rules against the same original input batch."""
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = _apply_predicate_transform(
+ transform, original_batch, null_type=target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ batch = batch.set_column(
+ col_idx,
+ pa.field(original_field.name, masked_array.type, nullable=True),
+ masked_array)
+ return batch
+
+
class AuthMaskingReader(RecordBatchReader):
def __init__(self, inner_reader: RecordBatchReader, masking_rules:
Dict[str, str], read_fields: List):
@@ -160,57 +199,18 @@ class AuthMaskingReader(RecordBatchReader):
self._masking_rules = masking_rules
self._read_fields = read_fields
self._adopt_metadata(inner_reader)
- read_field_names = {f.name for f in read_fields}
- parsed = {}
- for col, tj in masking_rules.items():
- if col not in read_field_names:
- continue
- if not tj:
- continue
- transform = json.loads(tj)
- if transform is None:
- continue
- parsed[col] = transform
- self._parsed_rules = parsed
- for col_name, transform in self._parsed_rules.items():
- for ref_name in _collect_all_field_refs_from_transform(transform):
- if ref_name not in read_field_names:
- raise RuntimeError(
- f"Column masking refers to field '{ref_name}' which is
not present "
- f"in output row type. Available fields:
{read_field_names}"
- )
+ self._parsed_rules = compile_auth_masking_rules(
+ masking_rules, read_fields)
def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
batch = self._inner.read_arrow_batch()
if batch is None:
return None
- original_batch = batch
- masked_columns = {}
- for col_name, transform in self._parsed_rules.items():
- if col_name in original_batch.schema.names:
- col_idx = original_batch.schema.get_field_index(col_name)
- target_col_type = original_batch.schema.field(col_idx).type
- masked_columns[col_idx] =
self._apply_masking_transform(transform, original_batch, target_col_type)
- for col_idx, masked_array in masked_columns.items():
- original_field = original_batch.schema.field(col_idx)
- batch = batch.set_column(
- col_idx,
- pa.field(original_field.name, masked_array.type,
nullable=True),
- masked_array)
- return batch
+ return apply_auth_masking(batch, self._parsed_rules)
def close(self):
self._inner.close()
- def _apply_masking_transform(
- self,
- transform: dict,
- original_batch: pa.RecordBatch,
- target_col_type: pa.DataType,
- ) -> pa.Array:
- return _apply_predicate_transform(
- transform, original_batch, null_type=target_col_type)
-
class ColumnProjectReader(RecordBatchReader):
diff --git
a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
index 33881bafcb..11bf86c1f7 100644
--- a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
@@ -198,8 +198,11 @@ class
ChunkShuffleSplitGeneratorBase(AbstractSplitGenerator):
deletion_files_map=None,
seed: int = 0,
chunk_size: int = 0,
+ snapshot_id: Optional[int] = None,
):
- super().__init__(table, target_split_size, open_file_cost,
deletion_files_map)
+ super().__init__(
+ table, target_split_size, open_file_cost, deletion_files_map,
+ snapshot_id)
self.seed = seed
self.chunk_size = chunk_size
@@ -429,6 +432,7 @@ class
AppendChunkShuffleSplitGenerator(ChunkShuffleSplitGeneratorBase):
bucket=chunk.bucket,
raw_convertible=True,
data_deletion_files=data_deletion_files,
+ snapshot_id=self.snapshot_id,
)
exact_merged_row_count = sum(
@@ -589,6 +593,7 @@ class
DataEvolutionChunkShuffleSplitGenerator(ChunkShuffleSplitGeneratorBase):
bucket=chunk.bucket,
raw_convertible=False,
data_deletion_files=data_deletion_files,
+ snapshot_id=self.snapshot_id,
)
return IndexedSplit(
data_split,
diff --git
a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
index 9a8015347c..8da8b57a6d 100644
--- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
@@ -41,8 +41,11 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
row_ranges: Optional[List] = None,
score_getter=None,
group_stats_filter=None,
+ snapshot_id: Optional[int] = None,
):
- super().__init__(table, target_split_size, open_file_cost,
deletion_files_map)
+ super().__init__(
+ table, target_split_size, open_file_cost, deletion_files_map,
+ snapshot_id)
self.row_ranges = row_ranges
self.score_getter = score_getter
self.group_stats_filter = group_stats_filter
@@ -154,7 +157,8 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
partition=file_entries[0].partition,
bucket=file_entries[0].bucket,
raw_convertible=raw_convertible,
- data_deletion_files=data_deletion_files
+ data_deletion_files=data_deletion_files,
+ snapshot_id=self.snapshot_id,
)
splits.append(split)
return splits
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index bc89cf16d6..17599d970d 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -355,6 +355,7 @@ class FileScanner:
self._deletion_files_map(entries),
seed=seed,
chunk_size=chunk_size,
+ snapshot_id=self._scanned_snapshot_id,
)
else:
split_generator = AppendChunkShuffleSplitGenerator(
@@ -364,6 +365,7 @@ class FileScanner:
self._deletion_files_map(entries),
seed=seed,
chunk_size=chunk_size,
+ snapshot_id=self._scanned_snapshot_id,
)
elif self.table.is_primary_key_table:
entries = self.plan_files()
@@ -371,7 +373,8 @@ class FileScanner:
self.table,
self.target_split_size,
self.open_file_cost,
- self._deletion_files_map(entries)
+ self._deletion_files_map(entries),
+ snapshot_id=self._scanned_snapshot_id,
)
elif self.data_evolution:
entries, split_generator =
self._create_data_evolution_split_generator()
@@ -381,7 +384,8 @@ class FileScanner:
self.table,
self.target_split_size,
self.open_file_cost,
- self._deletion_files_map(entries)
+ self._deletion_files_map(entries),
+ snapshot_id=self._scanned_snapshot_id,
)
if not entries:
@@ -400,7 +404,6 @@ class FileScanner:
while callable(getattr(split, 'data_split', None)):
split = split.data_split()
split.is_streaming = True
- split.snapshot_id = self._scanned_snapshot_id
if self.data_evolution and self.scan_stats is not None:
# Data-evolution stats pruning happens on complete row-id groups
@@ -496,6 +499,7 @@ class FileScanner:
row_ranges,
score_getter,
None,
+ snapshot_id=self._scanned_snapshot_id,
)
# Position selection counts the complete candidate row-id space. Early
@@ -534,6 +538,7 @@ class FileScanner:
row_ranges,
score_getter,
group_stats_filter,
+ snapshot_id=self._scanned_snapshot_id,
)
def plan_files(self) -> List[ManifestEntry]:
diff --git
a/paimon-python/pypaimon/read/scanner/primary_key_table_split_generator.py
b/paimon-python/pypaimon/read/scanner/primary_key_table_split_generator.py
index 5d7d7fc9a2..06b1fc3414 100644
--- a/paimon-python/pypaimon/read/scanner/primary_key_table_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/primary_key_table_split_generator.py
@@ -36,9 +36,12 @@ class PrimaryKeyTableSplitGenerator(AbstractSplitGenerator):
table,
target_split_size: int,
open_file_cost: int,
- deletion_files_map=None
+ deletion_files_map=None,
+ snapshot_id=None,
):
- super().__init__(table, target_split_size, open_file_cost,
deletion_files_map)
+ super().__init__(
+ table, target_split_size, open_file_cost, deletion_files_map,
+ snapshot_id)
self.deletion_vectors_enabled =
table.options.deletion_vectors_enabled()
self.merge_engine = table.options.merge_engine()
diff --git a/paimon-python/pypaimon/read/scanner/split_generator.py
b/paimon-python/pypaimon/read/scanner/split_generator.py
index 4c61ebae22..100fbec2e9 100644
--- a/paimon-python/pypaimon/read/scanner/split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/split_generator.py
@@ -41,12 +41,14 @@ class AbstractSplitGenerator(ABC):
table,
target_split_size: int,
open_file_cost: int,
- deletion_files_map: Optional[Dict] = None
+ deletion_files_map: Optional[Dict] = None,
+ snapshot_id: Optional[int] = None,
):
self.table = table
self.target_split_size = target_split_size
self.open_file_cost = open_file_cost
self.deletion_files_map = deletion_files_map or {}
+ self.snapshot_id = snapshot_id
self.default_part_value = table.options.options.get(
CoreOptions.PARTITION_DEFAULT_NAME, "__DEFAULT_PARTITION__")
@@ -123,7 +125,8 @@ class AbstractSplitGenerator(ABC):
partition=file_entries[0].partition,
bucket=file_entries[0].bucket,
raw_convertible=raw_convertible,
- data_deletion_files=data_deletion_files
+ data_deletion_files=data_deletion_files,
+ snapshot_id=self.snapshot_id,
)
splits.append(split)
return splits
diff --git a/paimon-python/pypaimon/read/sliced_split.py
b/paimon-python/pypaimon/read/sliced_split.py
index e166be0451..8d69f644c6 100644
--- a/paimon-python/pypaimon/read/sliced_split.py
+++ b/paimon-python/pypaimon/read/sliced_split.py
@@ -93,6 +93,10 @@ class SlicedSplit(Split):
def is_streaming(self):
return getattr(self._data_split, 'is_streaming', False)
+ @property
+ def snapshot_id(self):
+ return self._data_split.snapshot_id
+
@property
def raw_convertible(self):
return self._data_split.raw_convertible
diff --git a/paimon-python/pypaimon/read/split.py
b/paimon-python/pypaimon/read/split.py
index a7800da72e..ccb907beb4 100644
--- a/paimon-python/pypaimon/read/split.py
+++ b/paimon-python/pypaimon/read/split.py
@@ -81,15 +81,23 @@ class DataSplit(Split):
data_deletion_files: Optional[List[DeletionFile]] = None,
snapshot_id: Optional[int] = None,
is_streaming: bool = False,
+ bucket_path: Optional[str] = None,
+ total_buckets: Optional[int] = None,
):
self._files = files
self._partition = partition
self._bucket = bucket
self.raw_convertible = raw_convertible
self.data_deletion_files = data_deletion_files
- # Scanned snapshot; None unless populated (e.g. by the native planner).
+ # Scanned snapshot, supplied by planners so this split can be
+ # serialized independently from its enclosing Plan.
self.snapshot_id = snapshot_id
self.is_streaming = is_streaming
+ # Retained when decoding SplitSerializer v1 so a cross-language
+ # decode/encode round-trip does not lose wire metadata. Python-planned
+ # splits normally leave these unset and derive paths from DataFileMeta.
+ self.bucket_path = bucket_path
+ self.total_buckets = total_buckets
@property
def files(self) -> List[DataFileMeta]:
@@ -129,6 +137,8 @@ class DataSplit(Split):
data_deletion_files=filtered_data_deletion_files,
snapshot_id=self.snapshot_id,
is_streaming=self.is_streaming,
+ bucket_path=self.bucket_path,
+ total_buckets=self.total_buckets,
)
@property
diff --git a/paimon-python/pypaimon/read/split_serializer.py
b/paimon-python/pypaimon/read/split_serializer.py
index 9a4e3a4270..c329a34e2f 100644
--- a/paimon-python/pypaimon/read/split_serializer.py
+++ b/paimon-python/pypaimon/read/split_serializer.py
@@ -15,13 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-"""Deserialize the cross-language ``SplitSerializer`` v1 binary into a pypaimon
-:class:`DataSplit`.
+"""Read and write the cross-language ``SplitSerializer`` v1 binary format.
-Mirror of the Java ``DataSplit#serialize`` (versions 8 and 9) frame wrapped in
the
-``SplitSerializer`` v1 header, as produced by ``pypaimon_rust``'s
-``Split.serialize()``. Extracts the fields the reader needs, plus per-file
-min/max keys for PK merge-on-read; key/value stats (planning-only) stay empty.
+This is the stable bridge between Java, PyPaimon and ``pypaimon_rust``. It
+supports DataSplit and IndexedSplit (including scores on the Python side) and
+the current DataSplit v9/DataFileMeta v21 layout. The reader remains compatible
+with DataSplit v8.
"""
import struct
@@ -34,7 +33,8 @@ from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.read.split import DataSplit, Split
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.binary_row import BinaryRow
-from pypaimon.table.row.generic_row import GenericRow, GenericRowDeserializer
+from pypaimon.table.row.generic_row import (
+ GenericRow, GenericRowDeserializer, GenericRowSerializer)
from pypaimon.table.source.deletion_file import DeletionFile
from pypaimon.utils.range import Range
@@ -47,6 +47,7 @@ _INDEXED_SPLIT_MAGIC = -938472394838495695
_INDEXED_SPLIT_VERSION = 1
_SPLIT_MAGIC = -2394839472490812314
_DFM_ARITY_BY_VERSION = {8: 20, 9: 21}
+_SPLIT_VERSION = 9
def _f(idx, name, dtype):
@@ -54,7 +55,6 @@ def _f(idx, name, dtype):
# DataFileMeta layout (order/types mirror DataFileMetaSerializer#toRow).
-# Fields 3/4 (min/max key) are decoded for PK tables; 5/6 (stats) stay unread.
_DFM_FIELDS: List[DataField] = [
_f(0, '_FILE_NAME', AtomicType('STRING')),
_f(1, '_FILE_SIZE', AtomicType('BIGINT')),
@@ -82,23 +82,109 @@ _DFM_FIELDS: List[DataField] = [
assert len(_DFM_FIELDS) == max(_DFM_ARITY_BY_VERSION.values())
+_SIMPLE_STATS_FIELDS: List[DataField] = [
+ _f(0, '_MIN_VALUES', AtomicType('BYTES')),
+ _f(1, '_MAX_VALUES', AtomicType('BYTES')),
+ _f(2, '_NULL_COUNTS', AtomicType('BYTES')),
+]
+
+
+def _round_to_word(size: int) -> int:
+ return (size + 7) & ~7
+
+
+def _binary_array_header(size: int) -> int:
+ return 4 + ((size + 31) // 32) * 4
+
+
+def _encode_str_array(values: List[str]) -> bytes:
+ """Encode non-null strings using Paimon's BinaryArray layout."""
+ header = _binary_array_header(len(values))
+ fixed_size = _round_to_word(header + len(values) * 8)
+ data = bytearray(fixed_size)
+ struct.pack_into('<i', data, 0, len(values))
+ for index, value in enumerate(values):
+ if not isinstance(value, str):
+ raise TypeError("BinaryArray string values must be strings")
+ encoded = value.encode('utf-8')
+ element_offset = header + index * 8
+ if len(encoded) <= 7:
+ data[element_offset:element_offset + len(encoded)] = encoded
+ data[element_offset + 7] = 0x80 | len(encoded)
+ else:
+ variable_offset = len(data)
+ data.extend(encoded)
+ data.extend(b'\x00' * (_round_to_word(len(encoded)) -
len(encoded)))
+ struct.pack_into(
+ '<Q', data, element_offset,
+ (variable_offset << 32) | len(encoded))
+ return bytes(data)
+
+
+def _encode_long_array(values: List[Optional[int]]) -> bytes:
+ """Encode nullable BIGINTs using Paimon's BinaryArray layout."""
+ header = _binary_array_header(len(values))
+ data = bytearray(_round_to_word(header + len(values) * 8))
+ struct.pack_into('<i', data, 0, len(values))
+ for index, value in enumerate(values):
+ if value is None:
+ data[4 + index // 8] |= 1 << (index % 8)
+ else:
+ struct.pack_into('<q', data, header + index * 8, value)
+ return bytes(data)
+
+
+def _decode_long_array(data: Optional[bytes]) -> Optional[List[Optional[int]]]:
+ if data is None:
+ return None
+ if len(data) < 4:
+ raise ValueError("BinaryArray<long> is shorter than its header")
+ size = struct.unpack_from('<i', data, 0)[0]
+ if size < 0:
+ raise ValueError("negative BinaryArray<long> size")
+ header = _binary_array_header(size)
+ if header + size * 8 > len(data):
+ raise ValueError("BinaryArray<long> elements exceed its buffer")
+ result = []
+ for index in range(size):
+ is_null = data[4 + index // 8] & (1 << (index % 8))
+ result.append(
+ None if is_null else struct.unpack_from(
+ '<q', data, header + index * 8)[0])
+ return result
+
def _decode_str_array(b: Optional[bytes]) -> Optional[List[str]]:
"""Decode a Paimon ``BinaryArray<string>`` (non-null elements)."""
- if not b:
- return [] if b == b'' else None
+ if b is None:
+ return None
+ if len(b) < 4:
+ raise ValueError("BinaryArray<string> is shorter than its header")
n = struct.unpack_from('<i', b, 0)[0]
- header = 4 + ((n + 31) // 32) * 4 # count + null bitset
+ if n < 0:
+ raise ValueError("negative BinaryArray<string> size")
+ if n == 0:
+ return []
+ header = _binary_array_header(n)
+ fixed_size = _round_to_word(header + n * 8)
+ if fixed_size > len(b):
+ raise ValueError("BinaryArray<string> elements exceed its buffer")
out = []
for k in range(n):
+ if b[4 + k // 8] & (1 << (k % 8)):
+ raise ValueError("array<string not null> contains a null element")
eo = header + k * 8
if b[eo + 7] & 0x80: # inline: value in first 7 bytes, len in low 7
bits of byte 7
length = b[eo + 7] & 0x7F
+ if length > 7:
+ raise ValueError("invalid inline BinaryArray<string> length")
out.append(b[eo:eo + length].decode('utf-8'))
else: # pointer: (var_off << 32) | len
- slot = struct.unpack_from('<q', b, eo)[0]
+ slot = struct.unpack_from('<Q', b, eo)[0]
var_off = (slot >> 32) & 0xFFFFFFFF
length = slot & 0xFFFFFFFF
+ if var_off < fixed_size or var_off + length > len(b):
+ raise ValueError("BinaryArray<string> value exceeds its
buffer")
out.append(b[var_off:var_off + length].decode('utf-8'))
return out
@@ -114,11 +200,18 @@ def _decode_modified_utf8(raw: bytes) -> str:
units.append(b)
i += 1
elif b & 0xE0 == 0xC0:
+ if i + 1 >= n or raw[i + 1] & 0xC0 != 0x80:
+ raise ValueError("invalid modified UTF-8 two-byte sequence")
units.append(((b & 0x1F) << 6) | (raw[i + 1] & 0x3F))
i += 2
- else: # 0xE0: 3-byte
+ elif b & 0xF0 == 0xE0:
+ if (i + 2 >= n or raw[i + 1] & 0xC0 != 0x80
+ or raw[i + 2] & 0xC0 != 0x80):
+ raise ValueError("invalid modified UTF-8 three-byte sequence")
units.append(((b & 0x0F) << 12) | ((raw[i + 1] & 0x3F) << 6) |
(raw[i + 2] & 0x3F))
i += 3
+ else:
+ raise ValueError("invalid modified UTF-8 leading byte")
out = []
j, m = 0, len(units)
while j < m:
@@ -132,41 +225,298 @@ def _decode_modified_utf8(raw: bytes) -> str:
return ''.join(chr(c) for c in out)
+def _encode_modified_utf8(value: str) -> bytes:
+ """Encode Java DataOutput.writeUTF's modified UTF-8 payload."""
+ utf16 = value.encode('utf-16-be', errors='surrogatepass')
+ result = bytearray()
+ for offset in range(0, len(utf16), 2):
+ unit = struct.unpack_from('>H', utf16, offset)[0]
+ if 0x0001 <= unit <= 0x007F:
+ result.append(unit)
+ elif unit > 0x07FF:
+ result.extend((
+ 0xE0 | (unit >> 12),
+ 0x80 | ((unit >> 6) & 0x3F),
+ 0x80 | (unit & 0x3F),
+ ))
+ else:
+ result.extend((
+ 0xC0 | (unit >> 6),
+ 0x80 | (unit & 0x3F),
+ ))
+ if len(result) > 0xFFFF:
+ raise ValueError(
+ "string too long for Java writeUTF: %d bytes (max 65535)"
+ % len(result))
+ return bytes(result)
+
+
+class _Writer:
+ def __init__(self):
+ self.parts = []
+
+ def i32(self, value: int):
+ self.parts.append(struct.pack('>i', value))
+
+ def i64(self, value: int):
+ self.parts.append(struct.pack('>q', value))
+
+ def f32(self, value: float):
+ self.parts.append(struct.pack('>f', value))
+
+ def u8(self, value: int):
+ self.parts.append(struct.pack('B', value))
+
+ def take(self, value: bytes):
+ self.parts.append(value)
+
+ def java_utf(self, value: str):
+ encoded = _encode_modified_utf8(value)
+ self.parts.append(struct.pack('>H', len(encoded)))
+ self.parts.append(encoded)
+
+ def finish(self) -> bytes:
+ return b''.join(self.parts)
+
+
+def serialize_split_v1(split: Split, include_scores: bool = True) -> bytes:
+ """Serialize a Python DataSplit/IndexedSplit as SplitSerializer v1.
+
+ ``include_scores=False`` is used by the Rust read bridge because native
+ DataSplit models row ranges but not vector scores. Scores stay attached to
+ the original Python IndexedSplit and are never needed to select rows.
+ """
+ writer = _Writer()
+ writer.i64(_SPLIT_SER_MAGIC)
+ writer.i32(_SPLIT_SER_VERSION)
+ if isinstance(split, IndexedSplit):
+ data_split = split.data_split()
+ if not isinstance(data_split, DataSplit):
+ raise TypeError(
+ "IndexedSplit must directly wrap a DataSplit for native
serialization")
+ writer.i32(_TYPE_INDEXED_SPLIT)
+ writer.i64(_INDEXED_SPLIT_MAGIC)
+ writer.i32(_INDEXED_SPLIT_VERSION)
+ _write_datasplit_body(writer, data_split)
+ ranges = split.row_ranges()
+ writer.i32(len(ranges))
+ for row_range in ranges:
+ if row_range.from_ > row_range.to:
+ raise ValueError(
+ "row range from %d exceeds to %d"
+ % (row_range.from_, row_range.to))
+ writer.i64(row_range.from_)
+ writer.i64(row_range.to)
+ scores = split.scores()
+ if include_scores and scores is not None:
+ writer.u8(1)
+ writer.i32(len(scores))
+ for score in scores:
+ writer.f32(score)
+ else:
+ writer.u8(0)
+ elif isinstance(split, DataSplit):
+ writer.i32(_TYPE_DATA_SPLIT)
+ _write_datasplit_body(writer, split)
+ else:
+ raise TypeError(
+ "SplitSerializer v1 supports DataSplit and IndexedSplit, got %s"
+ % type(split).__name__)
+ return writer.finish()
+
+
+def _serialized_row(row) -> bytes:
+ if row is None:
+ row = GenericRow([], [])
+ return GenericRowSerializer.to_bytes(row)
+
+
+def _write_datasplit_body(writer: _Writer, split: DataSplit):
+ files = list(split.files)
+ bucket_path = (
+ split.bucket_path
+ if split.bucket_path is not None else _common_bucket_path(files))
+ writer.i64(_SPLIT_MAGIC)
+ writer.i32(_SPLIT_VERSION)
+ writer.i64(-1 if split.snapshot_id is None else split.snapshot_id)
+ partition = _serialized_row(split.partition)
+ writer.i32(len(partition))
+ writer.take(partition)
+ writer.i32(split.bucket)
+ writer.java_utf(bucket_path)
+ if split.total_buckets is None:
+ writer.u8(0)
+ else:
+ writer.u8(1)
+ writer.i32(split.total_buckets)
+ writer.i32(0) # deprecated beforeFiles
+ writer.u8(0) # beforeDeletionFiles = null
+ writer.i32(len(files))
+ for data_file in files:
+ row = _serialize_data_file_meta(data_file, bucket_path)
+ writer.i32(len(row))
+ writer.take(row)
+ _write_deletion_list(writer, split.data_deletion_files, len(files))
+ writer.u8(1 if split.is_streaming else 0)
+ writer.u8(1 if split.raw_convertible else 0)
+
+
+def _common_bucket_path(files: List[DataFileMeta]) -> str:
+ """Find a shared path only when every file path is unambiguous.
+
+ When Python planning has produced files from different directories, each
+ exact path is written as DataFileMeta.external_path instead.
+ """
+ if not files:
+ return ''
+ parents = []
+ for data_file in files:
+ if data_file.external_path:
+ return ''
+ file_path = data_file.file_path
+ suffix = '/' + data_file.file_name
+ if not file_path or not file_path.endswith(suffix):
+ return ''
+ parents.append(file_path[:-len(suffix)])
+ first = parents[0]
+ return first if all(path == first for path in parents) else ''
+
+
+def _effective_external_path(data_file: DataFileMeta, bucket_path: str):
+ if data_file.external_path is not None:
+ return data_file.external_path
+ file_path = data_file.file_path
+ if file_path is None:
+ return None
+ expected = '%s/%s' % (bucket_path.rstrip('/'), data_file.file_name)
+ if bucket_path and file_path == expected:
+ return None
+ return file_path
+
+
+def _serialize_data_file_meta(data_file: DataFileMeta, bucket_path: str) ->
bytes:
+ values = [
+ data_file.file_name,
+ data_file.file_size,
+ data_file.row_count,
+ _serialized_row(data_file.min_key),
+ _serialized_row(data_file.max_key),
+ _serialize_simple_stats(data_file.key_stats),
+ _serialize_simple_stats(data_file.value_stats),
+ data_file.min_sequence_number,
+ data_file.max_sequence_number,
+ data_file.schema_id,
+ data_file.level,
+ _encode_str_array(data_file.extra_files),
+ (data_file.creation_time.get_millisecond()
+ if data_file.creation_time is not None else None),
+ data_file.delete_row_count,
+ data_file.embedded_index,
+ data_file.file_source,
+ (_encode_str_array(data_file.value_stats_cols)
+ if data_file.value_stats_cols is not None else None),
+ _effective_external_path(data_file, bucket_path),
+ data_file.first_row_id,
+ (_encode_str_array(data_file.write_cols)
+ if data_file.write_cols is not None else None),
+ (_encode_long_array(data_file.write_cols_sequences)
+ if data_file.write_cols_sequences is not None else None),
+ ]
+ serialized = GenericRowSerializer.to_bytes(GenericRow(values, _DFM_FIELDS))
+ return serialized[4:] # DataSplit carries a raw BinaryRow without arity.
+
+
+def _serialize_simple_stats(stats: Optional[SimpleStats]) -> bytes:
+ if stats is None:
+ stats = SimpleStats.empty_stats()
+ values = [
+ _serialized_row(stats.min_values),
+ _serialized_row(stats.max_values),
+ _encode_long_array(stats.null_counts or []),
+ ]
+ return GenericRowSerializer.to_bytes(
+ GenericRow(values, _SIMPLE_STATS_FIELDS))[4:]
+
+
+def _write_deletion_list(writer: _Writer, deletion_files, file_count: int):
+ if deletion_files is None:
+ writer.u8(0)
+ return
+ if len(deletion_files) != file_count:
+ raise ValueError(
+ "deletion-file count %d does not match data-file count %d"
+ % (len(deletion_files), file_count))
+ writer.u8(1)
+ writer.i32(len(deletion_files))
+ for deletion_file in deletion_files:
+ if deletion_file is None:
+ writer.u8(0)
+ continue
+ writer.u8(1)
+ writer.java_utf(deletion_file.dv_index_path)
+ writer.i64(deletion_file.offset)
+ writer.i64(deletion_file.length)
+ writer.i64(
+ -1 if deletion_file.cardinality is None
+ else deletion_file.cardinality)
+
+
class _Reader:
def __init__(self, data: bytes):
self.d = data
self.p = 0
def i32(self) -> int:
+ self._require(4)
v = struct.unpack_from('>i', self.d, self.p)[0]
self.p += 4
return v
def i64(self) -> int:
+ self._require(8)
v = struct.unpack_from('>q', self.d, self.p)[0]
self.p += 8
return v
def f32(self) -> float:
+ self._require(4)
v = struct.unpack_from('>f', self.d, self.p)[0]
self.p += 4
return v
def u8(self) -> int:
+ self._require(1)
v = self.d[self.p]
self.p += 1
return v
def take(self, n: int) -> bytes:
+ if n < 0:
+ raise ValueError("negative byte length %d" % n)
+ self._require(n)
v = self.d[self.p:self.p + n]
self.p += n
return v
def java_utf(self) -> str:
+ self._require(2)
n = struct.unpack_from('>H', self.d, self.p)[0]
self.p += 2
return _decode_modified_utf8(self.take(n))
+ def _require(self, size: int):
+ available = len(self.d) - self.p
+ if available < size:
+ raise ValueError(
+ "split buffer underrun: need %d bytes, have %d"
+ % (size, available))
+
+ def finish(self):
+ remaining = len(self.d) - self.p
+ if remaining:
+ raise ValueError(
+ "%d trailing bytes after SplitSerializer v1 frame" % remaining)
+
def deserialize_split_v1(data: bytes, partition_fields: List[DataField],
key_fields: Optional[List[DataField]] = None) ->
Split:
@@ -185,8 +535,8 @@ def deserialize_split_v1(data: bytes, partition_fields:
List[DataField],
"unsupported SplitSerializer version %d (expected %d)" % (version,
_SPLIT_SER_VERSION))
type_id = r.i32()
if type_id == _TYPE_DATA_SPLIT:
- return _read_datasplit_body(r, partition_fields, key_fields)
- if type_id == _TYPE_INDEXED_SPLIT:
+ split = _read_datasplit_body(r, partition_fields, key_fields)
+ elif type_id == _TYPE_INDEXED_SPLIT:
imagic = r.i64()
if imagic != _INDEXED_SPLIT_MAGIC:
raise ValueError("bad IndexedSplit magic %d" % imagic)
@@ -197,15 +547,31 @@ def deserialize_split_v1(data: bytes, partition_fields:
List[DataField],
data_split = _read_datasplit_body(r, partition_fields, key_fields)
# row_ranges select which global row ids to read -- must be preserved,
# else the reader scans the whole file instead of the ANN/row-id
result.
- row_ranges = [Range(r.i64(), r.i64()) for _ in range(r.i32())]
- return IndexedSplit(data_split, row_ranges, _read_scores(r))
- raise ValueError("unsupported split type id %d" % type_id)
+ range_count = r.i32()
+ if range_count < 0:
+ raise ValueError("negative IndexedSplit row-range count")
+ row_ranges = []
+ for _ in range(range_count):
+ from_, to = r.i64(), r.i64()
+ if from_ > to:
+ raise ValueError(
+ "row range from %d exceeds to %d" % (from_, to))
+ row_ranges.append(Range(from_, to))
+ scores = _read_scores(r)
+ split = IndexedSplit(data_split, row_ranges, scores)
+ else:
+ raise ValueError("unsupported split type id %d" % type_id)
+ r.finish()
+ return split
def _read_scores(r: '_Reader') -> Optional[List[float]]:
if r.u8() == 0:
return None
- return [r.f32() for _ in range(r.i32())]
+ count = r.i32()
+ if count < 0:
+ raise ValueError("negative IndexedSplit score count")
+ return [r.f32() for _ in range(count)]
def _read_datasplit_body(r: _Reader, partition_fields: List[DataField],
@@ -220,13 +586,14 @@ def _read_datasplit_body(r: _Reader, partition_fields:
List[DataField],
partition = GenericRowDeserializer.from_bytes(r.take(r.i32()),
partition_fields)
bucket = r.i32()
bucket_path = r.java_utf()
- if r.u8() == 1:
- r.i32() # total_buckets
+ total_buckets = r.i32() if r.u8() != 0 else None
if r.i32() != 0: # deprecated beforeFiles; Java
rejects non-empty
raise ValueError("cannot deserialize a split with before files")
if r.u8() != 0: # beforeDeletionFiles must be null
raise ValueError("cannot deserialize a split with before deletion
files")
file_count = r.i32()
+ if file_count < 0:
+ raise ValueError("negative DataSplit file count")
arity = _DFM_ARITY_BY_VERSION[version]
files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, arity,
key_fields)
for _ in range(file_count)]
@@ -241,6 +608,8 @@ def _read_datasplit_body(r: _Reader, partition_fields:
List[DataField],
is_streaming=is_streaming,
data_deletion_files=data_deletion_files,
snapshot_id=snapshot_id,
+ bucket_path=bucket_path,
+ total_buckets=total_buckets,
)
@@ -259,15 +628,16 @@ def _datafilemeta_from_row(row_bytes: bytes, bucket_path:
str, arity: int,
external_path = g(17)
ct = g(12)
ct = Timestamp(int(ct)) if ct is not None else None
- # min/max keys drive PK merge-on-read; stats unused, left empty.
+ key_stats = _decode_simple_stats(g(5))
+ value_stats = _decode_simple_stats(g(6))
meta = DataFileMeta(
file_name=file_name,
file_size=g(1),
row_count=g(2),
min_key=_decode_key(g(3), key_fields),
max_key=_decode_key(g(4), key_fields),
- key_stats=SimpleStats.empty_stats(),
- value_stats=SimpleStats.empty_stats(),
+ key_stats=key_stats,
+ value_stats=value_stats,
min_sequence_number=g(7),
max_sequence_number=g(8),
schema_id=g(9),
@@ -281,16 +651,40 @@ def _datafilemeta_from_row(row_bytes: bytes, bucket_path:
str, arity: int,
external_path=external_path,
first_row_id=g(18),
write_cols=_decode_str_array(g(19)),
+ write_cols_sequences=(
+ _decode_non_null_long_array(g(20)) if arity >= 21 else None),
)
meta.file_path = external_path if external_path else "%s/%s" %
(bucket_path.rstrip('/'), file_name)
return meta
+def _decode_simple_stats(data: bytes) -> SimpleStats:
+ row = BinaryRow(struct.pack('>i', 3) + data, _SIMPLE_STATS_FIELDS)
+ null_counts = _decode_long_array(row.get_field(2))
+ return SimpleStats(
+ BinaryRow(row.get_field(0), []),
+ BinaryRow(row.get_field(1), []),
+ null_counts,
+ )
+
+
+def _decode_non_null_long_array(data: Optional[bytes]) -> Optional[List[int]]:
+ values = _decode_long_array(data)
+ if values is None:
+ return None
+ if any(value is None for value in values):
+ raise ValueError("array<bigint not null> contains a null element")
+ return values
+
+
def _read_deletion_list(r: _Reader) -> Optional[List[Optional[DeletionFile]]]:
if r.u8() == 0:
return None
result: List[Optional[DeletionFile]] = []
- for _ in range(r.i32()):
+ count = r.i32()
+ if count < 0:
+ raise ValueError("negative deletion-file count")
+ for _ in range(count):
if r.u8() == 0:
result.append(None)
continue
diff --git a/paimon-python/pypaimon/read/streaming_table_scan.py
b/paimon-python/pypaimon/read/streaming_table_scan.py
index 7cdc137e01..f34eb48ada 100644
--- a/paimon-python/pypaimon/read/streaming_table_scan.py
+++ b/paimon-python/pypaimon/read/streaming_table_scan.py
@@ -376,6 +376,14 @@ class AsyncStreamingTableScan:
if (not self.table.options.native_plan_enabled()
or self._bucket_filter is not None):
return None
+ # Rust's full-snapshot scan currently has batch visibility semantics.
+ # For first-row and non-MOR deletion-vector tables it hides level-0
+ # files, while an initial streaming scan must include those files.
+ # Incremental scans already use Rust's streaming split mode and do not
+ # need this fallback.
+ if (incremental_range is None
+ and self.table.options.batch_scan_skip_level0()):
+ return None
try:
from pypaimon.read.native_plan import native_plan
plan = native_plan(
@@ -440,20 +448,21 @@ class AsyncStreamingTableScan:
self.table,
target_split_size,
open_file_cost,
- deletion_files_map={}
+ deletion_files_map={},
+ snapshot_id=snapshot_id,
)
else:
split_generator = AppendTableSplitGenerator(
self.table,
target_split_size,
open_file_cost,
- deletion_files_map={}
+ deletion_files_map={},
+ snapshot_id=snapshot_id,
)
splits = split_generator.create_splits(entries)
for split in splits:
split.is_streaming = True
- split.snapshot_id = snapshot_id
return Plan(splits, snapshot_id=snapshot_id)
def _should_use_diff_catch_up(self) -> bool:
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 5595298422..61ce079032 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -421,8 +421,13 @@ class TableRead:
return []
if self._deferred_blob_limit_may_prune(splits):
return None
+ # Query authorization has additional filtering, masking and projection
+ # semantics which are already implemented by the Python reader.
+ if any(isinstance(split, QueryAuthSplit) for split in splits):
+ return None
try:
- from pypaimon.read.native_plan import native_read
+ from pypaimon.read.native_plan import (
+ native_read, native_split_from_python)
except Exception as e:
logger.warning(
"Native read failed, falling back to the Python reader: %s", e)
@@ -430,13 +435,19 @@ class TableRead:
rust_splits = []
split_weights = []
for split in splits:
- if isinstance(split, QueryAuthSplit):
- return None
if not self._native_split_files_supported(split):
return None
rust_split = getattr(split, '_native_split', None)
if rust_split is None:
- return None
+ try:
+ rust_split = native_split_from_python(split)
+ except Exception as e:
+ logger.warning(
+ "Native split conversion failed, falling back to the "
+ "Python reader: %s", e)
+ return None
+ if rust_split is None:
+ return None
rust_splits.append(rust_split)
split_weights.append(self._native_split_weight(split))
if (parallelism is not None
@@ -713,15 +724,14 @@ class TableRead:
return True
def _convert_native_batches(self, batches, schema):
- """Apply PyPaimon's exact output limit lazily to native batches."""
remaining = self.limit
try:
for batch in batches:
if batch.num_rows == 0:
continue
+ batch = self._prepare_native_batch(batch, schema)
if remaining is not None and batch.num_rows > remaining:
batch = batch.slice(0, remaining)
- batch = self._prepare_native_batch(batch, schema)
batch = self._project_batch_to_output(batch)
yield self._try_to_pad_batch_by_schema(batch, schema)
if remaining is not None:
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 6d269957e6..8a6536bc0a 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -369,14 +369,14 @@ class TableScan:
else AppendChunkShuffleSplitGenerator)
seed, chunk_size = fs.chunk_shuffle
generator = generator_type(self.table, fs.target_split_size,
fs.open_file_cost,
- deletions, seed=seed, chunk_size=chunk_size)
+ deletions, seed=seed, chunk_size=chunk_size,
+ snapshot_id=snapshot_id)
if fs.idx_of_this_subtask is not None:
generator.with_shard(fs.idx_of_this_subtask,
fs.number_of_para_subtasks)
chunks = generator.create_splits(entries)
for split in chunks:
while callable(getattr(split, 'data_split', None)):
split = split.data_split()
- split.snapshot_id = snapshot_id
split.is_streaming = fs.is_streaming
return chunks
diff --git a/paimon-python/pypaimon/table/source/primary_key_scored_result.py
b/paimon-python/pypaimon/table/source/primary_key_scored_result.py
index a8334d2437..c0e1c2fe83 100644
--- a/paimon-python/pypaimon/table/source/primary_key_scored_result.py
+++ b/paimon-python/pypaimon/table/source/primary_key_scored_result.py
@@ -54,7 +54,8 @@ class PrimaryKeyScoredResult(GlobalIndexSplitResult):
self._positions = tuple(positions)
if len(set(_position_key(position) for position in positions)) !=
len(positions):
raise ValueError("Primary-key search result contains duplicate
physical positions.")
- self._splits = tuple(_materialize(source_splits, positions))
+ self._splits = tuple(_materialize(
+ source_splits, positions, snapshot_id))
@property
def snapshot_id(self):
@@ -69,7 +70,7 @@ class PrimaryKeyScoredResult(GlobalIndexSplitResult):
return self._splits
-def _materialize(source_splits, positions):
+def _materialize(source_splits, positions, snapshot_id):
sources = {}
for split in source_splits:
partition_bytes = _partition_bytes(split.partition)
@@ -96,7 +97,9 @@ def _materialize(source_splits, positions):
deletion_files = None
if split.data_deletion_files is not None:
deletion_files = [split.data_deletion_files[file_index]]
- single = DataSplit([data_file], split.partition, split.bucket, False,
deletion_files)
+ single = DataSplit(
+ [data_file], split.partition, split.bucket, False, deletion_files,
+ snapshot_id=snapshot_id)
ordered = sorted(scores_by_position)
yield IndexedSplit(single, _ranges(ordered),
[scores_by_position[position] for position in
ordered])
diff --git
a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
index 4d2c3fb118..88fc69fb31 100644
--- a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
@@ -161,9 +161,11 @@ class SplitOrderTest(unittest.TestCase):
self._entry('c.parquet', 2),
]
splits = DataEvolutionSplitGenerator(
- self._Table(), target_split_size=1024, open_file_cost=0
+ self._Table(), target_split_size=1024, open_file_cost=0,
+ snapshot_id=7,
).create_splits(entries)
+ self.assertEqual(7, splits[0].snapshot_id)
self.assertEqual(
['a.parquet', 'b.parquet', 'c.parquet'],
[file.file_name for file in splits[0].files],
diff --git a/paimon-python/pypaimon/tests/interval_partition_test.py
b/paimon-python/pypaimon/tests/interval_partition_test.py
index 6fbae3624d..173ed6c9ee 100644
--- a/paimon-python/pypaimon/tests/interval_partition_test.py
+++ b/paimon-python/pypaimon/tests/interval_partition_test.py
@@ -62,8 +62,10 @@ def
test_signed_zero_key_ranges_keep_versions_in_one_split(type_name):
table = SimpleNamespace(table_path='/tmp/interval-test',
options=CoreOptions(Options({})))
entries = [ManifestEntry(0, GenericRow([], []), 0, 1, file) for file in
files]
- splits = PrimaryKeyTableSplitGenerator(table, 1, 1).create_splits(entries)
+ splits = PrimaryKeyTableSplitGenerator(
+ table, 1, 1, snapshot_id=7).create_splits(entries)
assert len(splits) == 1
+ assert splits[0].snapshot_id == 7
assert sorted(file.file_name for file in splits[0].files) == ['broad',
'point']
assert not splits[0].raw_convertible
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
index bb0c03420f..de585bccb9 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
@@ -478,6 +478,7 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
value_stats_cols=None, external_path=None,
first_row_id=0,
write_cols=["id", "_ROW_ID", "_SEQUENCE_NUMBER"],
+ write_cols_sequences=[12, 50, 50],
),
)
manager.write("dirty-manifest.avro", [entry])
@@ -486,6 +487,8 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
self.assertEqual(len(entries), 1)
self.assertEqual(
entries[0].file.write_cols, ["id", "_ROW_ID", "_SEQUENCE_NUMBER"])
+ self.assertEqual(
+ entries[0].file.write_cols_sequences, [12, 50, 50])
read_stats = entries[0].file.value_stats
stats_field_names = [f.name for f in read_stats.min_values.fields]
diff --git a/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py
b/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py
index c2755aca17..ea156d8dcc 100644
--- a/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py
@@ -119,6 +119,8 @@ def _chunks(table, seed, chunk_size=3, shard=None,
predicate=None, projection=No
plan = scan.plan()
else:
plan = scan.plan()
+ assert all(
+ split.snapshot_id == plan.snapshot_id for split in plan.splits())
chunks = []
for split in plan.splits():
if
table.options.options.contains_key('incremental-between-timestamp'):
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index afc46d5608..d633338f9e 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -140,14 +140,14 @@ def
test_timestamp_windows_preserve_append_events_and_end_snapshot(
plan, rows = _read(history, native, window)
assert plan.snapshot_id == snapshot_id
assert sorted((row['k'], row['v']) for row in rows) == expected
- if native:
- assert all(split.snapshot_id == snapshot_id for split in plan.splits())
+ assert all(split.snapshot_id == snapshot_id for split in plan.splits())
def test_predicate_can_select_an_earlier_event(native, history):
predicate = history.new_read_builder().new_predicate_builder().equal('v',
'intermediate')
plan, rows = _read(history, native, (100, 200), predicate=predicate,
with_stats=True)
assert plan.snapshot_id == 3
+ assert all(split.snapshot_id == plan.snapshot_id for split in
plan.splits())
assert rows == [{'k': 1, 'v': 'intermediate'}]
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 35eeb9edbc..9581cead18 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -18,6 +18,7 @@
import datetime
import importlib.util
import io
+import json
import os
import pickle
import tempfile
@@ -28,10 +29,12 @@ import pyarrow as pa
import pytest
from pypaimon import CatalogFactory, Schema
+from pypaimon.catalog.table_query_auth import TableQueryAuthResult
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.read.native_plan import (
native_family_search_modes_available, native_method_available, native_read,
- native_reader_available,
+ native_reader_available, native_split_bridge_available,
+ native_split_from_python,
)
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.table.row.blob import BlobDescriptor, BlobRef
@@ -203,6 +206,112 @@ class NativePlanIntegrationTest(unittest.TestCase):
self.assertEqual(streamed, rows)
self.assertTrue(builder.explain().native_planned)
+ @unittest.skipUnless(native_split_bridge_available(),
+ "pypaimon-rust split bridge API not installed")
+ def test_native_read_bridges_python_planned_splits(self):
+ self.cat.create_table(
+ 'default.python_plan_native_read',
+ Schema.from_pyarrow_schema(self.schema), False)
+ self._write('python_plan_native_read', [
+ {'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}, {'k': 3, 'v': 'c'}])
+ table = self.cat.get_table('default.python_plan_native_read').copy({
+ 'read.native.enabled': 'true',
+ })
+ builder = table.new_read_builder().with_projection(['k'])
+ scan = builder.new_scan()
+ # Force the capability branch that motivates this bridge: Python owns
+ # planning, while Rust still performs the physical read.
+ with patch.object(scan, '_native_plan_supported', return_value=False):
+ plan = scan.plan()
+ self.assertTrue(plan.splits())
+ self.assertTrue(all(
+ getattr(split, '_native_split', None) is None
+ for split in plan.splits()))
+
+ with patch(
+ 'pypaimon.read.native_plan.native_split_from_python',
+ wraps=native_split_from_python) as bridge, patch(
+ 'pypaimon.read.native_plan.native_read',
+ wraps=native_read) as rust_read, patch(
+
'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader was used')):
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+
+ self.assertEqual(rows, [{'k': 1}, {'k': 2}, {'k': 3}])
+ self.assertEqual(bridge.call_count, len(plan.splits()))
+ self.assertGreaterEqual(rust_read.call_count, 1)
+
+ @unittest.skipUnless(native_split_bridge_available(),
+ "pypaimon-rust split bridge API not installed")
+ def test_native_read_bridges_python_indexed_data_evolution_split(self):
+ self.cat.create_table(
+ 'default.python_indexed_native_read',
+ Schema.from_pyarrow_schema(self.schema, options={
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ }), False)
+ self._write('python_indexed_native_read', [
+ {'k': 10, 'v': 'a'}, {'k': 20, 'v': 'b'}, {'k': 30, 'v': 'c'}])
+ table = self.cat.get_table('default.python_indexed_native_read').copy({
+ 'read.native.enabled': 'true',
+ })
+ builder = table.new_read_builder()
+ scan = builder.new_scan().with_row_ranges([Range(1, 1)])
+ with patch.object(scan, '_native_plan_supported', return_value=False):
+ plan = scan.plan()
+
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader was used')), patch(
+ 'pypaimon.read.native_plan.native_read',
+ wraps=native_read) as rust_read:
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+
+ self.assertEqual(rows, [{'k': 20, 'v': 'b'}])
+ self.assertGreaterEqual(rust_read.call_count, 1)
+
+ def test_query_auth_uses_python_plan_and_reader(self):
+ self.cat.create_table(
+ 'default.query_auth_native_read',
+ Schema.from_pyarrow_schema(self.schema), False)
+ self._write('query_auth_native_read', [
+ {'k': 1, 'v': 'sales'},
+ {'k': 2, 'v': 'eng'},
+ {'k': 3, 'v': 'eng'},
+ ])
+ table = self.cat.get_table('default.query_auth_native_read').copy({
+ 'read.native.enabled': 'true',
+ })
+ auth_filter = json.dumps({
+ 'kind': 'LEAF',
+ 'transform': {
+ 'name': 'FIELD_REF',
+ 'fieldRef': {'index': 1, 'name': 'v', 'type': 'STRING'},
+ },
+ 'function': 'EQUAL',
+ 'literals': ['eng'],
+ })
+ auth = TableQueryAuthResult(
+ [auth_filter], {'k': json.dumps({'name': 'NULL'})})
+ table.catalog_environment.table_query_auth = (
+ lambda options, identifier: lambda select: auth)
+ builder = table.new_read_builder().with_projection(['k']).with_limit(1)
+ plan = builder.new_scan().plan()
+ self.assertTrue(all(
+ split.__class__.__name__ == 'QueryAuthSplit'
+ for split in plan.splits()))
+
+ with patch(
+ 'pypaimon.read.native_plan.native_read',
+ wraps=native_read) as rust_read:
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+ streamed = builder.new_read().to_arrow_batch_reader(
+ plan.splits()).read_all().to_pylist()
+
+ self.assertEqual(rows, [{'k': None}])
+ self.assertEqual(streamed, rows)
+ rust_read.assert_not_called()
+
@unittest.skipUnless(native_reader_available(),
"pypaimon-rust native reader API not installed")
def test_native_read_uses_effective_split_parallelism(self):
@@ -1219,7 +1328,7 @@ class NativePlanIntegrationTest(unittest.TestCase):
with patch('pypaimon.read.native_plan.native_read',
wraps=native_read) as read:
rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
- read.assert_not_called()
+ self.assertEqual(read.call_count, len(plan.splits()))
self.assertEqual(sorted(rows, key=lambda row: row['k']), [
{'k': 1, 'p': 'a/b'},
{'k': 2, 'p': 'a/b'},
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index 8376a878d7..841324086e 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -21,6 +21,7 @@ from unittest.mock import Mock, patch
import pyarrow as pa
import pytest
+from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.table_read import TableRead
from pypaimon.schema.data_types import AtomicType, DataField
@@ -156,6 +157,57 @@ def
test_native_read_falls_back_for_transformed_python_split():
native.assert_not_called()
+def test_native_read_bridges_python_planned_split():
+ read = _table_read()
+ schema = pa.schema([('id', pa.int32())])
+ split = _Split()
+ converted = object()
+
+ with patch(
+ 'pypaimon.read.native_plan.native_split_from_python',
+ return_value=converted) as bridge, patch(
+ 'pypaimon.read.native_plan.native_read',
+ return_value=[_id_batch([1])]) as native:
+ result = list(read._try_native_batches([split], schema))
+
+ assert result[0].column('id').to_pylist() == [1]
+ bridge.assert_called_once_with(split)
+ native.assert_called_once_with(
+ read.table,
+ [converted],
+ predicate=None,
+ limit=None,
+ projection=['id'],
+ )
+
+
+def test_native_split_bridge_failure_falls_back_before_starting_reader():
+ read = _table_read()
+ split = _Split()
+ with patch(
+ 'pypaimon.read.native_plan.native_split_from_python',
+ side_effect=ValueError('not serializable')), patch(
+ 'pypaimon.read.native_plan.native_read') as native:
+ assert read._try_native_batches(
+ [split], pa.schema([('id', pa.int32())])) is None
+ native.assert_not_called()
+
+
+def test_native_query_auth_falls_back_to_python_reader():
+ read = _table_read()
+ split = _Split()
+ split._native_split = object()
+ wrapped = QueryAuthSplit(split, object())
+
+ with patch(
+ 'pypaimon.read.native_plan.native_split_from_python') as convert,
patch(
+ 'pypaimon.read.native_plan.native_read') as native:
+ assert read._try_native_batches(
+ [wrapped], pa.schema([('id', pa.int32())])) is None
+ convert.assert_not_called()
+ native.assert_not_called()
+
+
def test_native_read_uses_effective_parallelism_from_table_option():
read = _table_read()
read._read_parallelism = 2
diff --git a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
index 2b9329dcd2..28b67863c1 100644
--- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
+++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
@@ -182,6 +182,7 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
])
self.assertEqual(7, result.snapshot_id)
+ self.assertEqual(7, result.splits[0].snapshot_id)
self.assertEqual([(1, 2), (4, 4)],
[(r.from_, r.to) for r in
result.splits[0].row_ranges()])
self.assertEqual([0.9, 0.8, 0.7], result.splits[0].scores())
diff --git
a/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
b/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
index 7bf7c9e927..b35f120f75 100644
--- a/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
@@ -66,7 +66,9 @@ def _mock_entry(partition_values, bucket, file_name,
row_count, file_size=1024):
return entry
-def _make_generator(seed, chunk_size, table=None, deletion_files_map=None):
+def _make_generator(
+ seed, chunk_size, table=None, deletion_files_map=None,
+ snapshot_id=None):
if table is None:
table = _mock_table()
return AppendChunkShuffleSplitGenerator(
@@ -76,6 +78,7 @@ def _make_generator(seed, chunk_size, table=None,
deletion_files_map=None):
deletion_files_map=deletion_files_map,
seed=seed,
chunk_size=chunk_size,
+ snapshot_id=snapshot_id,
)
@@ -233,12 +236,13 @@ class
ChunkShuffleSplitGeneratorAlgoTest(unittest.TestCase):
def test_chunk_truncates_inside_file(self):
# one file of 250 rows, chunk_size 100 → 3 chunks: 100, 100, 50
entries = [_mock_entry([], 0, 'f1', 250)]
- gen = _make_generator(seed=1, chunk_size=100)
+ gen = _make_generator(seed=1, chunk_size=100, snapshot_id=7)
splits = gen.create_splits(entries)
self.assertEqual(len(splits), 3)
# All three chunks slice the same file → all SlicedSplit
for s in splits:
self.assertIsInstance(s, SlicedSplit)
+ self.assertEqual(s.snapshot_id, 7)
# union of (start, end) intervals must cover [0, 250)
intervals = sorted(s.shard_file_idx_map()['f1'] for s in splits)
self.assertEqual(intervals, [(0, 100), (100, 200), (200, 250)])
diff --git a/paimon-python/pypaimon/tests/split_serializer_test.py
b/paimon-python/pypaimon/tests/split_serializer_test.py
index 50561905e2..21533e857e 100644
--- a/paimon-python/pypaimon/tests/split_serializer_test.py
+++ b/paimon-python/pypaimon/tests/split_serializer_test.py
@@ -21,7 +21,8 @@ import unittest
from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.read.split_serializer import (
- _decode_modified_utf8, _decode_str_array, deserialize_split_v1)
+ _decode_modified_utf8, _decode_str_array, _encode_modified_utf8,
+ deserialize_split_v1, serialize_split_v1)
from pypaimon.schema.data_types import AtomicType, DataField
# Hand-built BinaryArray<string> == ["id", "longcolumn12"]: hits both the
inline
@@ -201,6 +202,8 @@ class SplitSerializerTest(unittest.TestCase):
self.assertEqual(file.value_stats_cols, ['field1', 'field2',
'field3'])
self.assertEqual(file.first_row_id, 12)
self.assertEqual(file.write_cols, ['a', 'b', 'c', 'f'])
+ self.assertEqual(
+ file.write_cols_sequences, [15, 100, 150, 200])
self.assertEqual(len(split.data_deletion_files), 1)
dv = split.data_deletion_files[0]
self.assertEqual(
@@ -210,6 +213,90 @@ class SplitSerializerTest(unittest.TestCase):
self.assertEqual([(r.from_, r.to) for r in
split.row_ranges()], [(13, 14)])
self.assertEqual(split.scores(), [0.75])
+ def test_v9_roundtrip_is_byte_compatible_with_java(self):
+ fields = [DataField(0, 's', AtomicType('STRING'))]
+ frame = _GOLDEN_DATA_SPLIT_V1[:16] + _GOLDEN_DATA_SPLIT_V9
+
+ split = deserialize_split_v1(frame, fields, fields)
+
+ self.assertEqual(serialize_split_v1(split), frame)
+
+ def test_v8_is_upgraded_to_v9_without_losing_reader_metadata(self):
+ fields = self._partition_fields()
+ split = deserialize_split_v1(_GOLDEN_DATA_SPLIT_V1, fields)
+
+ upgraded = deserialize_split_v1(serialize_split_v1(split), fields)
+
+ self.assertEqual(upgraded.snapshot_id, split.snapshot_id)
+ self.assertEqual(upgraded.bucket_path, split.bucket_path)
+ self.assertEqual(upgraded.total_buckets, split.total_buckets)
+ self.assertEqual(
+ [file.file_path for file in upgraded.files],
+ [file.file_path for file in split.files])
+ self.assertEqual(
+ [file.max_sequence_number for file in upgraded.files], [100, 200])
+ self.assertEqual(
+ upgraded.files[0].key_stats.null_counts,
+ split.files[0].key_stats.null_counts)
+
+ def test_indexed_roundtrip_preserves_or_strips_scores_explicitly(self):
+ fields = self._partition_fields()
+ split = deserialize_split_v1(_GOLDEN_INDEXED_SPLIT_V1, fields)
+
+ with_scores = deserialize_split_v1(
+ serialize_split_v1(split), fields)
+ without_scores = deserialize_split_v1(
+ serialize_split_v1(split, include_scores=False), fields)
+
+ self.assertEqual(with_scores.scores(), [0.5, 0.25, 0.125])
+ self.assertIsNone(without_scores.scores())
+ self.assertEqual(
+ [(r.from_, r.to) for r in without_scores.row_ranges()],
+ [(1, 4), (11, 13)])
+
+ def test_disjoint_file_paths_use_per_file_external_paths(self):
+ fields = self._partition_fields()
+ split = deserialize_split_v1(_GOLDEN_DATA_SPLIT_V1, fields)
+ split.bucket_path = None
+ split.files[0].file_path = 's3://bucket-a/data/file-a'
+ split.files[1].file_path = 's3://bucket-b/data/file-b'
+
+ restored = deserialize_split_v1(serialize_split_v1(split), fields)
+
+ self.assertEqual(restored.bucket_path, '')
+ self.assertEqual(
+ [file.external_path for file in restored.files],
+ ['s3://bucket-a/data/file-a', 's3://bucket-b/data/file-b'])
+ self.assertEqual(
+ [file.file_path for file in restored.files],
+ ['s3://bucket-a/data/file-a', 's3://bucket-b/data/file-b'])
+
+ def test_modified_utf8_roundtrip_nul_bmp_and_supplementary(self):
+ value = 'a\x00\u07ff\u0800\U0001f600'
+ self.assertEqual(
+ _decode_modified_utf8(_encode_modified_utf8(value)), value)
+
+ def test_modified_utf8_roundtrip_unpaired_surrogates(self):
+ value = '\ud800x\udc00'
+ self.assertEqual(
+ _decode_modified_utf8(_encode_modified_utf8(value)), value)
+
+ def test_modified_utf8_rejects_oversized_value(self):
+ with self.assertRaisesRegex(ValueError, 'too long'):
+ _encode_modified_utf8('\u0800' * 21846)
+
+ def test_serializer_rejects_mismatched_deletion_files(self):
+ split = deserialize_split_v1(
+ _GOLDEN_DATA_SPLIT_V1, self._partition_fields())
+ split.data_deletion_files = [None]
+ with self.assertRaisesRegex(ValueError, 'does not match'):
+ serialize_split_v1(split)
+
+ def test_deserializer_rejects_trailing_bytes(self):
+ with self.assertRaisesRegex(ValueError, 'trailing bytes'):
+ deserialize_split_v1(
+ _GOLDEN_DATA_SPLIT_V1 + b'junk', self._partition_fields())
+
def test_unsupported_data_split_version_raises(self):
for version in (7, 10):
with self.subTest(version=version):
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
index e60c7bf5b3..08d529725a 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
@@ -36,7 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class CopyFilesUtilTest {
@Test
- void testClearColumnSequencesWhenChangingSchemaId() {
+ void testClearWriteColsSequencesWhenChangingSchemaId() {
DataFileMeta source =
DataFileMeta.forAppend(
"source.parquet",
@@ -53,14 +53,14 @@ public class CopyFilesUtilTest {
null,
null,
Arrays.asList("a", "b"))
- .withColumnMaxSequenceNumbers(new long[] {2L, 3L});
+ .withWriteColsSequences(new long[] {2L, 3L});
DataFileMeta copied = CopyFilesUtil.toNewDataFileMeta(source,
"copied.parquet", 6L);
assertThat(copied.fileName()).isEqualTo("copied.parquet");
assertThat(copied.schemaId()).isEqualTo(6L);
assertThat(copied.writeCols()).containsExactly("a", "b");
- assertThat(copied.columnMaxSequenceNumbers()).isNull();
+ assertThat(copied.writeColsSequences()).isNull();
}
@Test
