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 3c0fab6af4 [core] Add row sidecars for data evolution sparse reads
(#8365)
3c0fab6af4 is described below
commit 3c0fab6af424d216fc5dbbdcea645381030a4f21
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jun 27 08:45:09 2026 +0800
[core] Add row sidecars for data evolution sparse reads (#8365)
Add opt-in row-store sidecar files for data evolution tables so sparse
row-id reads can use `.row` files while normal data files keep their
configured columnar format. The implementation covers both Java and
Python read/write paths and keeps dedicated blob/vector writers
sidecar-free.
---
docs/generated/core_configuration.html | 18 ++
.../main/java/org/apache/paimon/CoreOptions.java | 52 +++++
.../org/apache/paimon/append/AppendOnlyWriter.java | 6 +-
.../org/apache/paimon/io/RowDataFileWriter.java | 258 +++++++++++++++++++--
.../apache/paimon/io/RowDataRollingFileWriter.java | 42 ++--
.../paimon/operation/BaseAppendFileStoreWrite.java | 11 +-
.../paimon/operation/DataEvolutionSplitRead.java | 136 ++++++++++-
.../apache/paimon/append/AppendOnlyWriterTest.java | 1 +
.../DedicatedFormatRollingFileWriterTest.java | 25 ++
.../paimon/io/KeyValueFileReadWriteTest.java | 1 +
.../apache/paimon/io/RollingFileWriterTest.java | 144 ++++++++++++
.../operation/DataEvolutionSplitReadTest.java | 243 +++++++++++++++++--
.../paimon/table/DataEvolutionTableTest.java | 78 +++++++
.../pypaimon/common/options/core_options.py | 46 ++++
.../pypaimon/read/reader/format_row_reader.py | 67 ++++--
paimon-python/pypaimon/read/split_read.py | 66 +++++-
.../pypaimon/tests/data_evolution_formats_test.py | 184 +++++++++++++++
paimon-python/pypaimon/write/writer/data_writer.py | 82 +++++--
18 files changed, 1369 insertions(+), 91 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 1e82e01041..3c88e12114 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -476,6 +476,24 @@ under the License.
<td>Boolean</td>
<td>Whether enable data evolution for row tracking table.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.row-sidecar.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to generate row-store sidecar files for normal data
files on data evolution tables. The sidecar files are used to accelerate sparse
row-id reads.</td>
+ </tr>
+ <tr>
+ <td><h5>data-evolution.row-sidecar.max-selected-rows</h5></td>
+ <td style="word-wrap: break-word;">4096</td>
+ <td>Long</td>
+ <td>Maximum selected row count for reading a row-store sidecar
file. The sidecar is used only when the selected rows are no more than this
value and the selected row ratio is no more than
data-evolution.row-sidecar.max-selection-ratio.</td>
+ </tr>
+ <tr>
+ <td><h5>data-evolution.row-sidecar.max-selection-ratio</h5></td>
+ <td style="word-wrap: break-word;">0.05</td>
+ <td>Double</td>
+ <td>Maximum selected row ratio for reading a row-store sidecar
file. The value must be in (0, 1]. The sidecar is used only when the selected
row ratio is no more than this value and the selected row count is no more than
data-evolution.row-sidecar.max-selected-rows.</td>
+ </tr>
<tr>
<td><h5>data-evolution.merge-into.file-pruning</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 8ca39caf66..e3b6b01952 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2354,6 +2354,36 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription("Whether enable data evolution for row
tracking table.");
+ public static final ConfigOption<Boolean>
DATA_EVOLUTION_ROW_SIDECAR_ENABLED =
+ key("data-evolution.row-sidecar.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to generate row-store sidecar files for
normal data files "
+ + "on data evolution tables. The sidecar
files are used to "
+ + "accelerate sparse row-id reads.");
+
+ public static final ConfigOption<Long>
DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS =
+ key("data-evolution.row-sidecar.max-selected-rows")
+ .longType()
+ .defaultValue(4096L)
+ .withDescription(
+ "Maximum selected row count for reading a
row-store sidecar file. "
+ + "The sidecar is used only when the
selected rows are no more "
+ + "than this value and the selected row
ratio is no more than "
+ +
"data-evolution.row-sidecar.max-selection-ratio.");
+
+ public static final ConfigOption<Double>
DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO =
+ key("data-evolution.row-sidecar.max-selection-ratio")
+ .doubleType()
+ .defaultValue(0.05d)
+ .withDescription(
+ "Maximum selected row ratio for reading a
row-store sidecar file. "
+ + "The value must be in (0, 1]. The
sidecar is used only when "
+ + "the selected row ratio is no more than
this value and the "
+ + "selected row count is no more than "
+ +
"data-evolution.row-sidecar.max-selected-rows.");
+
public static final ConfigOption<Boolean>
DATA_EVOLUTION_MERGE_INTO_FILE_PRUNING =
key("data-evolution.merge-into.file-pruning")
.booleanType()
@@ -3928,6 +3958,28 @@ public class CoreOptions implements Serializable {
return options.get(DATA_EVOLUTION_ENABLED);
}
+ public boolean dataEvolutionRowSidecarEnabled() {
+ return options.get(DATA_EVOLUTION_ROW_SIDECAR_ENABLED);
+ }
+
+ public long dataEvolutionRowSidecarMaxSelectedRows() {
+ long maxSelectedRows =
options.get(DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS);
+ checkArgument(
+ maxSelectedRows > 0,
+ "The option %s must be greater than 0.",
+ DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.key());
+ return maxSelectedRows;
+ }
+
+ public double dataEvolutionRowSidecarMaxSelectionRatio() {
+ double maxSelectionRatio =
options.get(DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO);
+ checkArgument(
+ maxSelectionRatio > 0 && maxSelectionRatio <= 1,
+ "The option %s must be in (0, 1].",
+ DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.key());
+ return maxSelectionRatio;
+ }
+
public boolean dataEvolutionMergeIntoFilePruning() {
return options.get(DATA_EVOLUTION_MERGE_INTO_FILE_PRUNING);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
index 6f081bf2a0..66dcc75613 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
@@ -85,6 +85,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
private final boolean forceCompact;
private final boolean asyncFileWrite;
private final boolean statsDenseStore;
+ @Nullable private final FileFormat rowSidecarFileFormat;
@Nullable private final BlobFileContext blobContext;
private final List<DataFileMeta> newFiles;
private final List<DataFileMeta> deletedFiles;
@@ -129,6 +130,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
boolean asyncFileWrite,
boolean statsDenseStore,
boolean dataEvolutionEnabled,
+ @Nullable FileFormat rowSidecarFileFormat,
@Nullable BlobFileContext blobContext) {
this.fileIO = fileIO;
this.schemaId = schemaId;
@@ -145,6 +147,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
this.forceCompact = forceCompact;
this.asyncFileWrite = asyncFileWrite;
this.statsDenseStore = statsDenseStore;
+ this.rowSidecarFileFormat = dataEvolutionEnabled ?
rowSidecarFileFormat : null;
this.blobContext = blobContext;
this.newFiles = new ArrayList<>();
this.deletedFiles = new ArrayList<>();
@@ -344,7 +347,8 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
FileSource.APPEND,
asyncFileWrite,
statsDenseStore,
- writeCols);
+ writeCols,
+ rowSidecarFileFormat);
}
private void trySyncLatestCompaction(boolean blocking)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
index aa3b203588..02dff0fd82 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
@@ -20,6 +20,8 @@ package org.apache.paimon.io;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.FileSource;
@@ -32,8 +34,10 @@ import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -48,7 +52,7 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
private final long schemaId;
private final boolean isExternalPath;
private final SimpleStatsConverter statsArraySerializer;
- @Nullable private final DataFileIndexWriter dataFileIndexWriter;
+ private final List<DataFileAuxiliaryWriter> auxiliaryFileWriters;
private final FileSource fileSource;
@Nullable private final List<String> writeCols;
private final RowDataFileSequenceNumberTracker sequenceNumberTracker;
@@ -66,13 +70,64 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
boolean statsDenseStore,
boolean isExternalPath,
@Nullable List<String> writeCols) {
+ this(
+ fileIO,
+ context,
+ path,
+ writeSchema,
+ schemaId,
+ seqNumCounterSupplier,
+ fileIndexOptions,
+ fileSource,
+ asyncFileWrite,
+ statsDenseStore,
+ isExternalPath,
+ writeCols,
+ null,
+ null);
+ }
+
+ public RowDataFileWriter(
+ FileIO fileIO,
+ FileWriterContext context,
+ Path path,
+ RowType writeSchema,
+ long schemaId,
+ Supplier<LongCounter> seqNumCounterSupplier,
+ FileIndexOptions fileIndexOptions,
+ FileSource fileSource,
+ boolean asyncFileWrite,
+ boolean statsDenseStore,
+ boolean isExternalPath,
+ @Nullable List<String> writeCols,
+ @Nullable FileFormat rowSidecarFormat,
+ @Nullable Path rowSidecarPath) {
super(fileIO, context, path, Function.identity(), writeSchema,
asyncFileWrite);
+ if ((rowSidecarFormat == null) != (rowSidecarPath == null)) {
+ throw new IllegalArgumentException(
+ "Row sidecar format and path should be both null or both
non-null.");
+ }
this.schemaId = schemaId;
this.isExternalPath = isExternalPath;
this.statsArraySerializer = new SimpleStatsConverter(writeSchema,
statsDenseStore);
- this.dataFileIndexWriter =
- DataFileIndexWriter.create(
- fileIO, dataFileToFileIndexPath(path), writeSchema,
fileIndexOptions);
+ List<DataFileAuxiliaryWriter> auxiliaryFileWriters = new ArrayList<>();
+ Path fileIndexPath = dataFileToFileIndexPath(path);
+ DataFileIndexWriter dataFileIndexWriter =
+ DataFileIndexWriter.create(fileIO, fileIndexPath, writeSchema,
fileIndexOptions);
+ if (dataFileIndexWriter != null) {
+ auxiliaryFileWriters.add(
+ new DataFileIndexAuxiliaryWriter(dataFileIndexWriter,
fileIO, fileIndexPath));
+ }
+ if (rowSidecarFormat != null) {
+ auxiliaryFileWriters.add(
+ new RowSidecarAuxiliaryWriter(
+ fileIO,
+ rowSidecarFormat.createWriterFactory(writeSchema),
+ rowSidecarPath,
+ context.compression(),
+ asyncFileWrite));
+ }
+ this.auxiliaryFileWriters =
Collections.unmodifiableList(auxiliaryFileWriters);
this.fileSource = fileSource;
this.writeCols = writeCols;
this.sequenceNumberTracker =
@@ -83,30 +138,73 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
@Override
public void write(InternalRow row) throws IOException {
super.write(row);
- // add row to index if needed
- if (dataFileIndexWriter != null) {
- dataFileIndexWriter.write(row);
+ for (DataFileAuxiliaryWriter auxiliaryFileWriter :
auxiliaryFileWriters) {
+ auxiliaryFileWriter.write(row);
}
sequenceNumberTracker.update(row);
}
+ @Override
+ public void writeBundle(BundleRecords bundle) throws IOException {
+ for (InternalRow row : bundle) {
+ write(row);
+ }
+ }
+
@Override
public void close() throws IOException {
- if (dataFileIndexWriter != null) {
- dataFileIndexWriter.close();
+ for (DataFileAuxiliaryWriter auxiliaryFileWriter :
auxiliaryFileWriters) {
+ auxiliaryFileWriter.close();
}
super.close();
}
+ @Override
+ public void abort() {
+ for (DataFileAuxiliaryWriter auxiliaryFileWriter :
auxiliaryFileWriters) {
+ auxiliaryFileWriter.abort();
+ }
+ super.abort();
+ }
+
+ @Override
+ public Optional<FileWriterAbortExecutor> abortExecutor() {
+ Optional<FileWriterAbortExecutor> mainAbortExecutor =
super.abortExecutor();
+ if (auxiliaryFileWriters.isEmpty()) {
+ return mainAbortExecutor;
+ }
+
+ List<FileWriterAbortExecutor> abortExecutors = new ArrayList<>();
+ mainAbortExecutor.ifPresent(abortExecutors::add);
+ for (DataFileAuxiliaryWriter auxiliaryFileWriter :
auxiliaryFileWriters) {
+ auxiliaryFileWriter.abortExecutor().ifPresent(abortExecutors::add);
+ }
+ if (abortExecutors.isEmpty()) {
+ return Optional.empty();
+ }
+ if (abortExecutors.size() == 1) {
+ return Optional.of(abortExecutors.get(0));
+ }
+ return Optional.of(new CompoundFileWriterAbortExecutor(fileIO, path,
abortExecutors));
+ }
+
@Override
public DataFileMeta result() throws IOException {
long fileSize = outputBytes();
Pair<List<String>, SimpleStats> statsPair =
statsArraySerializer.toBinary(fieldStats(fileSize));
- DataFileIndexWriter.FileIndexResult indexResult =
- dataFileIndexWriter == null
- ? DataFileIndexWriter.EMPTY_RESULT
- : dataFileIndexWriter.result();
+ List<String> extraFiles = new ArrayList<>();
+ byte[] embeddedIndex = null;
+ for (DataFileAuxiliaryWriter auxiliaryFileWriter :
auxiliaryFileWriters) {
+ DataFileAuxiliaryResult auxiliaryResult =
auxiliaryFileWriter.result();
+ extraFiles.addAll(auxiliaryResult.extraFiles());
+ if (auxiliaryResult.embeddedIndexBytes() != null) {
+ if (embeddedIndex != null) {
+ throw new IOException("Found more than one embedded index
for one data file.");
+ }
+ embeddedIndex = auxiliaryResult.embeddedIndexBytes();
+ }
+ }
String externalPath = isExternalPath ? path.toString() : null;
return DataFileMeta.forAppend(
path.getName(),
@@ -116,14 +214,140 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
sequenceNumberTracker.min(),
sequenceNumberTracker.max(),
schemaId,
- indexResult.independentIndexFile() == null
- ? Collections.emptyList()
- :
Collections.singletonList(indexResult.independentIndexFile()),
- indexResult.embeddedIndexBytes(),
+ extraFiles.isEmpty() ? Collections.emptyList() : extraFiles,
+ embeddedIndex,
fileSource,
statsPair.getKey(),
externalPath,
null,
writeCols);
}
+
+ private interface DataFileAuxiliaryWriter {
+
+ void write(InternalRow row) throws IOException;
+
+ void close() throws IOException;
+
+ void abort();
+
+ Optional<FileWriterAbortExecutor> abortExecutor();
+
+ DataFileAuxiliaryResult result() throws IOException;
+ }
+
+ private static class DataFileAuxiliaryResult {
+
+ private static final DataFileAuxiliaryResult EMPTY =
+ new DataFileAuxiliaryResult(null, Collections.emptyList());
+
+ @Nullable private final byte[] embeddedIndexBytes;
+ private final List<String> extraFiles;
+
+ private DataFileAuxiliaryResult(
+ @Nullable byte[] embeddedIndexBytes, List<String> extraFiles) {
+ this.embeddedIndexBytes = embeddedIndexBytes;
+ this.extraFiles = extraFiles;
+ }
+
+ @Nullable
+ private byte[] embeddedIndexBytes() {
+ return embeddedIndexBytes;
+ }
+
+ private List<String> extraFiles() {
+ return extraFiles;
+ }
+
+ private static DataFileAuxiliaryResult embeddedIndex(byte[]
embeddedIndexBytes) {
+ return new DataFileAuxiliaryResult(embeddedIndexBytes,
Collections.emptyList());
+ }
+
+ private static DataFileAuxiliaryResult extraFile(String extraFile) {
+ return new DataFileAuxiliaryResult(null,
Collections.singletonList(extraFile));
+ }
+ }
+
+ private static class DataFileIndexAuxiliaryWriter implements
DataFileAuxiliaryWriter {
+
+ private final DataFileIndexWriter writer;
+ private final FileIO fileIO;
+ private final Path indexPath;
+
+ private DataFileIndexAuxiliaryWriter(
+ DataFileIndexWriter writer, FileIO fileIO, Path indexPath) {
+ this.writer = writer;
+ this.fileIO = fileIO;
+ this.indexPath = indexPath;
+ }
+
+ @Override
+ public void write(InternalRow row) {
+ writer.write(row);
+ }
+
+ @Override
+ public void close() throws IOException {
+ writer.close();
+ }
+
+ @Override
+ public void abort() {
+ fileIO.deleteQuietly(indexPath);
+ }
+
+ @Override
+ public Optional<FileWriterAbortExecutor> abortExecutor() {
+ return Optional.of(new FileWriterAbortExecutor(fileIO, indexPath));
+ }
+
+ @Override
+ public DataFileAuxiliaryResult result() {
+ DataFileIndexWriter.FileIndexResult result = writer.result();
+ if (result.independentIndexFile() != null) {
+ return
DataFileAuxiliaryResult.extraFile(result.independentIndexFile());
+ }
+ if (result.embeddedIndexBytes() != null) {
+ return
DataFileAuxiliaryResult.embeddedIndex(result.embeddedIndexBytes());
+ }
+ return DataFileAuxiliaryResult.EMPTY;
+ }
+ }
+
+ private static class RowSidecarAuxiliaryWriter
+ extends SingleFileWriter<InternalRow, DataFileAuxiliaryResult>
+ implements DataFileAuxiliaryWriter {
+
+ private RowSidecarAuxiliaryWriter(
+ FileIO fileIO,
+ FormatWriterFactory factory,
+ Path path,
+ String compression,
+ boolean asyncWrite) {
+ super(fileIO, factory, path, Function.identity(), compression,
asyncWrite);
+ }
+
+ @Override
+ public DataFileAuxiliaryResult result() {
+ return DataFileAuxiliaryResult.extraFile(path.getName());
+ }
+ }
+
+ private static class CompoundFileWriterAbortExecutor extends
FileWriterAbortExecutor {
+
+ private final List<FileWriterAbortExecutor> abortExecutors;
+
+ private CompoundFileWriterAbortExecutor(
+ FileIO fileIO, Path path, List<FileWriterAbortExecutor>
abortExecutors) {
+ super(fileIO, path);
+ this.abortExecutors = abortExecutors;
+ }
+
+ @Override
+ public void abort() {
+ for (FileWriterAbortExecutor abortExecutor : abortExecutors) {
+ abortExecutor.abort();
+ }
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
index 81e0a0aefc..0da1badf62 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.types.RowType;
@@ -49,23 +50,32 @@ public class RowDataRollingFileWriter extends
RollingFileWriterImpl<InternalRow,
FileSource fileSource,
boolean asyncFileWrite,
boolean statsDenseStore,
- @Nullable List<String> writeCols) {
+ @Nullable List<String> writeCols,
+ @Nullable FileFormat rowSidecarFormat) {
super(
- () ->
- new RowDataFileWriter(
- fileIO,
- RollingFileWriter.createFileWriterContext(
- fileFormat, writeSchema,
statsCollectors, fileCompression),
- pathFactory.newPath(),
- writeSchema,
- schemaId,
- seqNumCounterSupplier,
- fileIndexOptions,
- fileSource,
- asyncFileWrite,
- statsDenseStore,
- pathFactory.isExternalPath(),
- writeCols),
+ () -> {
+ Path dataPath = pathFactory.newPath();
+ Path rowSidecarPath =
+ rowSidecarFormat == null
+ ? null
+ : new Path(dataPath.getParent(),
dataPath.getName() + ".row");
+ return new RowDataFileWriter(
+ fileIO,
+ RollingFileWriter.createFileWriterContext(
+ fileFormat, writeSchema, statsCollectors,
fileCompression),
+ dataPath,
+ writeSchema,
+ schemaId,
+ seqNumCounterSupplier,
+ fileIndexOptions,
+ fileSource,
+ asyncFileWrite,
+ statsDenseStore,
+ pathFactory.isExternalPath(),
+ writeCols,
+ rowSidecarFormat,
+ rowSidecarPath);
+ },
targetFileSize);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index b40e8ee2bc..574143192e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -155,6 +155,7 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
options.asyncFileWrite(),
options.statsDenseStore(),
options.dataEvolutionEnabled(),
+ rowSidecarFileFormat(),
blobContext);
}
@@ -278,7 +279,15 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
FileSource.COMPACT,
options.asyncFileWrite(),
options.statsDenseStore(),
- rowType.equals(writeType) ? null : writeType.getFieldNames());
+ rowType.equals(writeType) ? null : writeType.getFieldNames(),
+ rowSidecarFileFormat());
+ }
+
+ @Nullable
+ private FileFormat rowSidecarFileFormat() {
+ return options.dataEvolutionEnabled() &&
options.dataEvolutionRowSidecarEnabled()
+ ? FileFormat.fromIdentifier("row", options.toConfiguration())
+ : null;
}
private RecordReaderIterator<InternalRow> createFilesIterator(
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 145fdc9ad4..31ff6cd614 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
@@ -28,6 +28,7 @@ import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatKey;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.IndexedSplitRecordReader;
import org.apache.paimon.io.DataFileMeta;
@@ -89,6 +90,8 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
*/
public class DataEvolutionSplitRead implements SplitRead<InternalRow> {
+ private static final String ROW_SIDECAR_FORMAT = "row";
+
private final FileIO fileIO;
private final TableSchema schema;
private final FileFormatDiscover formatDiscover;
@@ -261,7 +264,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
for (int i = 0; i < fieldsFiles.size(); i++) {
FieldBunch bunch = fieldsFiles.get(i);
DataFileMeta firstFile = bunch.files().get(0);
- String formatIdentifier =
DataFilePathFactory.formatIdentifier(firstFile.fileName());
+ FileReadTarget readTarget = readTarget(firstFile,
dataFilePathFactory, rowRanges);
+ String formatIdentifier = readTarget.formatIdentifier;
long schemaId = firstFile.schemaId();
TableSchema dataSchema =
schemaFetcher.apply(schemaId).project(firstFile.writeCols());
int[] fieldIds =
@@ -434,7 +438,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType)
throws IOException {
- String formatIdentifier =
DataFilePathFactory.formatIdentifier(file.fileName());
+ FileReadTarget readTarget = readTarget(file, dataFilePathFactory,
rowRanges);
+ String formatIdentifier = readTarget.formatIdentifier;
long schemaId = file.schemaId();
FormatReaderMapping formatReaderMapping =
formatReaderMappings.computeIfAbsent(
@@ -447,7 +452,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
? schema
:
schemaFetcher.apply(schemaId)));
return createFileReader(
- partition, file, dataFilePathFactory, formatReaderMapping,
rowRanges, readRowType);
+ partition, file, formatReaderMapping, rowRanges, readRowType,
readTarget);
}
private FileRecordReader<InternalRow> createFileReader(
@@ -458,10 +463,26 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType)
throws IOException {
+ return createFileReader(
+ partition,
+ file,
+ formatReaderMapping,
+ rowRanges,
+ readRowType,
+ readTarget(file, dataFilePathFactory, rowRanges));
+ }
+
+ private FileRecordReader<InternalRow> createFileReader(
+ BinaryRow partition,
+ DataFileMeta file,
+ FormatReaderMapping formatReaderMapping,
+ List<Range> rowRanges,
+ RowType readRowType,
+ FileReadTarget readTarget)
+ throws IOException {
RoaringBitmap32 selection = file.toFileSelection(rowRanges);
FormatReaderContext formatReaderContext =
- new FormatReaderContext(
- fileIO, dataFilePathFactory.toPath(file),
file.fileSize(), selection);
+ new FormatReaderContext(fileIO, readTarget.path,
readTarget.fileSize, selection);
return new DataFileRecordReader(
readRowType,
formatReaderMapping.getReaderFactory(),
@@ -477,6 +498,111 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
formatReaderMapping.getSystemFields());
}
+ private FileReadTarget readTarget(
+ DataFileMeta file, DataFilePathFactory dataFilePathFactory,
List<Range> rowRanges)
+ throws IOException {
+ String rowSidecar = rowSidecarFileName(file);
+ if (rowSidecar != null
+ && shouldReadRowSidecar(
+ file,
+ rowRanges,
+ coreOptions.dataEvolutionRowSidecarMaxSelectedRows(),
+
coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) {
+ Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file);
+ try {
+ return new FileReadTarget(
+ ROW_SIDECAR_FORMAT, rowPath,
fileIO.getFileStatus(rowPath).getLen());
+ } catch (IOException e) {
+ if (!coreOptions.scanIgnoreLostFile()) {
+ throw e;
+ }
+ }
+ }
+ return new FileReadTarget(
+ DataFilePathFactory.formatIdentifier(file.fileName()),
+ dataFilePathFactory.toPath(file),
+ file.fileSize());
+ }
+
+ @Nullable
+ @VisibleForTesting
+ static String rowSidecarFileName(DataFileMeta file) {
+ List<String> rowFiles =
+ file.extraFiles().stream()
+ .filter(DataEvolutionSplitRead::isRowSidecarFile)
+ .collect(Collectors.toList());
+ return rowFiles.size() == 1 ? rowFiles.get(0) : null;
+ }
+
+ @VisibleForTesting
+ static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable
List<Range> rowRanges) {
+ return shouldReadRowSidecar(
+ file,
+ rowRanges,
+
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(),
+
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue());
+ }
+
+ @VisibleForTesting
+ static boolean shouldReadRowSidecar(
+ DataFileMeta file,
+ @Nullable List<Range> rowRanges,
+ long maxSelectedRows,
+ double maxSelectionRatio) {
+ if (rowRanges == null
+ || rowRanges.isEmpty()
+ || file.rowCount() <= 0
+ || isBlobFile(file.fileName())
+ || isVectorStoreFile(file.fileName())
+ || rowSidecarFileName(file) == null) {
+ return false;
+ }
+
+ long selectedRowCount = selectedRowCount(file, rowRanges);
+ if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) {
+ return false;
+ }
+
+ double selectionRatio = (double) selectedRowCount / file.rowCount();
+ return selectedRowCount <= maxSelectedRows && selectionRatio <=
maxSelectionRatio;
+ }
+
+ @VisibleForTesting
+ static long selectedRowCount(DataFileMeta file, List<Range> rowRanges) {
+ Range fileRange = file.nonNullRowIdRange();
+ List<Range> intersections = new ArrayList<>();
+ for (Range range : rowRanges) {
+ Range intersection = Range.intersection(fileRange, range);
+ if (intersection != null) {
+ intersections.add(intersection);
+ }
+ }
+ return Range.sortAndMergeOverlap(intersections, true).stream()
+ .mapToLong(Range::count)
+ .sum();
+ }
+
+ private static boolean isRowSidecarFile(String fileName) {
+ try {
+ return
ROW_SIDECAR_FORMAT.equals(DataFilePathFactory.formatIdentifier(fileName));
+ } catch (RuntimeException e) {
+ return false;
+ }
+ }
+
+ private static class FileReadTarget {
+
+ private final String formatIdentifier;
+ private final Path path;
+ private final long fileSize;
+
+ private FileReadTarget(String formatIdentifier, Path path, long
fileSize) {
+ this.formatIdentifier = formatIdentifier;
+ this.path = path;
+ this.fileSize = fileSize;
+ }
+ }
+
@VisibleForTesting
public static List<FieldBunch> splitFieldBunches(
List<DataFileMeta> needMergeFiles, Function<DataFileMeta, RowType>
fileToRowType) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
index c861296f2b..55965234bb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
@@ -827,6 +827,7 @@ public class AppendOnlyWriterTest {
true,
false,
options.dataEvolutionEnabled(),
+ null,
BlobFileContext.create(writeSchema, options));
writer.setMemoryPool(
new HeapMemorySegmentPool(options.writeBufferSize(),
options.pageSize()));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
index ffd6baedb6..0bff7e43b5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
@@ -147,6 +147,31 @@ public class DedicatedFormatRollingFileWriterTest {
metasResult.subList(1,
4).stream().mapToLong(DataFileMeta::rowCount).sum());
}
+ @Test
+ public void testDoesNotWriteRowSidecar() throws IOException {
+ // Tests that: dedicated blob files do not create row-store sidecars.
+ // Kills mutation: adding row sidecar writing to
DedicatedFormatRollingFileWriter.
+ writer.write(GenericRow.of(1, BinaryString.fromString("test"), new
BlobData(testBlobData)));
+ writer.close();
+
+ List<DataFileMeta> metasResult = writer.result();
+ assertThat(metasResult).hasSize(2);
+ assertThat(metasResult).anyMatch(file ->
"parquet".equals(file.fileFormat()));
+ assertThat(metasResult).anyMatch(file ->
"blob".equals(file.fileFormat()));
+ assertThat(metasResult)
+ .allSatisfy(
+ file ->
+ assertThat(file.extraFiles())
+ .noneMatch(extraFile ->
extraFile.endsWith(".row")));
+ try (Stream<java.nio.file.Path> files = Files.walk(tempDir)) {
+ assertThat(
+ files.filter(Files::isRegularFile)
+ .noneMatch(
+ file ->
file.getFileName().toString().endsWith(".row")))
+ .isTrue();
+ }
+ }
+
@Test
public void testBundleWritingPreservesMainFileIndexSideEffects() throws
IOException {
Options options = new Options();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
index 0169a6d5e3..d295ed575e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
@@ -295,6 +295,7 @@ public class KeyValueFileReadWriteTest {
true,
false,
options.dataEvolutionEnabled(),
+ null,
BlobFileContext.create(schema, options));
appendOnlyWriter.setMemoryPool(
new HeapMemorySegmentPool(options.writeBufferSize(),
options.pageSize()));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
index 84e7c6766b..ec334007e5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
@@ -23,10 +23,12 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.IntType;
@@ -41,6 +43,11 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
@@ -122,6 +129,119 @@ public class RollingFileWriterTest {
}
}
+ @Test
+ public void testWriteRowSidecar() throws IOException {
+ FileFormat fileFormat = FileFormat.fromIdentifier("parquet", new
Options());
+ FileFormat rowFormat = FileFormat.fromIdentifier("row", new Options());
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ new Path(tempDir + "/bucket-0"),
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null);
+ RowDataRollingFileWriter writer =
+ new RowDataRollingFileWriter(
+ LocalFileIO.create(),
+ 0L,
+ fileFormat,
+ TARGET_FILE_SIZE,
+ SCHEMA,
+ pathFactory,
+ () -> new LongCounter(0),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+
SimpleColStatsCollector.createFullStatsFactories(SCHEMA.getFieldCount()),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ true,
+ false,
+ null,
+ rowFormat);
+
+ writer.write(GenericRow.of(1));
+ writer.close();
+
+ List<DataFileMeta> files = writer.result();
+ assertThat(files).hasSize(1);
+ DataFileMeta file = files.get(0);
+ assertThat(file.extraFiles()).hasSize(1);
+ String rowSidecar = file.extraFiles().get(0);
+ assertThat(rowSidecar).endsWith(".row");
+ Path dataPath = pathFactory.toPath(file);
+ Path rowSidecarPath = pathFactory.toAlignedPath(rowSidecar, file);
+ assertThat(LocalFileIO.create().exists(dataPath)).isTrue();
+ assertThat(LocalFileIO.create().exists(rowSidecarPath)).isTrue();
+
+ writer.abort();
+ assertThat(LocalFileIO.create().exists(dataPath)).isFalse();
+ assertThat(LocalFileIO.create().exists(rowSidecarPath)).isFalse();
+ }
+
+ @Test
+ public void testWriteRowSidecarWithBundle() throws IOException {
+ FileFormat fileFormat = FileFormat.fromIdentifier("parquet", new
Options());
+ FileFormat rowFormat = FileFormat.fromIdentifier("row", new Options());
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ new Path(tempDir + "/bucket-0"),
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null);
+ RowDataRollingFileWriter writer =
+ new RowDataRollingFileWriter(
+ LocalFileIO.create(),
+ 0L,
+ fileFormat,
+ TARGET_FILE_SIZE,
+ SCHEMA,
+ pathFactory,
+ () -> new LongCounter(0),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+
SimpleColStatsCollector.createFullStatsFactories(SCHEMA.getFieldCount()),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ true,
+ false,
+ null,
+ rowFormat);
+
+ writer.writeBundle(
+ new SingleUseBundleRecords(Arrays.asList(GenericRow.of(1),
GenericRow.of(2))));
+ writer.close();
+
+ List<DataFileMeta> files = writer.result();
+ assertThat(files).hasSize(1);
+ DataFileMeta file = files.get(0);
+ assertThat(file.rowCount()).isEqualTo(2);
+ assertThat(file.minSequenceNumber()).isEqualTo(0);
+ assertThat(file.maxSequenceNumber()).isEqualTo(1);
+ assertThat(file.extraFiles()).hasSize(1);
+ assertThat(
+ readIntsFromRowFile(
+ rowFormat,
+
pathFactory.toAlignedPath(file.extraFiles().get(0), file)))
+ .containsExactly(1, 2);
+ }
+
+ private static List<Integer> readIntsFromRowFile(FileFormat rowFormat,
Path path)
+ throws IOException {
+ LocalFileIO fileIO = LocalFileIO.create();
+ List<Integer> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ rowFormat
+ .createReaderFactory(SCHEMA, SCHEMA,
Collections.emptyList())
+ .createReader(
+ new FormatReaderContext(fileIO, path,
fileIO.getFileSize(path)))) {
+ reader.forEachRemaining(row -> result.add(row.getInt(0)));
+ }
+ return result;
+ }
+
private void assertFileNum(int expected) {
File dataDir = tempDir.resolve("bucket-0").toFile();
File[] files = dataDir.listFiles();
@@ -139,4 +259,28 @@ public class RollingFileWriterTest {
assertThat(file.valueStatsCols()).isNull();
assertThat(file.valueStats().minValues().getFieldCount()).isEqualTo(SCHEMA.getFieldCount());
}
+
+ private static class SingleUseBundleRecords implements BundleRecords {
+
+ private final List<InternalRow> rows;
+ private boolean iterated;
+
+ private SingleUseBundleRecords(List<InternalRow> rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ public Iterator<InternalRow> iterator() {
+ if (iterated) {
+ throw new IllegalStateException("Bundle should only be
consumed once.");
+ }
+ iterated = true;
+ return rows.iterator();
+ }
+
+ @Override
+ public long rowCount() {
+ return rows.size();
+ }
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
index 95bae45a7d..89316c249b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
@@ -18,11 +18,34 @@
package org.apache.paimon.operation;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -30,9 +53,14 @@ import java.util.List;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class DataEvolutionSplitReadTest {
+ @TempDir java.nio.file.Path tempDir;
+
@Test
public void testDifferentRowIdRange() {
DataFileMeta file1 = createFile("file1.parquet", 1L, 100, 10);
@@ -133,25 +161,206 @@ class DataEvolutionSplitReadTest {
assertEquals(Arrays.asList(file4, file5, file6), result.get(1));
}
+ @Test
+ public void testRowSidecarFileName() {
+ DataFileMeta file =
+ createFile(
+ "file1.parquet",
+ 1L,
+ 100,
+ 1,
+ Arrays.asList("file1.parquet.index", "file1.row"));
+
+ assertEquals("file1.row",
DataEvolutionSplitRead.rowSidecarFileName(file));
+ }
+
+ @Test
+ public void testRowSidecarFileNameWithNoOrAmbiguousSidecar() {
+ DataFileMeta noSidecar =
+ createFile(
+ "file1.parquet",
+ 1L,
+ 100,
+ 1,
+ Arrays.asList("file1.parquet.index", "lookup.sst"));
+ DataFileMeta ambiguousSidecar =
+ createFile(
+ "file1.parquet", 1L, 100, 1,
Arrays.asList("file1.row", "file1-copy.row"));
+
+ assertNull(DataEvolutionSplitRead.rowSidecarFileName(noSidecar));
+
assertNull(DataEvolutionSplitRead.rowSidecarFileName(ambiguousSidecar));
+ }
+
+ @Test
+ public void testShouldReadRowSidecarForSparseRowSelection() {
+ DataFileMeta file =
+ createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
+
+ assertTrue(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ file, Arrays.asList(new Range(10L, 10L), new
Range(42L, 42L))));
+ }
+
+ @Test
+ public void testShouldReadRowSidecarRequiresSmallCountAndLowRatio() {
+ DataFileMeta smallFile =
+ createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
+ DataFileMeta largeFile =
+ createFile(
+ "file2.parquet", 10L, 1_000_000, 1,
Collections.singletonList("file2.row"));
+
+ assertTrue(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ largeFile, Collections.singletonList(new Range(10L,
4105L))));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ smallFile, Collections.singletonList(new Range(10L,
15L))));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ largeFile, Collections.singletonList(new Range(10L,
4106L))));
+ assertTrue(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ smallFile, Collections.singletonList(new Range(10L,
15L)), 64L, 0.25d));
+ }
+
+ @Test
+ public void testShouldNotReadRowSidecarWithoutSparseSelection() {
+ DataFileMeta file =
+ createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
+
+ assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null));
+ assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file,
Collections.emptyList()));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ file, Collections.singletonList(new Range(10L,
109L))));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ createFile("file2.parquet", 10L, 100, 1),
+ Collections.singletonList(new Range(10L, 10L))));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ createFile(
+ "file3.blob", 10L, 100, 1,
Collections.singletonList("file3.row")),
+ Collections.singletonList(new Range(10L, 10L))));
+ }
+
+ @Test
+ public void testSelectedRowCountMergesOverlappingRanges() {
+ DataFileMeta file = createFile("file1.parquet", 10L, 100, 1);
+
+ assertEquals(
+ 12,
+ DataEvolutionSplitRead.selectedRowCount(
+ file,
+ Arrays.asList(
+ new Range(5L, 12L), new Range(12L, 15L), new
Range(20L, 25L))));
+ }
+
+ @Test
+ public void testSparseRowIdReadUsesRowSidecar() throws Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ Path tableRoot = new Path(tempDir.toUri().toString());
+ CoreOptions coreOptions = new CoreOptions(new Options());
+ FileStorePathFactory pathFactory =
+ new FileStorePathFactory(
+ tableRoot,
+ RowType.of(),
+ coreOptions.partitionDefaultName(),
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null);
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.STRING())
+ .build();
+ SchemaManager schemaManager = new SchemaManager(fileIO, tableRoot);
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ RowType rowType = tableSchema.logicalRowType();
+
+ Path bucketPath = pathFactory.bucketPath(EMPTY_ROW, 0);
+ fileIO.mkdirs(bucketPath);
+ String rowSidecarName = "data-0.row";
+ writeRowFile(fileIO, new Path(bucketPath, rowSidecarName), rowType,
100);
+
+ DataFileMeta dataFile =
+ createFile(
+ "data-0.parquet", 10L, 100, 1,
Collections.singletonList(rowSidecarName));
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath(bucketPath.toString())
+ .withDataFiles(Collections.singletonList(dataFile))
+ .rawConvertible(false)
+ .build();
+
+ DataEvolutionSplitRead splitRead =
+ new DataEvolutionSplitRead(
+ fileIO, schemaManager, tableSchema, rowType,
coreOptions, pathFactory);
+ IndexedSplit indexedSplit =
+ new IndexedSplit(
+ dataSplit, Arrays.asList(new Range(10L, 10L), new
Range(42L, 42L)), null);
+
+ List<Integer> actual = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
splitRead.createReader(indexedSplit)) {
+ reader.forEachRemaining(row -> actual.add(row.getInt(0)));
+ }
+
+ assertEquals(Arrays.asList(1000, 1032), actual);
+ }
+
+ private static void writeRowFile(LocalFileIO fileIO, Path path, RowType
rowType, int rowCount)
+ throws IOException {
+ FileFormat format = FileFormat.fromIdentifier("row", new Options());
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ FormatWriter writer =
format.createWriterFactory(rowType).create(out, "zstd");
+ for (int i = 0; i < rowCount; i++) {
+ writer.addElement(GenericRow.of(1000 + i,
BinaryString.fromString("row-" + i)));
+ }
+ writer.close();
+ }
+ }
+
private static DataFileMeta createFile(
String name, long firstRowId, long rowCount, long maxSequence) {
+ return createFile(name, firstRowId, rowCount, maxSequence,
Collections.emptyList());
+ }
+
+ private static DataFileMeta createFile(
+ String name,
+ long firstRowId,
+ long rowCount,
+ long maxSequence,
+ List<String> extraFiles) {
return DataFileMeta.create(
- name,
- 10000L,
- (int) rowCount,
- EMPTY_ROW,
- EMPTY_ROW,
- null,
- null,
- 0L,
- maxSequence,
- 0,
- 0,
- 0L,
- null,
- FileSource.APPEND,
- null,
- firstRowId,
- null);
+ name,
+ 10000L,
+ (int) rowCount,
+ EMPTY_ROW,
+ EMPTY_ROW,
+ null,
+ null,
+ 0L,
+ maxSequence,
+ 0,
+ 0,
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ firstRowId,
+ null)
+ .copy(extraFiles);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index 0b1799ac81..89742874c1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -31,6 +31,7 @@ import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.predicate.Predicate;
@@ -43,6 +44,7 @@ import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ReadBuilder;
@@ -72,6 +74,56 @@ import static
org.assertj.core.api.AssertionsForClassTypes.assertThat;
/** Test for table with data evolution. */
public class DataEvolutionTableTest extends DataEvolutionTestBase {
+ @Test
+ public void testRowSidecarDisabledByDefault() throws Exception {
+ createTableDefault();
+ List<DataFileMeta> newFiles =
writeOneFullRowAndCollectNewFiles(getTableDefault());
+
+
assertThat(newFiles.stream().anyMatch(DataEvolutionTableTest::containsRowSidecar))
+ .isFalse();
+ }
+
+ @Test
+ public void testRowSidecarEnabledByTableOption() throws Exception {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("f0", DataTypes.INT());
+ schemaBuilder.column("f1", DataTypes.STRING());
+ schemaBuilder.column("f2", DataTypes.STRING());
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+
schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ Schema schema = schemaBuilder.build();
+ catalog.createTable(identifier(), schema, true);
+
+ FileStoreTable table = getTableDefault();
+ List<CommitMessage> messages;
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(schema.rowType())) {
+ write.write(
+ GenericRow.of(1, BinaryString.fromString("a"),
BinaryString.fromString("b")));
+ messages = write.prepareCommit();
+ builder.newCommit().commit(messages);
+ }
+
+ List<DataFileMeta> newFiles = newFiles(messages);
+
assertThat(newFiles.stream().anyMatch(DataEvolutionTableTest::containsRowSidecar)).isTrue();
+ for (CommitMessage message : messages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ DataFilePathFactory dataFilePathFactory =
+ table.store()
+ .pathFactory()
+ .createDataFilePathFactory(message.partition(),
message.bucket());
+ for (DataFileMeta file :
commitMessage.newFilesIncrement().newFiles()) {
+ for (String extraFile : file.extraFiles()) {
+ if (extraFile.endsWith(".row")) {
+ table.fileIO()
+
.getFileStatus(dataFilePathFactory.toAlignedPath(extraFile, file));
+ }
+ }
+ }
+ }
+ }
+
@Test
public void testBasic() throws Exception {
createTableDefault();
@@ -1892,6 +1944,32 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
assertThat(plannedFileCount(table, RowType.of(SpecialFields.ROW_ID),
null)).isEqualTo(2);
}
+ private List<DataFileMeta>
writeOneFullRowAndCollectNewFiles(FileStoreTable table)
+ throws Exception {
+ Schema schema = schemaDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(schema.rowType())) {
+ write.write(
+ GenericRow.of(1, BinaryString.fromString("a"),
BinaryString.fromString("b")));
+ List<CommitMessage> messages = write.prepareCommit();
+ builder.newCommit().commit(messages);
+ return newFiles(messages);
+ }
+ }
+
+ private static List<DataFileMeta> newFiles(List<CommitMessage> messages) {
+ return messages.stream()
+ .flatMap(
+ message ->
+ ((CommitMessageImpl) message)
+
.newFilesIncrement().newFiles().stream())
+ .collect(Collectors.toList());
+ }
+
+ private static boolean containsRowSidecar(DataFileMeta file) {
+ return file.extraFiles().stream().anyMatch(extraFile ->
extraFile.endsWith(".row"));
+ }
+
private static int plannedFileCount(FileStoreTable table, RowType
readType, Predicate filter) {
ReadBuilder rb = table.newReadBuilder();
if (readType != null) {
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 2b24bc1be4..abd2bac190 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -581,6 +581,33 @@ class CoreOptions:
.default_value(False)
.with_description("Whether to enable data evolution.")
)
+
+ DATA_EVOLUTION_ROW_SIDECAR_ENABLED: ConfigOption[bool] = (
+ ConfigOptions.key("data-evolution.row-sidecar.enabled")
+ .boolean_type()
+ .default_value(False)
+ .with_description(
+ "Whether to generate row-store sidecar files for normal data files
on data evolution tables."
+ )
+ )
+
+ DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS: ConfigOption[int] = (
+ ConfigOptions.key("data-evolution.row-sidecar.max-selected-rows")
+ .long_type()
+ .default_value(4096)
+ .with_description(
+ "Maximum selected row count for reading a row-store sidecar file."
+ )
+ )
+
+ DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO: ConfigOption[float] = (
+ ConfigOptions.key("data-evolution.row-sidecar.max-selection-ratio")
+ .double_type()
+ .default_value(0.05)
+ .with_description(
+ "Maximum selected row ratio for reading a row-store sidecar file."
+ )
+ )
# External paths options
DATA_FILE_EXTERNAL_PATHS: ConfigOption[str] = (
ConfigOptions.key("data-file.external-paths")
@@ -1021,6 +1048,25 @@ class CoreOptions:
def data_evolution_enabled(self, default=None):
return self.options.get(CoreOptions.DATA_EVOLUTION_ENABLED, default)
+ def data_evolution_row_sidecar_enabled(self, default=None):
+ return
self.options.get(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED, default)
+
+ def data_evolution_row_sidecar_max_selected_rows(self, default=None):
+ max_selected_rows = self.options.get(
+ CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS, default)
+ if max_selected_rows <= 0:
+ raise ValueError(
+ "data-evolution.row-sidecar.max-selected-rows must be greater
than 0.")
+ return max_selected_rows
+
+ def data_evolution_row_sidecar_max_selection_ratio(self, default=None):
+ max_selection_ratio = self.options.get(
+ CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO,
default)
+ if max_selection_ratio <= 0 or max_selection_ratio > 1:
+ raise ValueError(
+ "data-evolution.row-sidecar.max-selection-ratio must be in (0,
1].")
+ return max_selection_ratio
+
def global_index_column_update_action(self, default=None):
return self.options.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION,
default)
diff --git a/paimon-python/pypaimon/read/reader/format_row_reader.py
b/paimon-python/pypaimon/read/reader/format_row_reader.py
index 34a9c663a9..5b0bed51f4 100644
--- a/paimon-python/pypaimon/read/reader/format_row_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_row_reader.py
@@ -17,7 +17,7 @@
import struct
from decimal import Decimal
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Tuple
import pyarrow as pa
import pyarrow.dataset as ds
@@ -29,12 +29,23 @@ from pypaimon.read.reader.iface.record_batch_reader import
RecordBatchReader
from pypaimon.schema.data_types import (
ArrayType, DataField, MapType, MultisetType, PyarrowFieldParser, RowType,
VectorType, AtomicType
)
+from pypaimon.table.special_fields import SpecialFields
FOOTER_SIZE = 32
MAGIC = 0x524F5753 # "ROWS"
VERSION = 1
+def _special_field(name: str):
+ if name == SpecialFields.ROW_ID.name:
+ return SpecialFields.ROW_ID
+ if name == SpecialFields.SEQUENCE_NUMBER.name:
+ return SpecialFields.SEQUENCE_NUMBER
+ if name == SpecialFields.VALUE_KIND.name:
+ return SpecialFields.VALUE_KIND
+ return None
+
+
class FormatRowReader(RecordBatchReader):
def __init__(self, file_io: FileIO, file_path: str, read_fields: List[str],
@@ -48,9 +59,13 @@ class FormatRowReader(RecordBatchReader):
self._file_size = file_io.get_file_size(file_path)
full_fields_map = {field.name: field for field in full_fields}
- self._projected_fields = [full_fields_map[name] for name in
read_fields]
+ self._read_field_names = list(read_fields)
+ self._projected_fields = [full_fields_map[name] for name in read_fields
+ if name in full_fields_map]
+ self._missing_fields = [name for name in read_fields
+ if name not in full_fields_map]
self._all_fields = full_fields
- self._schema =
PyarrowFieldParser.from_paimon_schema(self._projected_fields)
+ self._physical_schema =
PyarrowFieldParser.from_paimon_schema(self._projected_fields)
self._block_compressed_sizes: List[int] = []
self._block_uncompressed_sizes: List[int] = []
@@ -97,13 +112,12 @@ class FormatRowReader(RecordBatchReader):
block_data = self._read_and_decompress_block(self._current_block_idx)
self._current_block_idx += 1
- columns = self._decode_block(block_data)
+ columns, row_count = self._decode_block(block_data)
- if not columns or all(len(col) == 0 for col in columns):
+ if row_count == 0:
return None
- pydict = {field.name: columns[i] for i, field in
enumerate(self._projected_fields)}
- table = pa.Table.from_pydict(pydict, self._schema)
+ table = self._build_table(columns, row_count)
if self._push_down_predicate is not None:
dataset = ds.InMemoryDataset(table)
@@ -123,13 +137,12 @@ class FormatRowReader(RecordBatchReader):
self._blocks_to_read_pos += 1
block_data = self._read_and_decompress_block(block_idx)
- columns = self._decode_block(block_data, row_filter=local_rows)
+ columns, row_count = self._decode_block(block_data,
row_filter=local_rows)
- if not columns or all(len(col) == 0 for col in columns):
+ if row_count == 0:
return self._read_batch_indexed()
- pydict = {field.name: columns[i] for i, field in
enumerate(self._projected_fields)}
- table = pa.Table.from_pydict(pydict, self._schema)
+ table = self._build_table(columns, row_count)
if self._push_down_predicate is not None:
dataset = ds.InMemoryDataset(table)
@@ -211,8 +224,35 @@ class FormatRowReader(RecordBatchReader):
uncompressed_size = self._block_uncompressed_sizes[block_idx]
return decompressor.decompress(compressed_data,
max_output_size=uncompressed_size)
+ def _build_table(self, columns: List[List], row_count: int) -> pa.Table:
+ pydict = {
+ field.name: columns[i]
+ for i, field in enumerate(self._projected_fields)
+ }
+ if not self._missing_fields:
+ return pa.Table.from_pydict(pydict, self._physical_schema)
+
+ arrays = []
+ fields = []
+ projected_by_name = {field.name: field for field in
self._projected_fields}
+ for name in self._read_field_names:
+ if name in projected_by_name:
+ field = projected_by_name[name]
+ pa_type = PyarrowFieldParser.from_paimon_type(field.type)
+ arrays.append(pa.array(pydict[name], type=pa_type))
+ fields.append(pa.field(name, pa_type,
nullable=field.type.nullable))
+ else:
+ special_field = _special_field(name)
+ pa_type = (
+ PyarrowFieldParser.from_paimon_type(special_field.type)
+ if special_field is not None else pa.null()
+ )
+ arrays.append(pa.nulls(row_count, type=pa_type))
+ fields.append(pa.field(name, pa_type, nullable=not
SpecialFields.is_system_field(name)))
+ return pa.Table.from_arrays(arrays, schema=pa.schema(fields))
+
def _decode_block(self, block_data: bytes,
- row_filter: Optional[List[int]] = None) -> List[List]:
+ row_filter: Optional[List[int]] = None) ->
Tuple[List[List], int]:
data_len = len(block_data)
row_count = struct.unpack_from('<i', block_data, data_len - 4)[0]
offset_array_start = data_len - 4 - row_count * 4
@@ -229,6 +269,7 @@ class FormatRowReader(RecordBatchReader):
header_size = (arity + 7) // 8
rows_to_read = row_filter if row_filter is not None else
range(row_count)
+ rows_to_read_count = len(rows_to_read)
proj_set = set(projected_indices)
proj_col_map = {field_idx: col_idx for col_idx, field_idx in
enumerate(projected_indices)}
@@ -249,7 +290,7 @@ class FormatRowReader(RecordBatchReader):
if field_idx in proj_set:
columns[proj_col_map[field_idx]].append(value)
- return columns
+ return columns, rows_to_read_count
class _RowDecoder:
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index d6a74353f3..231b525bbf 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -72,6 +72,7 @@ from pypaimon.globalindex.indexed_split import IndexedSplit
KEY_PREFIX = "_KEY_"
KEY_FIELD_ID_START = 1000000
NULL_FIELD_INDEX = -1
+ROW_SIDECAR_FORMAT = CoreOptions.FILE_FORMAT_ROW
_COMPRESS_EXTENSIONS = frozenset(['gz', 'bz2', 'deflate', 'snappy', 'lz4',
'zst'])
@@ -204,12 +205,25 @@ class SplitRead(ABC):
batch_size = self.table.options.read_batch_size()
- # Convert global row_ranges (IndexedSplit) to local row_indices for
native pushdown.
- row_indices = None
+ effective_row_ranges = None
if row_ranges is not None:
effective_row_ranges = Range.and_(row_ranges,
[file.row_id_range()])
if len(effective_row_ranges) == 0:
return EmptyRecordBatchReader()
+
+ row_sidecar_file = self._row_sidecar_file_name(file)
+ if row_sidecar_file is not None and self._should_read_row_sidecar(
+ file,
+ effective_row_ranges,
+ row_sidecar_file,
+
self.table.options.data_evolution_row_sidecar_max_selected_rows(),
+
self.table.options.data_evolution_row_sidecar_max_selection_ratio()):
+ file_path = self._aligned_extra_file_path(file, row_sidecar_file)
+ file_format = ROW_SIDECAR_FORMAT
+
+ # Convert global row_ranges (IndexedSplit) to local row_indices for
native pushdown.
+ row_indices = None
+ if effective_row_ranges is not None:
row_index_formats = (CoreOptions.FILE_FORMAT_BLOB,
CoreOptions.FILE_FORMAT_VORTEX,
CoreOptions.FILE_FORMAT_LANCE,
@@ -392,6 +406,54 @@ class SplitRead(ABC):
return reader
+ @staticmethod
+ def _row_sidecar_file_name(file: DataFileMeta) -> Optional[str]:
+ row_files = [
+ extra_file for extra_file in file.extra_files
+ if SplitRead._is_row_sidecar_file(extra_file)
+ ]
+ return row_files[0] if len(row_files) == 1 else None
+
+ @staticmethod
+ def _should_read_row_sidecar(file: DataFileMeta,
+ effective_row_ranges: Optional[List[Range]],
+ row_sidecar_file: Optional[str],
+ max_selected_rows: int,
+ max_selection_ratio: float) -> bool:
+ if (not effective_row_ranges
+ or file.row_count <= 0
+ or DataFileMeta.is_blob_file(file.file_name)
+ or DataFileMeta.is_vector_file(file.file_name)
+ or row_sidecar_file is None):
+ return False
+
+ selected_row_count = sum(
+ r.count()
+ for r in Range.sort_and_merge_overlap(effective_row_ranges, True)
+ )
+ if selected_row_count <= 0 or selected_row_count >= file.row_count:
+ return False
+
+ selection_ratio = float(selected_row_count) / float(file.row_count)
+ return (selected_row_count <= max_selected_rows
+ and selection_ratio <= max_selection_ratio)
+
+ @staticmethod
+ def _is_row_sidecar_file(file_name: str) -> bool:
+ try:
+ return format_identifier(file_name) == ROW_SIDECAR_FORMAT
+ except Exception:
+ return False
+
+ @staticmethod
+ def _aligned_extra_file_path(file: DataFileMeta, extra_file: str) -> str:
+ if "://" in extra_file or extra_file.startswith("/"):
+ return extra_file
+ file_path = file.external_path if file.external_path else
file.file_path
+ if not file_path or "/" not in file_path:
+ return extra_file
+ return f"{file_path.rsplit('/', 1)[0]}/{extra_file}"
+
def _get_fields_and_predicate(self, schema_id: int, read_fields):
key = (schema_id, tuple(read_fields))
if key not in self.schema_id_2_fields:
diff --git a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
index b30560c400..50801ac7a6 100644
--- a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
@@ -31,6 +31,10 @@ import pyarrow as pa
from pypaimon import CatalogFactory, Schema
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.split_read import SplitRead
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.range import Range
class DataEvolutionFormatsTest(unittest.TestCase):
@@ -50,10 +54,190 @@ class DataEvolutionFormatsTest(unittest.TestCase):
def _file_path(file_meta):
return file_meta.external_path if file_meta.external_path else
file_meta.file_path
+ @staticmethod
+ def _extra_file_path(file_meta, extra_file):
+ if "://" in extra_file or extra_file.startswith("/"):
+ return extra_file
+ file_path = DataEvolutionFormatsTest._file_path(file_meta)
+ return f"{file_path.rsplit('/', 1)[0]}/{extra_file}"
+
+ @staticmethod
+ def _row_sidecar_files(file_meta):
+ return [extra_file for extra_file in file_meta.extra_files
+ if extra_file.endswith('.row')]
+
+ @staticmethod
+ def _data_file_meta(file_name, first_row_id, row_count, extra_files):
+ return DataFileMeta.create(
+ file_name=file_name,
+ file_size=1,
+ row_count=row_count,
+ min_key=GenericRow([], []),
+ max_key=GenericRow([], []),
+ key_stats=SimpleStats.empty_stats(),
+ value_stats=SimpleStats.empty_stats(),
+ min_sequence_number=0,
+ max_sequence_number=0,
+ schema_id=0,
+ level=0,
+ extra_files=extra_files,
+ first_row_id=first_row_id,
+ )
+
# ------------------------------------------------------------------
# Parquet-format data evolution
# ------------------------------------------------------------------
+ def test_row_sidecar_selection_requires_small_count_and_low_ratio(self):
+ small_file = self._data_file_meta('file1.parquet', 10, 100,
['file1.row'])
+ large_file = self._data_file_meta('file2.parquet', 10, 1000000,
['file2.row'])
+
+ self.assertTrue(SplitRead._should_read_row_sidecar(
+ large_file, [Range(10, 4105)], 'file2.row', 4096, 0.05))
+ self.assertFalse(SplitRead._should_read_row_sidecar(
+ small_file, [Range(10, 15)], 'file1.row', 4096, 0.05))
+ self.assertFalse(SplitRead._should_read_row_sidecar(
+ large_file, [Range(10, 4106)], 'file2.row', 4096, 0.05))
+ self.assertTrue(SplitRead._should_read_row_sidecar(
+ small_file, [Range(10, 15)], 'file1.row', 64, 0.25))
+
+ def test_row_sidecar_disabled_by_default(self):
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('val', pa.string()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'file.format': 'parquet',
+ })
+ self.catalog.create_table('default.fmt_row_sidecar_default_off',
schema, False)
+ table = self.catalog.get_table('default.fmt_row_sidecar_default_off')
+
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict(
+ {'id': [1, 2], 'val': ['a', 'b']}, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ data_files = [nf for m in cmts for nf in m.new_files
+ if nf.file_name.endswith('.parquet')]
+ self.assertGreater(len(data_files), 0)
+ for file_meta in data_files:
+ self.assertEqual([], self._row_sidecar_files(file_meta))
+
+ def test_row_sidecar_enabled_writes_extra_file(self):
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('val', pa.string()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'data-evolution.row-sidecar.enabled': 'true',
+ 'file.format': 'parquet',
+ })
+ self.catalog.create_table('default.fmt_row_sidecar_enabled', schema,
False)
+ table = self.catalog.get_table('default.fmt_row_sidecar_enabled')
+
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict(
+ {'id': [1, 2, 3], 'val': ['a', 'b', 'c']}, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ data_files = [nf for m in cmts for nf in m.new_files
+ if nf.file_name.endswith('.parquet')]
+ self.assertGreater(len(data_files), 0)
+ for file_meta in data_files:
+ sidecars = self._row_sidecar_files(file_meta)
+ self.assertEqual(1, len(sidecars))
+ self.assertTrue(os.path.exists(self._extra_file_path(file_meta,
sidecars[0])))
+
+ def test_row_sidecar_not_written_by_dedicated_format_writer(self):
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('payload', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'data-evolution.row-sidecar.enabled': 'true',
+ 'file.format': 'parquet',
+ })
+ self.catalog.create_table('default.fmt_row_sidecar_dedicated_writer',
schema, False)
+ table =
self.catalog.get_table('default.fmt_row_sidecar_dedicated_writer')
+
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'payload': pa.array([b'a', b'b'], type=pa.large_binary()),
+ }, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ all_files = [nf for m in cmts for nf in m.new_files]
+ self.assertGreater(len([f for f in all_files if
f.file_name.endswith('.parquet')]), 0)
+ self.assertGreater(len([f for f in all_files if
f.file_name.endswith('.blob')]), 0)
+ for file_meta in all_files:
+ self.assertEqual([], self._row_sidecar_files(file_meta))
+
+ def test_row_sidecar_serves_sparse_row_id_read(self):
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('val', pa.string()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'data-evolution.row-sidecar.enabled': 'true',
+ 'file.format': 'parquet',
+ })
+ self.catalog.create_table('default.fmt_row_sidecar_sparse_read',
schema, False)
+ table = self.catalog.get_table('default.fmt_row_sidecar_sparse_read')
+
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict({
+ 'id': list(range(100)),
+ 'val': [f'v{i}' for i in range(100)],
+ }, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ data_files = [nf for m in cmts for nf in m.new_files
+ if nf.file_name.endswith('.parquet')]
+ self.assertEqual(1, len(data_files))
+ sidecars = self._row_sidecar_files(data_files[0])
+ self.assertEqual(1, len(sidecars))
+ self.assertTrue(os.path.exists(self._extra_file_path(data_files[0],
sidecars[0])))
+
+ os.remove(self._file_path(data_files[0]))
+
+ rb = table.new_read_builder().with_projection(['id', 'val', '_ROW_ID'])
+ pb = rb.new_predicate_builder()
+ rb.with_filter(pb.equal('_ROW_ID', 5))
+ actual = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ self.assertEqual(actual.num_rows, 1)
+ self.assertEqual(actual.column('id').to_pylist(), [5])
+ self.assertEqual(actual.column('val').to_pylist(), ['v5'])
+ self.assertEqual(actual.column('_ROW_ID').to_pylist(), [5])
+
def test_parquet_column_subset_write_and_merge_read(self):
"""Write disjoint column subsets as parquet, merge-read via data
evolution."""
pa_schema = pa.schema([
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 75a6359a09..c32bb6968b 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -34,6 +34,8 @@ from pypaimon.table.row.generic_row import GenericRow
class DataWriter(ABC):
"""Base class for data writers that handle PyArrow tables directly."""
+ ROW_SIDECAR_SUFFIX = ".row"
+
def __init__(self, table, partition: Tuple, bucket: int, max_seq_number:
int, options: CoreOptions = None,
write_cols: Optional[List[str]] = None,
changelog_producer: ChangelogProducer =
ChangelogProducer.NONE):
@@ -152,6 +154,8 @@ class DataWriter(ABC):
if path_to_delete:
path_str = str(path_to_delete)
self.file_io.delete_quietly(path_str)
+ for extra_file in file_meta.extra_files:
+
self.file_io.delete_quietly(self._aligned_extra_file_path(file_meta,
extra_file))
except Exception as e:
import logging
logger = logging.getLogger(__name__)
@@ -198,27 +202,46 @@ class DataWriter(ABC):
is_external_path = self.external_path_provider is not None
external_path_str = file_path if is_external_path else None
+ logical_data = data
+ extra_files = []
+ row_sidecar_path = None
if self._variant_shredding:
data = self._apply_variant_shredding(data)
- if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
- self.file_io.write_parquet(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
- self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
- self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_BLOB:
- self.file_io.write_blob(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
- self.file_io.write_lance(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
- self.file_io.write_vortex(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
- self.file_io.write_row(file_path, data, zstd_level=self.zstd_level)
- else:
- raise ValueError(f"Unsupported file format: {self.file_format}")
+ try:
+ if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
+ self.file_io.write_parquet(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
+ self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
+ self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_BLOB:
+ self.file_io.write_blob(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
+ self.file_io.write_lance(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
+ self.file_io.write_vortex(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
+ self.file_io.write_mosaic(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
+ self.file_io.write_row(file_path, data,
zstd_level=self.zstd_level)
+ else:
+ raise ValueError(f"Unsupported file format:
{self.file_format}")
+
+ if self._should_write_row_sidecar():
+ row_sidecar_name = f"{file_name}{self.ROW_SIDECAR_SUFFIX}"
+ row_sidecar_path = f"{file_path}{self.ROW_SIDECAR_SUFFIX}"
+ self.file_io.write_row(
+ row_sidecar_path,
+ logical_data,
+ fields=self._row_sidecar_fields(logical_data),
+ zstd_level=self.zstd_level)
+ extra_files.append(row_sidecar_name)
+ except Exception:
+ self.file_io.delete_quietly(file_path)
+ if row_sidecar_path is not None:
+ self.file_io.delete_quietly(row_sidecar_path)
+ raise
# min key & max key
@@ -264,7 +287,7 @@ class DataWriter(ABC):
max_sequence_number=max_seq,
schema_id=self.table.table_schema.id,
level=0,
- extra_files=[],
+ extra_files=extra_files,
creation_time=creation_time,
delete_row_count=0,
file_source=0,
@@ -359,6 +382,27 @@ class DataWriter(ABC):
bucket_path = self.path_factory.bucket_path(self.partition,
self.bucket)
return f"{bucket_path.rstrip('/')}/{file_name}"
+ def _should_write_row_sidecar(self) -> bool:
+ return (
+ self.options.data_evolution_enabled(False)
+ and self.options.data_evolution_row_sidecar_enabled(False)
+ )
+
+ def _row_sidecar_fields(self, data: pa.Table) -> List:
+ if self.write_cols:
+ field_map = {field.name: field for field in
self.table.table_schema.fields}
+ return [field_map[name] for name in self.write_cols if name in
field_map]
+ return PyarrowFieldParser.to_paimon_schema(data.schema)
+
+ @staticmethod
+ def _aligned_extra_file_path(file_meta: DataFileMeta, extra_file: str) ->
str:
+ if "://" in extra_file or extra_file.startswith("/"):
+ return extra_file
+ file_path = file_meta.external_path if file_meta.external_path else
file_meta.file_path
+ if not file_path or "/" not in file_path:
+ return extra_file
+ return f"{file_path.rsplit('/', 1)[0]}/{extra_file}"
+
@staticmethod
def _find_optimal_split_point(data: pa.RecordBatch, target_size: int) ->
int:
total_rows = data.num_rows
