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 e3e06017e0 [core] Select row sidecars using the final Data Evolution
selection (#10415)
e3e06017e0 is described below
commit e3e06017e0a255354e02630a6f834ad622217081
Author: sanshi <[email protected]>
AuthorDate: Fri Oct 16 13:10:56 2026 +0800
[core] Select row sidecars using the final Data Evolution selection (#10415)
---
.../paimon/operation/DataEvolutionSplitRead.java | 182 +++++--
.../operation/DataEvolutionSplitReadTest.java | 28 +
.../paimon/table/DataEvolutionFileIndexTest.java | 567 +++++++++++++++++++++
3 files changed, 742 insertions(+), 35 deletions(-)
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 b4d58af88b..783a763369 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
@@ -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;
@@ -61,6 +65,8 @@ import org.apache.paimon.types.RowType;
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;
@@ -212,7 +218,15 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
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);
@@ -228,7 +242,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
// 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) {
@@ -317,7 +331,13 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
}
- 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 {
@@ -465,7 +485,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
fieldsFiles.get(0),
bunchDataSchemas[0],
dataFilePathFactory,
- formatBuilder,
+ nestedFieldEnabled,
rowRanges,
readRowType,
deletionVector,
@@ -481,20 +501,23 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
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,
@@ -525,17 +548,23 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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(),
@@ -732,27 +761,14 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
@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;
@@ -764,8 +780,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
fileIndexResult =
FileIndexEvaluator.evaluate(
fileIO,
- dataSchema,
- devolveFilters(fileFilters, dataSchema),
+ physicalDataSchema,
+ devolveFilters(fileFilters, physicalDataSchema),
null,
null,
dataFilePathFactory,
@@ -777,6 +793,27 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
}
+ 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,
@@ -824,7 +861,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
formatReaderMapping,
rowRanges,
readRowType,
- readTarget(file, dataFilePathFactory, rowRanges),
+ readTarget(file, dataFilePathFactory, rowRanges,
groupSelection),
deletionVector,
groupSelection);
}
@@ -866,7 +903,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
coreOptions.scanIgnoreCorruptFile(),
coreOptions.scanIgnoreLostFile(),
formatReaderMapping.getIndexMapping(),
- formatReaderMapping.getCastMapping(),
+ trackingCastMapping(readTarget, formatReaderMapping),
PartitionUtils.create(formatReaderMapping.getPartitionPair(), partition),
true,
file.firstRowId(),
@@ -882,6 +919,32 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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,
@@ -1103,6 +1166,20 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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,
+ 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
@@ -1198,13 +1275,17 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
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);
@@ -1238,6 +1319,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return shouldReadRowSidecar(
file,
rowRanges,
+ null,
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(),
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue());
}
@@ -1248,16 +1330,24 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
@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);
if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) {
return false;
}
@@ -1281,6 +1371,20 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
.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));
@@ -1300,6 +1404,14 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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);
+ }
}
/**
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 cd60d98164..2a1c5c4472 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
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.Path;
@@ -42,6 +43,7 @@ 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.apache.paimon.utils.RoaringBitmap32;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -225,6 +227,32 @@ class DataEvolutionSplitReadTest {
smallFile, Collections.singletonList(new Range(10L,
15L)), 64L, 0.25d));
}
+ @Test
+ public void testShouldReadRowSidecarUsesFinalBitmapSelection() {
+ DataFileMeta file =
+ createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
+ BitmapIndexResult bitmap = new BitmapIndexResult(() ->
RoaringBitmap32.bitmapOf(0, 5, 20));
+ BitmapIndexResult singleRowBitmap =
+ new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(50));
+ List<Range> rowRanges = Collections.singletonList(new Range(10L, 19L));
+
+ assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file,
rowRanges, bitmap, 2L, 0.02d));
+ assertTrue(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ file, null, singleRowBitmap, 1L, 0.01d));
+ assertTrue(
+ DataEvolutionSplitRead.shouldReadRowSidecar(
+ file,
+ Collections.singletonList(new Range(10L, 109L)),
+ singleRowBitmap,
+ 1L,
+ 0.01d));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges,
bitmap, 1L, 0.02d));
+ assertFalse(
+ DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges,
bitmap, 2L, 0.01d));
+ }
+
@Test
public void testShouldNotReadRowSidecarWithoutSparseSelection() {
DataFileMeta file =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
index be205c1e3a..1e7d086679 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
@@ -37,9 +37,12 @@ import
org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.BundleRecords;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -75,6 +78,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -114,6 +118,485 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
/** Inside the min/max range of f2 ("b000".."b099"), but not written. */
private static final String MISSING_F2 = "b050x";
+ @Test
+ public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("projected_schema_row_sidecar",
options);
+ writeSplitColumns(table, ROW_COUNT, Collections.emptyMap(),
bitmapOptions("f2"));
+
+ FileStoreTable latest = getTable(identifier(table.name()));
+ RowType projection = latest.rowType().project("f2");
+ Predicate filter = equalF2(f2(50));
+ ReadBuilder readBuilder =
+
latest.newReadBuilder().withReadType(projection).withFilter(filter);
+ List<InternalRow> rows =
+ collect(
+ readBuilder.newRead().executeFilter(),
+ readBuilder.newScan().plan(),
+ projection);
+
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo(f2(50));
+
+ ReadBuilder rangeReadBuilder =
+ latest.newReadBuilder()
+ .withReadType(projection)
+ .withFilter(filter)
+ .withRowRanges(Collections.singletonList(new Range(0L,
ROW_COUNT - 1L)));
+ rows =
+ collect(
+ rangeReadBuilder.newRead().executeFilter(),
+ rangeReadBuilder.newScan().plan(),
+ projection);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo(f2(50));
+
+ RowType rowIdType = rowTypeWithRowId(latest.rowType());
+ RowType rowIdProjection =
rowIdType.project(SpecialFields.ROW_ID.name(), "f2");
+ PredicateBuilder rowIdBuilder = new PredicateBuilder(rowIdType);
+ Predicate rowIdFilter =
+ PredicateBuilder.and(
+ rowIdBuilder.between(3, 0L, ROW_COUNT - 1L),
+ rowIdBuilder.equal(2,
BinaryString.fromString(f2(50))));
+ ReadBuilder rowIdReadBuilder =
+
latest.newReadBuilder().withReadType(rowIdProjection).withFilter(rowIdFilter);
+ rows =
+ collect(
+ rowIdReadBuilder.newRead().executeFilter(),
+ rowIdReadBuilder.newScan().plan(),
+ rowIdProjection);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getLong(0)).isEqualTo(50L);
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo(f2(50));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testPositionalRowPreservesTrackingFieldsAfterSchemaCast(boolean
ordinaryRowFormat)
+ throws Exception {
+ Map<String, String> options = bitmapOptions("f1");
+ options.put(
+ CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
+ Boolean.toString(!ordinaryRowFormat));
+ if (ordinaryRowFormat) {
+ options.put(CoreOptions.FILE_FORMAT.key(), "row");
+ }
+ FileStoreTable table = createTable("row_sidecar_tracking_cast",
options);
+ writeAllColumns(table, ROW_COUNT);
+ catalog.alterTable(
+ identifier("row_sidecar_tracking_cast"),
+ SchemaChange.updateColumnType("f0", DataTypes.BIGINT()),
+ false);
+
+ FileStoreTable latest =
getTable(identifier("row_sidecar_tracking_cast"));
+ RowType readType =
+ SpecialFields.rowTypeWithRowTracking(latest.rowType(), true,
true)
+ .project(
+ "f0",
+ SpecialFields.ROW_ID.name(),
+ SpecialFields.SEQUENCE_NUMBER.name());
+ Predicate filter =
+ new PredicateBuilder(latest.rowType()).equal(1,
BinaryString.fromString(f1(50)));
+
+ ReadBuilder readBuilder =
latest.newReadBuilder().withReadType(readType).withFilter(filter);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ long maxSequenceNumber =
+ ((DataSplit)
plan.splits().get(0)).dataFiles().get(0).maxSequenceNumber();
+ List<InternalRow> rows =
collect(readBuilder.newRead().executeFilter(), plan, readType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getLong(0)).isEqualTo(50L);
+ assertThat(rows.get(0).getLong(1)).isEqualTo(50L);
+ assertThat(rows.get(0).isNullAt(2)).isFalse();
+ assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber);
+
+ ReadBuilder rangeReadBuilder =
+ latest.newReadBuilder()
+ .withReadType(readType)
+ .withFilter(filter)
+ .withRowRanges(Collections.singletonList(new Range(0L,
ROW_COUNT - 1L)));
+ rows =
+ collect(
+ rangeReadBuilder.newRead().executeFilter(),
+ rangeReadBuilder.newScan().plan(),
+ readType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getLong(0)).isEqualTo(50L);
+ assertThat(rows.get(0).getLong(1)).isEqualTo(50L);
+ assertThat(rows.get(0).isNullAt(2)).isFalse();
+ assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber);
+
+ RowType trackedType =
SpecialFields.rowTypeWithRowTracking(latest.rowType(), true, true);
+ PredicateBuilder trackedBuilder = new PredicateBuilder(trackedType);
+ Predicate rowIdFilter =
+ PredicateBuilder.and(
+ filter,
+ trackedBuilder.between(
+
trackedType.getFieldIndex(SpecialFields.ROW_ID.name()),
+ 0L,
+ ROW_COUNT - 1L));
+ ReadBuilder rowIdReadBuilder =
+
latest.newReadBuilder().withReadType(readType).withFilter(rowIdFilter);
+ rows =
+ collect(
+ rowIdReadBuilder.newRead().executeFilter(),
+ rowIdReadBuilder.newScan().plan(),
+ readType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getLong(0)).isEqualTo(50L);
+ assertThat(rows.get(0).getLong(1)).isEqualTo(50L);
+ assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRowSidecarRestoresMissingPartitionField(boolean merged)
throws Exception {
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .column("pt", DataTypes.STRING())
+ .column("f1", DataTypes.STRING())
+ .column("f2", DataTypes.STRING())
+ .partitionKeys("pt")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+
.option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true");
+ Identifier identifier = identifier("row_sidecar_partition_restore");
+ catalog.createTable(identifier, schemaBuilder.build(), false);
+ FileStoreTable table = getTable(identifier);
+
+ RowType writeType = table.rowType().project("f1");
+ BatchWriteBuilder writeBuilder =
table.copy(bitmapOptions("f1")).newBatchWriteBuilder();
+ List<InternalRow> writtenRows = new ArrayList<>();
+ for (int i = 0; i < ROW_COUNT; i++) {
+ writtenRows.add(GenericRow.of(BinaryString.fromString(f1(i))));
+ }
+ try (BatchTableWrite write =
writeBuilder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.writeBundle(
+ new
InternalRowSerializer(table.schema().logicalPartitionType())
+
.toBinaryRow(GenericRow.of(BinaryString.fromString("p1"))),
+ 0,
+ new BundleRecords() {
+ @Override
+ public long rowCount() {
+ return writtenRows.size();
+ }
+
+ @Override
+ public Iterator<InternalRow> iterator() {
+ return writtenRows.iterator();
+ }
+ });
+ commit.commit(write.prepareCommit());
+ }
+
+ if (merged) {
+ BatchWriteBuilder mergeWriteBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ mergeWriteBuilder
+ .newWrite()
+
.withWriteType(table.rowType().project("f2"));
+ BatchTableCommit commit = mergeWriteBuilder.newCommit()) {
+ write.writeBundle(
+ new
InternalRowSerializer(table.schema().logicalPartitionType())
+
.toBinaryRow(GenericRow.of(BinaryString.fromString("p1"))),
+ 0,
+ new BundleRecords() {
+ @Override
+ public long rowCount() {
+ return ROW_COUNT;
+ }
+
+ @Override
+ public Iterator<InternalRow> iterator() {
+ return writtenRows.stream()
+ .map(row -> (InternalRow)
GenericRow.of(row.getString(0)))
+ .iterator();
+ }
+ });
+ List<CommitMessage> messages = write.prepareCommit();
+ setFirstRowId(messages, 0L);
+ commit.commit(messages);
+ }
+ }
+
+ FileStoreTable latest = getTable(identifier);
+ RowType readType =
+ latest.rowType()
+ .project(
+ merged
+ ? new String[] {"f2", "pt", "f1"}
+ : new String[] {"pt", "f1"});
+ Predicate filter =
+ new PredicateBuilder(latest.rowType()).equal(1,
BinaryString.fromString(f1(50)));
+ for (List<Range> ranges : Arrays.<List<Range>>asList(null,
fullRowRanges())) {
+ ReadBuilder readBuilder =
+
latest.newReadBuilder().withReadType(readType).withFilter(filter);
+ if (ranges != null) {
+ readBuilder.withRowRanges(ranges);
+ }
+ List<InternalRow> rows =
+ collect(
+ readBuilder.newRead().executeFilter(),
+ readBuilder.newScan().plan(),
+ readType);
+
+ InternalRow row = assertSingleRow(rows);
+
assertThat(row.getString(readType.getFieldIndex("pt")).toString()).isEqualTo("p1");
+
assertThat(row.getString(readType.getFieldIndex("f1")).toString()).isEqualTo(f1(50));
+ if (merged) {
+ assertThat(row.getString(0).toString()).isEqualTo(f1(50));
+ }
+ }
+
+ RowType partitionOnly = latest.rowType().project("pt");
+ ReadBuilder partitionReadBuilder =
latest.newReadBuilder().withReadType(partitionOnly);
+ List<InternalRow> partitions =
+ collect(
+ partitionReadBuilder.newRead(),
+ partitionReadBuilder.newScan().plan(),
+ partitionOnly);
+ assertThat(partitions).hasSize(ROW_COUNT);
+ assertThat(partitions)
+ .allSatisfy(row ->
assertThat(row.getString(0).toString()).isEqualTo("p1"));
+ }
+
+ // Deleting the data files proves the rows come from the row sidecars:
without them the read
+ // would fail on the missing files.
+ @Test
+ public void testMergedGroupFinalSelectionChoosesRowSidecar() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("merged_bitmap_dv_row_sidecar",
options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter,
null)), 50);
+ }
+
+ @Test
+ public void testSingleFileFinalSelectionChoosesRowSidecar() throws
Exception {
+ Map<String, String> options = bitmapOptions("f1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("single_bitmap_dv_row_sidecar",
options);
+ writeAllColumns(table, ROW_COUNT);
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter,
null)), 50);
+ }
+
+ @Test
+ public void testMergedGroupFinalSelectionChoosesRowSidecarWithRowRanges()
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table =
createTable("merged_bitmap_dv_row_sidecar_ranges", options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter,
fullRowRanges())), 50);
+ }
+
+ @Test
+ public void testSingleFileFinalSelectionChoosesRowSidecarWithRowRanges()
throws Exception {
+ Map<String, String> options = bitmapOptions("f1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table =
createTable("single_bitmap_dv_row_sidecar_ranges", options);
+ writeAllColumns(table, ROW_COUNT);
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter,
fullRowRanges())), 50);
+ }
+
+ private static List<Range> fullRowRanges() {
+ return Collections.singletonList(new Range(0L, ROW_COUNT - 1L));
+ }
+
+ // The f0/f1 file is read from its row sidecar (its data file is deleted),
the f2 file has no
+ // sidecar and is read from its data file.
+ @Test
+ public void testMergedGroupMixedSidecarAndDataFileRead() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("merged_mixed_sidecar_data_file",
options);
+ Map<String, String> withoutSidecar = new HashMap<>();
+
withoutSidecar.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"false");
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
withoutSidecar);
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ assertRow(
+ assertSingleRow(
+ readAfterDeletingDataFiles(
+ table,
+ filter,
+ null,
+ file ->
+ file.writeCols() != null
+ &&
file.writeCols().contains("f1"))),
+ 50);
+ }
+
+ // The first file is read from its row sidecar (its data file is deleted),
the second file has
+ // no sidecar and is read from its data file.
+ @Test
+ public void testSingleFileMixedSidecarAndDataFileRead() throws Exception {
+ Map<String, String> options = bitmapOptions("f1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("single_mixed_sidecar_data_file",
options);
+ writeAllColumns(table, ROW_COUNT);
+ Map<String, String> withoutSidecar = new
HashMap<>(bitmapOptions("f1"));
+
withoutSidecar.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"false");
+
writeAllColumns(getTable(identifier(table.name())).copy(withoutSidecar),
ROW_COUNT);
+ // deleting the overlapping rows leaves the first file with a
single-row bitmap selection,
+ // sparse enough to be read from its sidecar, while the second file
has no sidecar at all
+ deleteRows(table, 51, 52, 53, 54, 55);
+
+ Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55);
+ List<InternalRow> rows =
+ readAfterDeletingDataFiles(
+ table,
+ filter,
+ null,
+ file -> file.firstRowId() != null && file.firstRowId()
== 0L);
+ // row 50 of the first file comes from its sidecar, rows 50-55 of the
second file come
+ // from its data file
+ assertThat(rows).hasSize(7);
+ for (InternalRow row : rows) {
+ int index = row.getInt(0);
+ assertThat(index).isBetween(50, 55);
+ assertThat(row.getString(1).toString()).isEqualTo(f1(index));
+ assertThat(row.getString(2).toString()).isEqualTo(f2(index));
+ }
+
+ // the same mixed read with row tracking: the row id of the sidecar
row comes from the
+ // manifest entry, the row ids of the data file rows from its own range
+ RowType readType =
+ rowTypeWithRowId(getTable(identifier(table.name())).rowType())
+ .project(SpecialFields.ROW_ID.name(), "f0", "f1");
+ List<InternalRow> trackedRows = readWithFilter(table, filter,
readType);
+ assertThat(trackedRows).hasSize(7);
+ assertThat(trackedRows.stream().map(row -> row.getLong(0)))
+ .containsExactlyInAnyOrder(50L, 150L, 151L, 152L, 153L, 154L,
155L);
+ }
+
+ // Ordinary row-format data files use the same physical schema mapping as
row sidecars. The
+ // row format has no format-level filter pushdown, so the bitmap index
provides the selection.
+ @Test
+ public void testOrdinaryRowFormatDataFile() throws Exception {
+ Map<String, String> options = bitmapOptions("f1");
+ options.put(CoreOptions.FILE_FORMAT.key(), "row");
+ FileStoreTable table = createTable("ordinary_row_format_file",
options);
+ writeAllColumns(table, ROW_COUNT);
+
+ assertRow(assertSingleRow(readWithFilter(table, equalF1(f1(50)),
null)), 50);
+
+ RowType readType =
+
rowTypeWithRowId(table.rowType()).project(SpecialFields.ROW_ID.name(), "f1");
+ List<InternalRow> rows = readWithFilter(table, equalF1(f1(50)),
readType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getLong(0)).isEqualTo(50L);
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo(f1(50));
+ }
+
+ @Test
+ public void testFormatReaderCacheKeyIncludesPhysicalSchema() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ FileStoreTable table = createTable("row_sidecar_physical_schema_key",
options);
+ writeSplitColumns(table, ROW_COUNT, Collections.emptyMap(),
bitmapOptions("f2"));
+ writeSecondSplitColumns(table, ROW_COUNT, bitmapOptions("f2"));
+
+ FileStoreTable latest = getTable(identifier(table.name()));
+ RowType projection = latest.rowType().project(Arrays.asList("f1",
"f2"));
+ List<InternalRow> rows = readWithFilter(table, equalF2(f2(50)),
projection);
+
+ assertThat(rows).hasSize(2);
+ assertThat(rows)
+ .extracting(row -> row.getString(0).toString())
+ .containsExactlyInAnyOrder(f1(50), "second-" + f1(50));
+ assertThat(rows).extracting(row ->
row.getString(1).toString()).containsOnly(f2(50));
+ }
+
+ @Test
+ public void testBitmapFilterWithWideWriteSchemaAndRowSidecar() throws
Exception {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ for (int i = 0; i < 7; i++) {
+ schemaBuilder.column("c" + i, DataTypes.INT());
+ }
+ schemaBuilder
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true");
+ bitmapOptions("c6").forEach(schemaBuilder::option);
+ Identifier identifier = identifier("wide_schema_row_sidecar");
+ catalog.createTable(identifier, schemaBuilder.build(), false);
+ FileStoreTable table = getTable(identifier);
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ write.write(GenericRow.of(i, i, i, i, i, i, i));
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ Predicate filter = new PredicateBuilder(table.rowType()).equal(6, 50);
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(filter);
+ List<InternalRow> rows =
+ collect(
+ readBuilder.newRead().executeFilter(),
+ readBuilder.newScan().plan(),
+ table.rowType());
+
+ InternalRow row = assertSingleRow(rows);
+ for (int i = 0; i < 7; i++) {
+ assertThat(row.getInt(i)).isEqualTo(50);
+ }
+
+ ReadBuilder rangeReadBuilder =
+ table.newReadBuilder()
+ .withFilter(filter)
+ .withRowRanges(Collections.singletonList(new Range(0L,
ROW_COUNT - 1L)));
+ rows =
+ collect(
+ rangeReadBuilder.newRead().executeFilter(),
+ rangeReadBuilder.newScan().plan(),
+ table.rowType());
+ row = assertSingleRow(rows);
+ for (int i = 0; i < 7; i++) {
+ assertThat(row.getInt(i)).isEqualTo(50);
+ }
+
+ RowType rowIdType = rowTypeWithRowId(table.rowType());
+ PredicateBuilder rowIdBuilder = new PredicateBuilder(rowIdType);
+ Predicate rowIdFilter =
+ PredicateBuilder.and(
+ rowIdBuilder.between(7, 0L, ROW_COUNT - 1L),
rowIdBuilder.equal(6, 50));
+ ReadBuilder rowIdReadBuilder =
+
table.newReadBuilder().withReadType(rowIdType).withFilter(rowIdFilter);
+ rows =
+ collect(
+ rowIdReadBuilder.newRead().executeFilter(),
+ rowIdReadBuilder.newScan().plan(),
+ rowIdType);
+ row = assertSingleRow(rows);
+ for (int i = 0; i < 7; i++) {
+ assertThat(row.getInt(i)).isEqualTo(50);
+ }
+ assertThat(row.getLong(7)).isEqualTo(50L);
+ }
+
@Test
public void testSingleFileSkippedByFileIndex() throws Exception {
// a standalone .index file, only the reader can evaluate it
@@ -1048,6 +1531,35 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
}
}
+ /** Writes f1 and f2 into a second merged group with a different physical
schema. */
+ private void writeSecondSplitColumns(
+ FileStoreTable table, int count, Map<String, String>
secondOptions) throws Exception {
+ FileStoreTable latest = getTable(identifier(table.name()));
+ RowType writeType0 =
latest.rowType().project(Collections.singletonList("f1"));
+ BatchWriteBuilder builder = latest.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType0);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < count; i++) {
+ write.write(GenericRow.of(BinaryString.fromString("second-" +
f1(i))));
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ latest = getTable(identifier(table.name()));
+ long firstRowId =
latest.snapshotManager().latestSnapshot().nextRowId() - count;
+ RowType writeType1 =
latest.rowType().project(Collections.singletonList("f2"));
+ builder = latest.copy(secondOptions).newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType1);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < count; i++) {
+ write.write(GenericRow.of(BinaryString.fromString(f2(i))));
+ }
+ List<CommitMessage> commitables = write.prepareCommit();
+ setSingleFileFirstRowId(commitables, firstRowId);
+ commit.commit(commitables);
+ }
+ }
+
/**
* Writes f0/f1 (a*) with a bloom index on f1, then overwrites f1 with c*
in a second file that
* shares the same row id range and has a higher sequence number, so the
column merge takes f1
@@ -1097,6 +1609,53 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
return readWithFilter(table, predicate, null);
}
+ private List<InternalRow> readAfterDeletingDataFiles(
+ FileStoreTable table, Predicate predicate, @Nullable List<Range>
rowRanges)
+ throws Exception {
+ return readAfterDeletingDataFiles(table, predicate, rowRanges, null);
+ }
+
+ /**
+ * Reads after deleting the data files. A non-null delete filter restricts
the deletion to the
+ * matching files, so part of the split is read from the row sidecars and
the rest from the
+ * remaining data files.
+ */
+ private List<InternalRow> readAfterDeletingDataFiles(
+ FileStoreTable table,
+ Predicate predicate,
+ @Nullable List<Range> rowRanges,
+ @Nullable java.util.function.Predicate<DataFileMeta> deleteFilter)
+ throws Exception {
+ FileStoreTable latest = getTable(identifier(table.name()));
+ ReadBuilder readBuilder =
latest.newReadBuilder().withFilter(predicate);
+ if (rowRanges != null) {
+ readBuilder.withRowRanges(rowRanges);
+ }
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ // with row ranges the scan wraps the split into an IndexedSplit
+ Split first = plan.splits().get(0);
+ DataSplit split =
+ first instanceof IndexedSplit
+ ? ((IndexedSplit) first).dataSplit()
+ : (DataSplit) first;
+ DataFilePathFactory pathFactory =
+ latest.store()
+ .pathFactory()
+ .createDataFilePathFactory(split.partition(),
split.bucket());
+ for (DataFileMeta file : split.dataFiles()) {
+ if (deleteFilter == null || deleteFilter.test(file)) {
+ assertThat(latest.fileIO().delete(pathFactory.toPath(file),
false)).isTrue();
+ }
+ }
+
+ List<InternalRow> rows = new ArrayList<>();
+ InternalRowSerializer serializer = new
InternalRowSerializer(latest.rowType());
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> rows.add(serializer.copy(row)));
+ }
+ return rows;
+ }
+
private List<InternalRow> readWithFilter(
FileStoreTable table, Predicate predicate, @Nullable RowType
readType)
throws Exception {
@@ -1252,6 +1811,14 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
return new PredicateBuilder(rowType()).equal(1,
BinaryString.fromString(value));
}
+ private static Predicate equalAnyF1(int... values) {
+ Predicate[] predicates = new Predicate[values.length];
+ for (int i = 0; i < values.length; i++) {
+ predicates[i] = equalF1(f1(values[i]));
+ }
+ return PredicateBuilder.or(predicates);
+ }
+
private static Predicate equalF2(String value) {
return new PredicateBuilder(rowType()).equal(2,
BinaryString.fromString(value));
}
