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 a996b8a3c2 [core] Reuse file read targets for merged data evolution
reads (#10793)
a996b8a3c2 is described below
commit a996b8a3c2889da6527b072405aef1e3e2da0ddd
Author: sanshi <[email protected]>
AuthorDate: Sun Oct 25 10:51:25 2026 +0800
[core] Reuse file read targets for merged data evolution reads (#10793)
---
.../paimon/operation/DataEvolutionSplitRead.java | 5 +-
.../paimon/table/DataEvolutionFileIndexTest.java | 69 ++++++++++++++++++++++
2 files changed, 73 insertions(+), 1 deletion(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
index ee6f4ab8a3..953f73f935 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
@@ -571,6 +571,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
bunch,
dataFilePathFactory,
formatReaderMapping,
+ readTarget,
rowRanges,
partialReadRowType,
deletionVector,
@@ -617,6 +618,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
bunch,
dataFilePathFactory,
mapping,
+ readTarget,
rowRanges,
readRowType,
deletionVector,
@@ -684,6 +686,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FieldBunch bunch,
DataFilePathFactory dataFilePathFactory,
FormatReaderMapping formatReaderMapping,
+ FileReadTarget readTarget,
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
@@ -695,10 +698,10 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return createFileReader(
partition,
bunch.files().get(0),
- dataFilePathFactory,
formatReaderMapping,
rowRanges,
readRowType,
+ readTarget,
deletionVector,
groupSelection,
metadataCache);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
index 149ce51310..53724961a1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.append.ForceSingleBatchReader;
+import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
@@ -39,7 +40,9 @@ import
org.apache.paimon.fileindex.bitmap.ApplyBitmapIndexRecordReader;
import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
@@ -79,6 +82,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.util.ArrayList;
@@ -612,6 +616,58 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter,
null)), 50);
}
+ @Test
+ public void testMergedGroupResolvesRowSidecarOncePerFile() throws
Exception {
+ RowSidecarStatusCountingFileIO fileIO = new
RowSidecarStatusCountingFileIO();
+ catalog =
+ new FileSystemCatalog(
+ fileIO,
+ new Path(tempPath.toUri().toString(),
"counting-row-sidecar-status"));
+ catalog.createDatabase(database, true);
+
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("merged_row_sidecar_status",
options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+
+ fileIO.rowSidecarStatusReads.clear();
+ assertRow(assertSingleRow(readWithFilter(table, equalAnyF1(50))), 50);
+ assertThat(fileIO.rowSidecarStatusReads).hasSize(2);
+ assertThat(fileIO.rowSidecarStatusReads.values()).containsOnly(1);
+ }
+
+ @Test
+ public void testMissingFieldsReaderResolvesRowSidecarOnce() throws
Exception {
+ RowSidecarStatusCountingFileIO fileIO = new
RowSidecarStatusCountingFileIO();
+ catalog =
+ new FileSystemCatalog(
+ fileIO,
+ new Path(tempPath.toUri().toString(),
"counting-missing-field-status"));
+ catalog.createDatabase(database, true);
+
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("missing_field_row_sidecar_status",
options);
+ writeSplitColumns(table, ROW_COUNT, Collections.emptyMap(),
Collections.emptyMap());
+ catalog.alterTable(
+ identifier(table.name()), SchemaChange.addColumn("f3",
DataTypes.STRING()), false);
+
+ FileStoreTable latest = getTable(identifier(table.name()));
+ RowType readType = latest.rowType().project("f3");
+ ReadBuilder readBuilder =
+ latest.newReadBuilder()
+ .withReadType(readType)
+ .withRowRanges(Collections.singletonList(new
Range(50L, 50L)));
+ fileIO.rowSidecarStatusReads.clear();
+ List<InternalRow> rows =
+ collect(readBuilder.newRead(), readBuilder.newScan().plan(),
readType);
+
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).isNullAt(0)).isTrue();
+ assertThat(fileIO.rowSidecarStatusReads).hasSize(1);
+ assertThat(fileIO.rowSidecarStatusReads.values()).containsOnly(1);
+ }
+
@Test
public void testSingleFileFinalSelectionChoosesRowSidecar() throws
Exception {
Map<String, String> options = bitmapOptions("f1");
@@ -1621,6 +1677,19 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
return options;
}
+ private static class RowSidecarStatusCountingFileIO extends LocalFileIO {
+
+ private final Map<String, Integer> rowSidecarStatusReads = new
HashMap<>();
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ if (path.getName().endsWith(".row")) {
+ rowSidecarStatusReads.merge(path.getName(), 1, Integer::sum);
+ }
+ return super.getFileStatus(path);
+ }
+ }
+
@ParameterizedTest
@ValueSource(strings = {"parquet", "orc"})
public void testSubMicrosecondTimestampBsiMatchesUnindexed(String format)
throws Exception {
