Skip to content

[core] Select row sidecars using the final Data Evolution selection #10415

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 7 commits into from
Oct 16, 2026
Merged

[core] Select row sidecars using the final Data Evolution selection #10415

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
165cefa
[core] Select row sidecars using the final Data Evolution selection
lilei1128 Oct 7, 2026
f6eb605
Avoid row sidecar for bitmap-only reads without row ranges
lilei1128 Oct 8, 2026
5ee90ac
fix code format
lilei1128 Oct 8, 2026
506ac4a
fix row sidecar schema consistency
lilei1128 Oct 8, 2026
1369458
fix row sidecar reader cache key
lilei1128 Oct 10, 2026
fa8fbe0
Preserve tracking and partition fields in data evolution reads
lilei1128 Oct 15, 2026
bab496f
chore: trigger ci
lilei1128 Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 147 additions & 35 deletions paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,14 @@
package org.apache.paimon.operation;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.PartitionSettedRow;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.append.ForceSingleBatchReader;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.casting.CastFieldGetter;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.PartitionInfo;
import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.disk.IOManager;
Expand Down Expand Up @@ -61,6 +65,8 @@
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.FormatReaderMapping;
import org.apache.paimon.utils.FormatReaderMapping.Builder;
import org.apache.paimon.utils.InternalRowUtils;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RangeHelper;
Expand Down Expand Up @@ -212,7 +218,15 @@ private RecordReader<InternalRow> createReader(DataSplit dataSplit) throws IOExc
}

private RecordReader<InternalRow> createReader(
DataSplit dataSplit, List<Range> rowRanges, RowType readRowType) throws IOException {
DataSplit dataSplit, List<Range> rowRanges, RowType outputRowType) throws IOException {
// Partition values belong to the split and may be absent from every physical file.
// Plan and merge only data columns, then restore partitions in the requested order.
Pair<Pair<int[], RowType>, List<DataField>> partitionMapping =
PartitionUtils.getPartitionMapping2fieldsWithoutPartition(
schema.partitionKeys(),
outputRowType.getFields(),
schema.logicalPartitionType());
RowType readRowType = new RowType(partitionMapping.getRight());
List<DataFileMeta> files = dataSplit.dataFiles();

List<DeletionFile> deletionFiles = dataSplit.deletionFiles().orElse(null);
Expand All @@ -228,7 +242,7 @@ private RecordReader<InternalRow> createReader(

// the suppliers below run lazily, so take the filters now, the same way the read type is
// already taken by the caller
List<Predicate> filters = readTypeFilters(this.filters, readRowType);
List<Predicate> filters = readTypeFilters(this.filters, outputRowType);

List<List<DataFileMeta>> splitByRowId = mergeRangesAndSort(files);
for (List<DataFileMeta> needMergeFiles : splitByRowId) {
Expand Down Expand Up @@ -308,7 +322,13 @@ private RecordReader<InternalRow> createReader(
}
}

return ConcatRecordReader.create(suppliers);
RecordReader<InternalRow> reader = ConcatRecordReader.create(suppliers);
PartitionInfo partitionInfo = PartitionUtils.create(partitionMapping.getLeft(), partition);
if (partitionInfo != null) {
PartitionSettedRow partitionRow = PartitionSettedRow.from(partitionInfo);
return reader.transform(partitionRow::replaceRow);
}
return reader;
}

private RecordReader<InternalRow> createReader(IndexedSplit indexedSplit) throws IOException {
Expand Down Expand Up @@ -455,7 +475,7 @@ private RecordReader<InternalRow> createUnionReader(
fieldsFiles.get(0),
bunchDataSchemas[0],
dataFilePathFactory,
formatBuilder,
nestedFieldEnabled,
rowRanges,
readRowType,
deletionVector,
Expand All @@ -471,20 +491,23 @@ private RecordReader<InternalRow> createUnionReader(
}
FieldBunch bunch = fieldsFiles.get(i);
DataFileMeta firstFile = bunch.files().get(0);
FileReadTarget readTarget = readTarget(firstFile, dataFilePathFactory, rowRanges);
FileReadTarget readTarget =
readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection);
String formatIdentifier = readTarget.formatIdentifier;
Builder targetFormatBuilder =
readTarget.isPositionalRowFormat()
? positionalRowFormatBuilder(readRowType, null, nestedFieldEnabled)
: formatBuilder;
long schemaId = firstFile.schemaId();
TableSchema dataSchema = bunchDataSchemas[i];
RowType partialReadRowType = new RowType(readFields);
List<String> cacheKey =
nestedFieldEnabled
? readerCacheKey(readFields, dataSchema.fields(), true)
: readFields.stream().map(DataField::name).collect(Collectors.toList());
readerCacheKey(readFields, dataSchema.fields(), nestedFieldEnabled);
FormatReaderMapping formatReaderMapping =
formatReaderMappings.computeIfAbsent(
new FormatKey(schemaId, formatIdentifier, cacheKey),
key ->
formatBuilder.build(
targetFormatBuilder.build(
formatIdentifier,
schema,
dataSchema,
Expand Down Expand Up @@ -515,17 +538,23 @@ private RecordReader<InternalRow> createMissingFieldsReader(
FieldBunch bunch,
TableSchema dataSchema,
DataFilePathFactory dataFilePathFactory,
Builder formatBuilder,
boolean nestedFieldEnabled,
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
@Nullable BitmapIndexResult groupSelection)
throws IOException {
DataFileMeta firstFile = bunch.files().get(0);
FileReadTarget readTarget =
readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection);
Builder formatBuilder =
readTarget.isPositionalRowFormat()
? positionalRowFormatBuilder(readRowType, null, nestedFieldEnabled)
: formatBuilder(readRowType, null, nestedFieldEnabled);
// Use the physical schema: the full table schema may declare columns this file never wrote.
FormatReaderMapping mapping =
formatBuilder.build(
readTarget(firstFile, dataFilePathFactory, rowRanges).formatIdentifier,
readTarget.formatIdentifier,
schema,
dataSchema,
readRowType.getFields(),
Expand Down Expand Up @@ -722,27 +751,14 @@ private FileRecordReader<InternalRow> createFileReader(
@Nullable DeletionVectorWithRange deletionVector,
@Nullable FileIndexResult precomputedFileIndexResult)
throws IOException {
FileReadTarget readTarget = readTarget(file, dataFilePathFactory, rowRanges);
String formatIdentifier = readTarget.formatIdentifier;
long schemaId = file.schemaId();
TableSchema dataSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId);
TableSchema fileSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId);
TableSchema physicalDataSchema = fileSchema.dataFileSchema(file.writeCols());
boolean nestedFieldEnabled = nestedFieldEnabledFor(Collections.singletonList(file));

// no column merge here, so the filters this file can answer reach both the file index and
// the format reader
List<Predicate> fileFilters = fileFilters(filters, file);
FormatReaderMapping formatReaderMapping =
singleFileReaderMappings.computeIfAbsent(
new SingleFileKey(
schemaId,
formatIdentifier,
file.writeCols(),
readRowType,
nestedFieldEnabled),
key ->
formatBuilder(readRowType, fileFilters, nestedFieldEnabled)
.build(formatIdentifier, schema, dataSchema));

FileIndexResult fileIndexResult = precomputedFileIndexResult;
if (fileIndexResult == null && fileIndexReadEnabled) {
DeletionVector dv = deletionVector == null ? null : deletionVector.deletionVector;
Expand All @@ -754,8 +770,8 @@ private FileRecordReader<InternalRow> createFileReader(
fileIndexResult =
FileIndexEvaluator.evaluate(
fileIO,
dataSchema,
devolveFilters(fileFilters, dataSchema),
physicalDataSchema,
devolveFilters(fileFilters, physicalDataSchema),
null,
null,
dataFilePathFactory,
Expand All @@ -767,6 +783,27 @@ private FileRecordReader<InternalRow> createFileReader(
}
}

FileReadTarget readTarget =
readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult);
String formatIdentifier = readTarget.formatIdentifier;
boolean positionalRowFormat = readTarget.isPositionalRowFormat();
Builder targetFormatBuilder =
positionalRowFormat
? positionalRowFormatBuilder(readRowType, fileFilters, nestedFieldEnabled)
: formatBuilder(readRowType, fileFilters, nestedFieldEnabled);
TableSchema readerDataSchema = positionalRowFormat ? physicalDataSchema : fileSchema;
FormatReaderMapping formatReaderMapping =
singleFileReaderMappings.computeIfAbsent(
new SingleFileKey(
schemaId,
formatIdentifier,
file.writeCols(),
readRowType,
nestedFieldEnabled),
key ->
targetFormatBuilder.build(
formatIdentifier, schema, readerDataSchema));

return createFileReader(
partition,
file,
Expand Down Expand Up @@ -814,7 +851,7 @@ private FileRecordReader<InternalRow> createFileReader(
formatReaderMapping,
rowRanges,
readRowType,
readTarget(file, dataFilePathFactory, rowRanges),
readTarget(file, dataFilePathFactory, rowRanges, groupSelection),
deletionVector,
groupSelection);
}
Expand Down Expand Up @@ -856,7 +893,7 @@ private FileRecordReader<InternalRow> createFileReader(
coreOptions.scanIgnoreCorruptFile(),
coreOptions.scanIgnoreLostFile(),
formatReaderMapping.getIndexMapping(),
formatReaderMapping.getCastMapping(),
trackingCastMapping(readTarget, formatReaderMapping),
PartitionUtils.create(formatReaderMapping.getPartitionPair(), partition),
true,
file.firstRowId(),
Expand All @@ -872,6 +909,32 @@ private FileRecordReader<InternalRow> createFileReader(
fileRecordReader, file.nonNullRowIdRange(), rowRanges, deletionVector);
}

@Nullable
private CastFieldGetter[] trackingCastMapping(
FileReadTarget target, FormatReaderMapping mapping) {
CastFieldGetter[] casts = mapping.getCastMapping();
if (!target.isPositionalRowFormat()
|| casts == null
|| mapping.getSystemFields().isEmpty()) {
return casts;
}
// The physical schema omits tracking fields, but DataFileRecordReader supplies them
// before casting. Preserve those values without changing the cached cast mapping.
CastFieldGetter[] result = casts.clone();
for (DataField field :
new DataField[] {SpecialFields.ROW_ID, SpecialFields.SEQUENCE_NUMBER}) {
Integer index = mapping.getSystemFields().get(field.name());
if (index != null) {
result[index] =
new CastFieldGetter(
InternalRowUtils.createNullCheckingFieldGetter(field.type(), index),
CastExecutors.identityCastExecutor(),
field.name());
}
}
return result;
}

private FileRecordReader<InternalRow> applyDeletionVector(
FileRecordReader<InternalRow> reader,
Range readerRange,
Expand Down Expand Up @@ -1078,6 +1141,20 @@ private Builder formatBuilder(
nestedFieldEnabled);
}

// Row sidecars and ordinary row-format data files decode against the physical write schema.
// Row tracking fields are supplied by DataFileRecordReader from the manifest entry.
private Builder positionalRowFormatBuilder(
RowType readRowType, @Nullable List<Predicate> filters, boolean nestedFieldEnabled) {
return new Builder(
formatDiscover,
readRowType.getFields(),
TableSchema::fields,

@JingsongLi JingsongLi Oct 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve manifest tracking fields through schema casts

Using the physical schema is necessary for positional decoding, but it leaves _ROW_ID and _SEQUENCE_NUMBER absent from readDataFields. SchemaEvolutionUtil.createCastFieldGetterMapping creates a getter returning null for each missing field. When another column requires a schema-evolution cast, the entire cast mapping becomes active. DataFileRecordReader currently restores manifest tracking values before applying CastedRow, so those values are then hidden by the null getters. isNullAt() still delegates to the underlying row and returns false, but getLong() unboxes null and throws a NullPointerException.

I reproduced this using real table writes: write 100 rows with f0 INT, a bitmap index on f1, and row sidecars enabled; ALTER f0 to BIGINT; then read the widened column and tracking fields with a selective f1 = 'a050' filter through executeFilter(). Both the ordinary bitmap read and the variant with _ROW_ID BETWEEN 0 AND 99 throw NPE. The same cases pass on the pre-change implementation.

Please preserve tracking values through casting, for example by restoring non-columnar tracking fields after applying the cast, or by using identity getters for the tracking slots. These virtual fields should remain excluded from the physical row decoder schema.

filters,
null,
null,
nestedFieldEnabled);
}

/**
* Filters the file can answer. A column missing from a data evolution file is not null, its
* values live in another file of the same row id range, so a predicate on it must not be pushed
Expand Down Expand Up @@ -1173,13 +1250,17 @@ private DeletionVectorWithRange readDeletionVector(
}

private FileReadTarget readTarget(
DataFileMeta file, DataFilePathFactory dataFilePathFactory, List<Range> rowRanges)
DataFileMeta file,
DataFilePathFactory dataFilePathFactory,
List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult)
throws IOException {
String rowSidecar = rowSidecarFileName(file);
if (rowSidecar != null
&& shouldReadRowSidecar(
file,
rowRanges,
fileIndexResult,
coreOptions.dataEvolutionRowSidecarMaxSelectedRows(),
coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) {
Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file);
Expand Down Expand Up @@ -1213,6 +1294,7 @@ static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable List<Range> row
return shouldReadRowSidecar(
file,
rowRanges,
null,
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(),
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue());
}
Expand All @@ -1223,16 +1305,24 @@ static boolean shouldReadRowSidecar(
@Nullable List<Range> rowRanges,
long maxSelectedRows,
double maxSelectionRatio) {
if (rowRanges == null
|| rowRanges.isEmpty()
|| file.rowCount() <= 0
return shouldReadRowSidecar(file, rowRanges, null, maxSelectedRows, maxSelectionRatio);
}

@VisibleForTesting
static boolean shouldReadRowSidecar(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult,
long maxSelectedRows,
double maxSelectionRatio) {
if (file.rowCount() <= 0
|| isBlobFile(file.fileName())
|| isVectorStoreFile(file.fileName())
|| rowSidecarFileName(file) == null) {
return false;
}

long selectedRowCount = selectedRowCount(file, rowRanges);
long selectedRowCount = selectedRowCount(file, rowRanges, fileIndexResult);

@JingsongLi JingsongLi Oct 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Non-empty row ranges still expose the physical-schema mismatch

The null/empty guard fixes the ordinary filtered-read cases, but this bitmap-based count still enables new sidecar reads for wide, non-empty ranges. For a 100-row file with rowRanges = [0, 99] and a bitmap containing only position 50, the baseline counts 100 selected rows and keeps reading Parquet; this version counts one row and switches to the row sidecar. The unchanged reader mapping then decodes the sidecar using the wrong physical schema.

This is reachable through normal APIs: ReadBuilder.withRowRanges(Collections.singletonList(new Range(0L, 99L))) combined with a selective file-index predicate, or a predicate such as _ROW_ID BETWEEN 0 AND 99 AND c6 = 50. Non-empty ranges are not guaranteed to be sparse or to include the file-index filtering result.

I extended the two new regression tests to exercise both range sources, including _ROW_ID in the read type for the row-id predicate cases. The projected [f2] file and seven-INT-column cases both incorrectly return no rows through executeFilter(). All four cases pass with the pre-change implementation (3f297583) and fail on this head (5ee90ac7); the existing 58 tests still pass on JDK 8.

Please fix the row decoder's physical schema mapping before enabling these additional sidecar switches: use dataFileSchema(file.writeCols()) and exclude the virtual tracking fields from the decoding schema, while continuing to assign tracking values from the manifest. Requiring non-empty rowRanges alone does not protect this path.

@lilei1128 lilei1128 Oct 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed by using the physical data schema for row-sidecar decoding, while continuing to supply row-tracking fields from the manifest. Added regression coverage for full explicit row ranges and _ROW_ID predicates.

if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) {
return false;
}
Expand All @@ -1256,6 +1346,20 @@ static long selectedRowCount(DataFileMeta file, List<Range> rowRanges) {
.sum();
}

private static long selectedRowCount(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult) {
if (fileIndexResult instanceof BitmapIndexResult) {
RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get();
RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges);
return rangeSelection == null
? indexSelection.getCardinality()
: RoaringBitmap32.and(rangeSelection, indexSelection).getCardinality();
}
return isNullOrEmpty(rowRanges) ? file.rowCount() : selectedRowCount(file, rowRanges);
}

private static boolean isRowSidecarFile(String fileName) {
try {
return ROW_SIDECAR_FORMAT.equals(DataFilePathFactory.formatIdentifier(fileName));
Expand All @@ -1275,6 +1379,14 @@ private FileReadTarget(String formatIdentifier, Path path, long fileSize) {
this.path = path;
this.fileSize = fileSize;
}

/**
* Sidecars and ordinary "row" data files use the same positional decoder and physical
* schema mapping. They can share a cache entry for the same physical and requested fields.
*/
private boolean isPositionalRowFormat() {
return ROW_SIDECAR_FORMAT.equals(formatIdentifier);
}
}

/**
Expand Down
Loading
Loading