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 bf8ddb9312 [core] Push filters into merged data evolution groups by
file statistics (#10394)
bf8ddb9312 is described below
commit bf8ddb93129f08ccdf3e09ab80e0b8c150c17b38
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Fri Oct 16 21:58:09 2026 +0800
[core] Push filters into merged data evolution groups by file statistics
(#10394)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 16 +
.../apache/paimon/format/FileMetadataCache.java | 47 ++
.../apache/paimon/format/FormatReaderContext.java | 18 +
.../apache/paimon/format/FormatReaderFactory.java | 19 +
.../paimon/operation/DataEvolutionSplitRead.java | 297 ++++++-
.../DataEvolutionStatsPushDownScenarioTest.java | 885 +++++++++++++++++++++
.../table/DataEvolutionStatsPushDownTest.java | 627 +++++++++++++++
.../format/parquet/ParquetReaderFactory.java | 57 +-
.../apache/parquet/hadoop/ParquetFileReader.java | 34 +
.../parquet/ParquetCandidateRowRangesTest.java | 216 +++++
11 files changed, 2190 insertions(+), 32 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index d0ad00368d..601ed2a358 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -548,6 +548,12 @@ under the License.
<td>Boolean</td>
<td>Whether to persist source when process merge into action on
data evolution table.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.merged-read.stats-pushdown.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to push filters down into a row-id range whose columns
are merged from several data files. The rows which may match are computed from
the file statistics and page indexes of the files holding the latest values of
the filtered columns, then every file of the range reads only these rows. Only
Parquet files are supported.</td>
+ </tr>
<tr>
<td><h5>data-evolution.nested-field.enabled</h5></td>
<td style="word-wrap: break-word;">false</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 3bca19e557..4e8a1cec46 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2689,6 +2689,18 @@ public class CoreOptions implements Serializable {
+ "selected row count is no more than "
+
"data-evolution.row-sidecar.max-selected-rows.");
+ public static final ConfigOption<Boolean>
DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED =
+ key("data-evolution.merged-read.stats-pushdown.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to push filters down into a row-id range
whose columns are "
+ + "merged from several data files. The
rows which may match "
+ + "are computed from the file statistics
and page indexes of "
+ + "the files holding the latest values of
the filtered "
+ + "columns, then every file of the range
reads only these "
+ + "rows. Only Parquet files are
supported.");
+
public static final ConfigOption<Boolean>
DATA_EVOLUTION_MERGE_INTO_FILE_PRUNING =
key("data-evolution.merge-into.file-pruning")
.booleanType()
@@ -4644,6 +4656,10 @@ public class CoreOptions implements Serializable {
return maxSelectedRows;
}
+ public boolean dataEvolutionMergedReadStatsPushdownEnabled() {
+ return options.get(DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED);
+ }
+
public double dataEvolutionRowSidecarMaxSelectionRatio() {
double maxSelectionRatio =
options.get(DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO);
checkArgument(
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/FileMetadataCache.java
b/paimon-common/src/main/java/org/apache/paimon/format/FileMetadataCache.java
new file mode 100644
index 0000000000..af711b607c
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/format/FileMetadataCache.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.format;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOFunction;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Format specific file metadata, such as a Parquet footer, shared by the
readers of the same files
+ * so that the metadata is read only once. The metadata of a file is cached by
its path, so all
+ * users of a cache must read the same kind of metadata of a file with the
same options. Not thread
+ * safe.
+ */
+public class FileMetadataCache {
+
+ private final Map<Path, Object> metadata = new HashMap<>();
+
+ @SuppressWarnings("unchecked")
+ public <T> T getOrLoad(Path path, IOFunction<Path, T> loader) throws
IOException {
+ Object value = metadata.get(path);
+ if (value == null) {
+ value = loader.apply(path);
+ metadata.put(path, value);
+ }
+ return (T) value;
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java
b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java
index f92741517e..63ea7b9f2a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java
@@ -34,6 +34,7 @@ public class FormatReaderContext implements
FormatReaderFactory.Context {
private final long fileSize;
@Nullable private final RoaringBitmap32 selection;
@Nullable private final ReadBatchSizer readBatchSizer;
+ @Nullable private final FileMetadataCache metadataCache;
public FormatReaderContext(
FileIO fileIO,
@@ -41,11 +42,22 @@ public class FormatReaderContext implements
FormatReaderFactory.Context {
long fileSize,
@Nullable RoaringBitmap32 selection,
@Nullable ReadBatchSizer readBatchSizer) {
+ this(fileIO, file, fileSize, selection, readBatchSizer, null);
+ }
+
+ public FormatReaderContext(
+ FileIO fileIO,
+ Path file,
+ long fileSize,
+ @Nullable RoaringBitmap32 selection,
+ @Nullable ReadBatchSizer readBatchSizer,
+ @Nullable FileMetadataCache metadataCache) {
this.fileIO = fileIO;
this.file = file;
this.fileSize = fileSize;
this.selection = selection;
this.readBatchSizer = readBatchSizer;
+ this.metadataCache = metadataCache;
}
@Override
@@ -74,4 +86,10 @@ public class FormatReaderContext implements
FormatReaderFactory.Context {
public ReadBatchSizer readBatchSizer() {
return readBatchSizer;
}
+
+ @Nullable
+ @Override
+ public FileMetadataCache metadataCache() {
+ return metadataCache;
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java
b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java
index 1fc80d1aea..ccb1304096 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java
@@ -24,11 +24,13 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.ReadBatchSizer;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringBitmap32;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.List;
/** A factory to create {@link RecordReader} for file. */
public interface FormatReaderFactory {
@@ -43,6 +45,17 @@ public interface FormatReaderFactory {
getClass().getName()));
}
+ /**
+ * Row positions of the file which may satisfy the pushed down filters,
computed from file
+ * metadata only, without reading any data. The ranges are sorted, not
overlapping and relative
+ * to the start of the file. Returns null when the format can not prune
rows this way, callers
+ * then read the whole file.
+ */
+ @Nullable
+ default List<Range> candidateRowRanges(Context context) throws IOException
{
+ return null;
+ }
+
/** Context for creating reader. */
interface Context {
@@ -60,5 +73,11 @@ public interface FormatReaderFactory {
default ReadBatchSizer readBatchSizer() {
return null;
}
+
+ /** Cache of file metadata already read for the same file, null when
not shared. */
+ @Nullable
+ default FileMetadataCache metadataCache() {
+ return null;
+ }
}
}
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 783a763369..6a1a2cc94d 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
@@ -34,6 +34,7 @@ import org.apache.paimon.fileindex.FileIndexResult;
import org.apache.paimon.fileindex.bitmap.ApplyBitmapIndexRecordReader;
import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.format.FileFormatDiscover;
+import org.apache.paimon.format.FileMetadataCache;
import org.apache.paimon.format.FormatKey;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.fs.FileIO;
@@ -72,6 +73,9 @@ import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RangeHelper;
import org.apache.paimon.utils.RoaringBitmap32;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.IOException;
@@ -106,10 +110,12 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
* A union {@link SplitRead} to read multiple inner files to merge columns.
*
* <p>Filters can only be pushed down where they can not interfere with the
column merging: a file
- * read without merging gets both the file index and the format level push
down, while an eligible
- * merged group shares one bitmap selection across all aligned field readers.
Other merged groups
- * only use the file index to skip the whole group, as dropping rows in one
reader would break the
- * positional alignment between them.
+ * read without merging gets both the file index and the format level push
down. Dropping rows in
+ * only one of the merged readers of a group would break the positional
alignment between them, so
+ * an eligible merged group instead shares one selection across all its
aligned field readers: the
+ * rows of the bitmap file indexes, and the rows which may match according to
the file metadata of
+ * the files winning the filtered fields, see {@link #candidateRanges}. Other
merged groups only use
+ * the file index to skip the whole group.
*
* <p>Only a filter whose every field belongs to the read type is pushed down,
see {@link
* #readTypeFilters}, and only to the files that wrote those fields, see
{@link #fileFilters}.
@@ -118,6 +124,8 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
*/
public class DataEvolutionSplitRead implements SplitRead<InternalRow> {
+ private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionSplitRead.class);
+
private static final String ROW_SIDECAR_FORMAT = "row";
private final FileIO fileIO;
@@ -128,9 +136,12 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
// Kept apart from formatReaderMappings: the single file path pushes per
file filters into the
// mapping, so it must not share entries with the merge path, which pushes
none.
private final Map<SingleFileKey, FormatReaderMapping>
singleFileReaderMappings;
+ // Mappings computing candidate rows of a merged group, they carry the
filters as well.
+ private final Map<SingleFileKey, FormatReaderMapping>
candidateReaderMappings;
private final Function<Long, TableSchema> schemaFetcher;
private final CoreOptions coreOptions;
private final boolean fileIndexReadEnabled;
+ private final boolean statsPushDownEnabled;
protected RowType readRowType;
@Nullable private List<Predicate> filters;
@@ -153,7 +164,9 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
this.pathFactory = pathFactory;
this.formatReaderMappings = new HashMap<>();
this.singleFileReaderMappings = new HashMap<>();
+ this.candidateReaderMappings = new HashMap<>();
this.fileIndexReadEnabled = coreOptions.fileIndexReadEnabled();
+ this.statsPushDownEnabled =
coreOptions.dataEvolutionMergedReadStatsPushdownEnabled();
this.readRowType = rowType;
}
@@ -180,6 +193,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
// the single file mappings carry the filters they were built
with, and a read can be
// reconfigured after it created readers, see
AppendTableRead#innerWithFilter
singleFileReaderMappings.clear();
+ candidateReaderMappings.clear();
}
return this;
}
@@ -319,6 +333,27 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
groupSelection =
buildGroupSelection(needMergeFiles,
fileIndexResults);
}
+ FileMetadataCache metadataCache = new
FileMetadataCache();
+ List<Range> candidates =
+ candidateRanges(
+ filters,
+ needMergeFiles,
+ dataFilePathFactory,
+ metadataCache);
+ // Candidates restrict the group selection, which
every file of the
+ // group reads.
+ if (candidates != null) {
+ if (candidates.isEmpty()) {
+ return new EmptyFileRecordReader<>();
+ }
+ groupSelection =
+ restrictToCandidates(
+ groupSelection,
needMergeFiles.get(0), candidates);
+ if (isSelectionEmpty(
+ needMergeFiles.get(0), rowRanges,
groupSelection)) {
+ return new EmptyFileRecordReader<>();
+ }
+ }
return createUnionReader(
needMergeFiles,
partition,
@@ -326,7 +361,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
});
}
}
@@ -356,7 +392,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable BitmapIndexResult groupSelection)
+ @Nullable BitmapIndexResult groupSelection,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
List<DataEvolutionVectorReadPlanner.ReadRange> vectorRanges =
DataEvolutionVectorReadPlanner.plan(
@@ -403,6 +440,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
readRanges,
readRowType,
deletionVector,
+ null,
null));
}
return ConcatRecordReader.create(suppliers);
@@ -428,7 +466,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
}
private RecordReader<InternalRow> createUnionReader(
@@ -439,7 +478,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable BitmapIndexResult groupSelection)
+ @Nullable BitmapIndexResult groupSelection,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
long rowCount = fieldsFiles.get(0).rowCount();
@@ -489,7 +529,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
}
// Build the per-bunch readers from the planned partial read row types.
@@ -533,7 +574,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
partialReadRowType,
deletionVector,
- groupSelection));
+ groupSelection,
+ metadataCache));
}
return nestedFieldEnabled
@@ -552,7 +594,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable BitmapIndexResult groupSelection)
+ @Nullable BitmapIndexResult groupSelection,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
DataFileMeta firstFile = bunch.files().get(0);
FileReadTarget readTarget =
@@ -577,7 +620,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
}
private boolean nestedFieldEnabledFor(List<DataFileMeta> files) {
@@ -643,7 +687,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable BitmapIndexResult groupSelection)
+ @Nullable BitmapIndexResult groupSelection,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
if (bunch instanceof DataBunch) {
// for data bunch, directly read the single file
@@ -655,7 +700,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
} else if (bunch instanceof VectorFileBunch) {
// for vector bunch, sequential read all data files and concat them
return sequentialReadFiles(
@@ -717,7 +763,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
dataFilePathFactory.toPath(file),
file.fileSize()),
deletionVector,
- groupSelection));
+ groupSelection,
+ null));
}
return ConcatRecordReader.create(readerSuppliers);
}
@@ -822,7 +869,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
readRowType,
readTarget,
deletionVector,
- fileIndexResult);
+ fileIndexResult,
+ null);
}
private FileRecordReader<InternalRow> createFileReader(
@@ -842,6 +890,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
rowRanges,
readRowType,
deletionVector,
+ null,
null);
}
@@ -853,7 +902,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<Range> rowRanges,
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable BitmapIndexResult groupSelection)
+ @Nullable BitmapIndexResult groupSelection,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
return createFileReader(
partition,
@@ -863,7 +913,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
readRowType,
readTarget(file, dataFilePathFactory, rowRanges,
groupSelection),
deletionVector,
- groupSelection);
+ groupSelection,
+ metadataCache);
}
private FileRecordReader<InternalRow> createFileReader(
@@ -874,7 +925,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
RowType readRowType,
FileReadTarget readTarget,
@Nullable DeletionVectorWithRange deletionVector,
- @Nullable FileIndexResult fileIndexResult)
+ @Nullable FileIndexResult fileIndexResult,
+ @Nullable FileMetadataCache metadataCache)
throws IOException {
RoaringBitmap32 selection = file.toFileSelection(rowRanges);
BitmapIndexResult bitmapIndexResult =
@@ -894,7 +946,12 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FormatReaderContext formatReaderContext =
new FormatReaderContext(
- fileIO, readTarget.path, readTarget.fileSize,
selection, readBatchSizer);
+ fileIO,
+ readTarget.path,
+ readTarget.fileSize,
+ selection,
+ readBatchSizer,
+ metadataCache);
FileRecordReader<InternalRow> fileRecordReader =
new DataFileRecordReader(
readRowType,
@@ -1019,9 +1076,9 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
*
* <p>Only normal data files are considered. Blob and vector-store files
can cover only a subset
* of the group range, so their indexes cannot prove that the whole group
has no match. Files
- * are visited newest first; {@code claimedFieldIds} prevents an older
copy of an overwritten
- * field from vetoing the group. The returned results are retained so a
later deletion-vector
- * pass can intersect them without reopening file-index sidecars.
+ * are visited newest first; only the fields a file wins may veto, see
{@link #winningFieldIds},
+ * so an older copy of an overwritten field can not. The returned results
are retained so a
+ * later deletion-vector pass can intersect them without reopening
file-index sidecars.
*/
private List<FileIndexResultEntry> evaluateFileIndexes(
@Nullable List<Predicate> filters,
@@ -1033,18 +1090,14 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
List<FileIndexResultEntry> results = new ArrayList<>();
- Set<Integer> claimedFieldIds = new HashSet<>();
- for (DataFileMeta file : files) {
- if (isBlobFile(file.fileName()) ||
isVectorStoreFile(file.fileName())) {
- continue;
- }
-
+ for (Pair<DataFileMeta, Set<Integer>> winning :
winningFieldIds(files)) {
+ DataFileMeta file = winning.getKey();
TableSchema dataSchema =
schemaFetcher.apply(file.schemaId()).dataFileSchema(file.writeCols());
// columns this file wrote but a newer file already won: their
values here are stale
Set<String> overwrittenCols = new HashSet<>();
for (DataField field : dataSchema.fields()) {
- if (!claimedFieldIds.add(field.id())) {
+ if (!winning.getValue().contains(field.id())) {
overwrittenCols.add(field.name());
}
}
@@ -1067,6 +1120,171 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return results;
}
+ /**
+ * The normal files of a merged group, newest first, each with the ids of
the fields it wins,
+ * those no newer file of the group wrote. Blob and vector-store files are
skipped, they can
+ * cover only a subset of the group range.
+ */
+ private List<Pair<DataFileMeta, Set<Integer>>>
winningFieldIds(List<DataFileMeta> files) {
+ List<Pair<DataFileMeta, Set<Integer>>> result = new ArrayList<>();
+ Set<Integer> claimedFieldIds = new HashSet<>();
+ for (DataFileMeta file : files) {
+ if (isBlobFile(file.fileName()) ||
isVectorStoreFile(file.fileName())) {
+ continue;
+ }
+ Set<Integer> won = new HashSet<>();
+ for (DataField field :
+ schemaFetcher
+ .apply(file.schemaId())
+ .dataFileSchema(file.writeCols())
+ .fields()) {
+ if (claimedFieldIds.add(field.id())) {
+ won.add(field.id());
+ }
+ }
+ result.add(Pair.of(file, won));
+ }
+ return result;
+ }
+
+ /**
+ * Row id ranges of a merged group which may satisfy the filters, or null
when none can be
+ * excluded.
+ *
+ * <p>Each filter is evaluated against the metadata of the single file
winning all of its
+ * fields, such as Parquet row group statistics and page indexes: an older
copy of a field is
+ * stale and would exclude rows that do match. The ranges of the files are
intersected and added
+ * to the selection shared by all files of the group, see {@link
#restrictToCandidates}, so
+ * every file reads only these rows, which keeps the readers aligned.
+ *
+ * <p>Only a group of normal files sharing one row id range qualifies. The
file metadata read
+ * here is cached in {@code metadataCache}, which the caller passes to the
readers of the group
+ * so that they do not read it again.
+ */
+ @Nullable
+ private List<Range> candidateRanges(
+ @Nullable List<Predicate> filters,
+ List<DataFileMeta> files,
+ DataFilePathFactory dataFilePathFactory,
+ FileMetadataCache metadataCache) {
+ try {
+ return computeCandidateRanges(filters, files, dataFilePathFactory,
metadataCache);
+ } catch (Exception e) {
+ // Pruning is only an optimization, any failure reads the whole
group. Opening the files
+ // then reports a lost or corrupt one, respecting
scan.ignore-lost-file and
+ // scan.ignore-corrupt-file, which the exceptions of the format
are not meant for.
+ LOG.warn(
+ "Failed to compute candidate rows of the group {}, reading
the whole group.",
+ files.get(0).nonNullRowIdRange(),
+ e);
+ return null;
+ }
+ }
+
+ @Nullable
+ private List<Range> computeCandidateRanges(
+ @Nullable List<Predicate> filters,
+ List<DataFileMeta> files,
+ DataFilePathFactory dataFilePathFactory,
+ FileMetadataCache metadataCache)
+ throws IOException {
+ if (!statsPushDownEnabled
+ || isNullOrEmpty(filters)
+ || !isNormalGroup(files)
+ || nestedFieldEnabledFor(files)) {
+ return null;
+ }
+
+ List<Range> candidates = null;
+ for (Pair<DataFileMeta, Set<Integer>> winning :
winningFieldIds(files)) {
+ DataFileMeta file = winning.getKey();
+ Set<String> wonFields = new HashSet<>();
+ for (DataField field : schema.fields()) {
+ if (winning.getValue().contains(field.id())) {
+ wonFields.add(field.name());
+ }
+ }
+ List<Predicate> fileFilters = filtersWithin(filters, wonFields);
+ if (isNullOrEmpty(fileFilters)) {
+ continue;
+ }
+
+ Set<String> filterFields = new HashSet<>();
+ fileFilters.forEach(filter ->
filterFields.addAll(collectFieldNames(filter)));
+ RowType readType =
+ new RowType(
+ schema.fields().stream()
+ .filter(field ->
filterFields.contains(field.name()))
+ .collect(Collectors.toList()));
+ String formatIdentifier =
DataFilePathFactory.formatIdentifier(file.fileName());
+ // The filters of the mapping are those of this read whose fields
are all within the
+ // read type. They are determined by the key: the read type is the
union of the fields
+ // of the file filters, and any filter within it is also within
the fields the file
+ // wins, so it is one of the file filters, whatever read type the
split was planned
+ // with. withFilter clears the mappings when the filters change.
+ FormatReaderMapping mapping =
+ candidateReaderMappings.computeIfAbsent(
+ new SingleFileKey(
+ file.schemaId(),
+ formatIdentifier,
+ file.writeCols(),
+ readType,
+ false),
+ key ->
+ formatBuilder(readType, fileFilters, false)
+ .build(
+ formatIdentifier,
+ schema,
+ schemaFetcher
+
.apply(file.schemaId())
+
.dataFileSchema(file.writeCols())));
+ List<Range> fileRanges =
+ mapping.getReaderFactory()
+ .candidateRowRanges(
+ new FormatReaderContext(
+ fileIO,
+ dataFilePathFactory.toPath(file),
+ file.fileSize(),
+ null,
+ null,
+ metadataCache));
+ if (fileRanges == null) {
+ continue;
+ }
+
+ long firstRowId = file.nonNullFirstRowId();
+ List<Range> rowIdRanges = new ArrayList<>(fileRanges.size());
+ for (Range range : fileRanges) {
+ rowIdRanges.add(new Range(firstRowId + range.from, firstRowId
+ range.to));
+ }
+ rowIdRanges = Range.sortAndMergeOverlap(rowIdRanges, true);
+ candidates = candidates == null ? rowIdRanges :
Range.and(candidates, rowIdRanges);
+ if (candidates.isEmpty()) {
+ return candidates;
+ }
+ }
+
+ if (candidates == null
+ || (candidates.size() == 1
+ &&
candidates.get(0).equals(files.get(0).nonNullRowIdRange()))) {
+ return null;
+ }
+ return candidates;
+ }
+
+ /** Whether all files are normal data files of the same row id range. */
+ private static boolean isNormalGroup(List<DataFileMeta> files) {
+ Range range = files.get(0).nonNullRowIdRange();
+ for (DataFileMeta file : files) {
+ if (isBlobFile(file.fileName())
+ || isVectorStoreFile(file.fileName())
+ || !range.equals(file.nonNullRowIdRange())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Applies a group deletion vector to already evaluated file-index results.
*
@@ -1122,6 +1340,25 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return selection;
}
+ /**
+ * Restricts the selection shared by the files of a merged group to its
candidate rows. The
+ * files share one row id range, so a row has the same position in all of
them.
+ */
+ private static BitmapIndexResult restrictToCandidates(
+ @Nullable BitmapIndexResult selection, DataFileMeta file,
List<Range> candidates) {
+ long firstRowId = file.nonNullFirstRowId();
+ RoaringBitmap32 positions = new RoaringBitmap32();
+ for (Range range : candidates) {
+ positions.or(
+ RoaringBitmap32.bitmapOfRange(
+ range.from - firstRowId, range.to - firstRowId +
1));
+ }
+ BitmapIndexResult candidateSelection = new BitmapIndexResult(() ->
positions);
+ return selection == null
+ ? candidateSelection
+ : (BitmapIndexResult) selection.and(candidateSelection);
+ }
+
/** Returns whether exact bitmap candidates and the requested row ranges
are disjoint. */
private static boolean isSelectionEmpty(
DataFileMeta file,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownScenarioTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownScenarioTest.java
new file mode 100644
index 0000000000..11896601db
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownScenarioTest.java
@@ -0,0 +1,885 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
+import org.apache.paimon.fileindex.FileIndexOptions;
+import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+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.InnerTableRead;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InternalRowUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static
org.apache.paimon.CoreOptions.DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Scenarios of pushing filters into merged data evolution groups by file
statistics. Every read is
+ * compared with the same read without the pushdown and with an in-memory
model of the table.
+ */
+public class DataEvolutionStatsPushDownScenarioTest extends TableTestBase {
+
+ private static final int RANGE_ROWS = 3000;
+
+ // ----------------------------------------------------------------
scenarios
+
+ /** Several row id ranges, so the candidates of the later ones do not
start at row id 0. */
+ @Test
+ public void testOnlyLaterRangeUpdated() throws Exception {
+ Model m = create("later_range", "parquet", false,
Collections.emptyMap());
+ for (int r = 0; r < 3; r++) {
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ }
+ // only the last range can match, at its end, far from row id 0
+ update(
+ m,
+ 2,
+ Collections.singletonList("a"),
+ (row, col) -> 100_000 + (int) (row - 2 * RANGE_ROWS));
+
+ Result result =
+ assertMatchesModel(
+ m, builder(m).greaterOrEqual(index(m, "a"), 100_000 +
RANGE_ROWS - 100));
+ assertThat(result.matched).hasSize(100);
+ assertThat(result.returned).isLessThan(RANGE_ROWS / 4);
+ }
+
+ /** The candidates of the statistics and the rows of a bitmap index are
intersected. */
+ @Test
+ public void testBitmapIndexAndStatisticsBothPrune() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(
+ FileIndexOptions.FILE_INDEX
+ + "."
+ + BitmapFileIndexFactory.BITMAP_INDEX
+ + "."
+ + CoreOptions.COLUMNS,
+ "b");
+ // as a write option of the base file only: a partial update can not
index a column it
+ // does not write
+ Model m = create("bitmap", "parquet", false, Collections.emptyMap());
+ m.table = m.table.copy(options);
+ writeRange(m, RANGE_ROWS, null, row -> base(row, (int) (row % 10)));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - 1 - row));
+
+ PredicateBuilder builder = builder(m);
+ Predicate filter =
+ PredicateBuilder.and(
+ builder.lessThan(index(m, "a"), 300),
builder.equal(index(m, "b"), 3));
+ Result both = assertMatchesModel(m, filter);
+ assertThat(both.matched).hasSize(30);
+
+ long statsOnly =
+ read(copy(m.table, "file-index.read.enabled", "false"),
filter, null).returned;
+ long bitmapOnly =
+ read(
+ copy(
+ m.table,
+
DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED.key(),
+ "false"),
+ filter,
+ null)
+ .returned;
+ assertThat(both.returned).isLessThan(statsOnly).isLessThan(bitmapOnly);
+ }
+
+ @Test
+ public void testPartitionedTable() throws Exception {
+ Model m = create("partitioned", "parquet", true,
Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, "p0", row -> base(row));
+ writeRange(m, RANGE_ROWS, "p1", row -> base(row));
+ for (int r = 0; r < 2; r++) {
+ update(m, r, Collections.singletonList("a"), (row, col) -> 100_000
+ row.intValue());
+ }
+
+ PredicateBuilder builder = builder(m);
+ int last = 100_000 + 2 * RANGE_ROWS;
+ Result result = assertMatchesModel(m, builder.greaterOrEqual(index(m,
"a"), last - 100));
+ assertThat(result.matched).hasSize(100);
+ assertThat(result.returned).isLessThan(RANGE_ROWS / 4);
+ Result p0 =
+ assertMatchesModel(
+ m,
+ PredicateBuilder.and(
+ builder.equal(index(m, "pt"),
BinaryString.fromString("p0")),
+ builder.greaterThan(index(m, "a"), 100_000 +
RANGE_ROWS - 100)));
+ assertThat(p0.matched).hasSize(99);
+ }
+
+ /** A column added after the base file was written lives only in the
update file. */
+ @Test
+ public void testAddedColumnBackfilled() throws Exception {
+ Model m = create("added", "parquet", false, Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ addColumn(m, "x", DataTypes.INT());
+ // backfill the first range only, the second keeps nulls
+ update(m, 0, Collections.singletonList("x"), (row, col) -> (int)
(RANGE_ROWS - row));
+
+ PredicateBuilder builder = builder(m);
+ Result result = assertMatchesModel(m, builder.lessThan(index(m, "x"),
100));
+ assertThat(result.matched).hasSize(99);
+ // the range which is not backfilled is a single file, read as before
+ assertThat(result.returned).isLessThan(result.baseline - RANGE_ROWS /
2);
+ assertThat(assertMatchesModel(m, builder.isNull(index(m,
"x"))).matched)
+ .hasSize(RANGE_ROWS);
+ }
+
+ @Test
+ public void testWidenedColumnType() throws Exception {
+ Model m = create("widened", "parquet", false, Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - row));
+ alterType(m, "a", DataTypes.BIGINT());
+
+ assertThat(assertMatchesModel(m, builder(m).lessThan(index(m, "a"),
100L)).matched)
+ .hasSize(99);
+ }
+
+ /** Formats without candidate rows and groups the pushdown skips read as
before. */
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "avro"})
+ public void testOtherFormats(String format) throws Exception {
+ Model m = create("format_" + format, format, false,
Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - row));
+
+ assertThat(assertMatchesModel(m, builder(m).lessThan(index(m, "a"),
100)).matched)
+ .hasSize(99);
+ }
+
+ @Test
+ public void testNestedFieldEvolutionTable() throws Exception {
+ Model m =
+ create(
+ "nested",
+ "parquet",
+ false,
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true"));
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - row));
+
+ assertThat(assertMatchesModel(m, builder(m).lessThan(index(m, "a"),
100)).matched)
+ .hasSize(99);
+ }
+
+ /** A group with a blob file is not pruned, and reads as without the
pushdown. */
+ @Test
+ public void testGroupWithBlobFile() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("a", DataTypes.INT())
+ .column("f", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option("parquet.page.size", "1024")
+ .build();
+ catalog.createTable(identifier("blob"), schema, false);
+ FileStoreTable table = getTable(identifier("blob"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < RANGE_ROWS; i++) {
+ write.write(GenericRow.of((long) i, i, new BlobData(new byte[]
{(byte) i})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+
table.rowType().project(Collections.singletonList("a")));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < RANGE_ROWS; i++) {
+ write.write(GenericRow.of(RANGE_ROWS - 1 - i));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, 0);
+ commit.commit(messages);
+ }
+ table = getTable(identifier("blob"));
+
+ Predicate filter = new PredicateBuilder(table.rowType()).lessThan(1,
100);
+ List<String> expected = new ArrayList<>();
+ for (int i = RANGE_ROWS - 100; i < RANGE_ROWS; i++) {
+ expected.add(i + ":" + (RANGE_ROWS - 1 - i) + ":" + (byte) i);
+ }
+ Collections.sort(expected);
+ for (boolean enabled : new boolean[] {true, false}) {
+ ReadBuilder read = pushDown(table,
enabled).newReadBuilder().withFilter(filter);
+ List<String> matched = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
read.newRead().createReader(read.newScan().plan().splits())) {
+ reader.forEachRemaining(
+ row -> {
+ if (row.getInt(1) < 100) {
+ matched.add(
+ row.getLong(0)
+ + ":"
+ + row.getInt(1)
+ + ":"
+ + row.getBlob(2).toData()[0]);
+ }
+ });
+ }
+ Collections.sort(matched);
+ assertThat(matched).as("pushdown %s", enabled).isEqualTo(expected);
+ }
+ }
+
+ /**
+ * A filter on a column outside the projection is not pushed down at all,
and a predicate the
+ * format can not evaluate prunes nothing.
+ */
+ @Test
+ public void testFiltersWhichCanNotPrune() throws Exception {
+ Model m = create("not_pruned", "parquet", false,
Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Arrays.asList("a", "s"), (row, col) -> updated(row, col,
RANGE_ROWS));
+
+ PredicateBuilder builder = builder(m);
+ Predicate onA = builder.lessThan(index(m, "a"), 100);
+ // s is updated too, so the projection still merges the base and the
update file
+ RowType projection = m.table.rowType().project(Arrays.asList("id",
"s"));
+ assertThat(read(pushDown(m.table, true), onA, projection).all)
+ .isEqualTo(read(pushDown(m.table, false), onA,
projection).all);
+
+ assertMatchesModel(m, builder.startsWith(index(m, "s"),
BinaryString.fromString("u0001")));
+ }
+
+ /** The mappings computing candidate rows are keyed without the filters. */
+ @Test
+ public void testFilterChangedOnTheSameRead() throws Exception {
+ Model m = create("refiltered", "parquet", false,
Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - row));
+
+ PredicateBuilder builder = builder(m);
+ Predicate first = builder.lessThan(index(m, "a"), 100);
+ Predicate second = builder.greaterThan(index(m, "a"), RANGE_ROWS -
100);
+ ReadBuilder readBuilder = pushDown(m.table, true).newReadBuilder();
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ InnerTableRead read = (InnerTableRead) readBuilder.newRead();
+
+ read.withFilter(first);
+ assertThat(matched(read, splits, first, m)).isEqualTo(expected(m,
first));
+ read.withFilter(second);
+ assertThat(matched(read, splits, second, m)).isEqualTo(expected(m,
second));
+ }
+
+ /** The footer read for the candidate rows is reused when the files of the
group are opened. */
+ @Test
+ public void testFooterReadOncePerFile() throws Exception {
+ FooterCountingFileIO fileIO = new FooterCountingFileIO();
+ catalog = new FileSystemCatalog(fileIO, new
Path(tempPath.toUri().toString(), "counting"));
+ catalog.createDatabase(database, true);
+ Model m = create("footer", "parquet", false, Collections.emptyMap());
+ writeRange(m, RANGE_ROWS, null, row -> base(row));
+ update(m, 0, Collections.singletonList("a"), (row, col) -> (int)
(RANGE_ROWS - row));
+
+ fileIO.footerReads.clear();
+ Result result =
+ read(pushDown(m.table, true), builder(m).lessThan(index(m,
"a"), 100), null);
+ assertThat(result.matched).hasSize(99);
+ // the update file computes the candidate rows, then both files of the
group are opened
+ assertThat(fileIO.footerReads).hasSize(2);
+ assertThat(fileIO.footerReads.values()).containsOnly(1);
+ }
+
+ /** Counts the footer reads of Parquet files, which seek to the footer
length first. */
+ private static class FooterCountingFileIO extends LocalFileIO {
+
+ private final Map<String, Integer> footerReads = new HashMap<>();
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ SeekableInputStream in = super.newInputStream(path);
+ if (!path.getName().endsWith(".parquet")) {
+ return in;
+ }
+ long footerLengthPosition = getFileStatus(path).getLen() - 8;
+ return new SeekableInputStream() {
+ @Override
+ public void seek(long desired) throws IOException {
+ if (desired == footerLengthPosition) {
+ footerReads.merge(path.getName(), 1, Integer::sum);
+ }
+ in.seek(desired);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return in.getPos();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return in.read();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException
{
+ return in.read(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ in.close();
+ }
+ };
+ }
+ }
+
+ /**
+ * Random tables of several row id ranges, with nulls, a string column, a
column added and
+ * backfilled for some ranges, deletion vectors and random filters.
+ */
+ @Test
+ public void testRandomTables() throws Exception {
+ Random random = new Random(20261013L);
+ for (int round = 0; round < 20; round++) {
+ boolean deletionVectors = round % 3 == 0;
+ Map<String, String> options =
+ deletionVectors
+ ? Collections.singletonMap(
+
CoreOptions.DELETION_VECTORS_ENABLED.key(), "true")
+ : Collections.emptyMap();
+ Model m = create("random_" + round, "parquet", false, options);
+ int ranges = 2 + random.nextInt(3);
+ for (int r = 0; r < ranges; r++) {
+ int rows = 500 + random.nextInt(2000);
+ writeRange(m, rows, null, row -> randomBase(row, random));
+ }
+ int updates = 1 + random.nextInt(3);
+ for (int u = 0; u < updates; u++) {
+ List<String> columns = randomColumns(random);
+ BiFunction<Long, String, Object> values = randomValues(random);
+ for (int r = 0; r < ranges; r++) {
+ if (random.nextInt(10) < 6) {
+ update(m, r, columns, values);
+ }
+ }
+ }
+ if (random.nextBoolean()) {
+ addColumn(m, "x", DataTypes.INT());
+ BiFunction<Long, String, Object> values = randomValues(random);
+ for (int r = 0; r < ranges; r++) {
+ if (random.nextBoolean()) {
+ update(m, r, Collections.singletonList("x"), values);
+ }
+ }
+ }
+ if (deletionVectors) {
+ List<Long> rowIds = new ArrayList<>();
+ for (int d = 0; d < 30; d++) {
+ rowIds.add((long) random.nextInt(m.rows.size()));
+ }
+ delete(m, rowIds);
+ }
+ for (int f = 0; f < 6; f++) {
+ assertMatchesModel(m, randomFilter(random, m));
+ }
+ }
+ }
+
+ // -------------------------------------------------------------- random
data
+
+ private static Map<String, Object> randomBase(long row, Random random) {
+ Map<String, Object> values = base(row);
+ if (random.nextInt(20) == 0) {
+ values.put("a", null);
+ }
+ if (random.nextInt(20) == 0) {
+ values.put("s", null);
+ }
+ values.put("b", random.nextInt(1000));
+ return values;
+ }
+
+ private static List<String> randomColumns(Random random) {
+ List<String> columns = new ArrayList<>();
+ for (String column : Arrays.asList("a", "b", "s")) {
+ if (random.nextBoolean()) {
+ columns.add(column);
+ }
+ }
+ if (columns.isEmpty()) {
+ columns.add("a");
+ }
+ return columns;
+ }
+
+ private static BiFunction<Long, String, Object> randomValues(Random
random) {
+ int pattern = random.nextInt(5);
+ int shift = random.nextInt(5000);
+ long seed = random.nextLong();
+ return (row, column) -> {
+ Random cell = new Random(seed + row * 31 + column.hashCode());
+ // half nulls for pattern 4, a tenth for the others
+ if (pattern == 4 ? cell.nextBoolean() : cell.nextInt(10) == 0) {
+ return null;
+ }
+ long value;
+ switch (pattern) {
+ case 0:
+ value = row + shift;
+ break;
+ case 1:
+ value = 20000 - row + shift;
+ break;
+ case 2:
+ value = cell.nextInt(20000);
+ break;
+ default:
+ value = (row / 97) * 3 + shift;
+ }
+ return column.equals("s") ? String.format("u%06d", value) :
(Object) (int) value;
+ };
+ }
+
+ private static Predicate randomFilter(Random random, Model m) {
+ Predicate first = randomLeaf(random, m);
+ switch (random.nextInt(3)) {
+ case 0:
+ return first;
+ case 1:
+ return PredicateBuilder.and(first, randomLeaf(random, m));
+ default:
+ return PredicateBuilder.or(first, randomLeaf(random, m));
+ }
+ }
+
+ private static Predicate randomLeaf(Random random, Model m) {
+ List<String> columns = new ArrayList<>(Arrays.asList("a", "b", "s"));
+ if (m.columns.containsKey("x")) {
+ columns.add("x");
+ }
+ String column = columns.get(random.nextInt(columns.size()));
+ int field = index(m, column);
+ PredicateBuilder builder = builder(m);
+ int value = random.nextInt(20000);
+ Object literal =
+ column.equals("s")
+ ? BinaryString.fromString(String.format("u%06d",
value))
+ : (Object) value;
+ Object upper =
+ column.equals("s")
+ ? BinaryString.fromString(String.format("u%06d", value
+ 1000))
+ : (Object) (value + 1000);
+ switch (random.nextInt(9)) {
+ case 0:
+ return builder.lessThan(field, literal);
+ case 1:
+ return builder.greaterThan(field, literal);
+ case 2:
+ return builder.equal(field, literal);
+ case 3:
+ return builder.notEqual(field, literal);
+ case 4:
+ return builder.between(field, literal, upper);
+ case 5:
+ return builder.isNull(field);
+ case 6:
+ return builder.isNotNull(field);
+ case 7:
+ return builder.in(field, Arrays.asList(literal, upper));
+ default:
+ return column.equals("s")
+ ? builder.startsWith(field,
BinaryString.fromString("u00"))
+ : builder.notIn(field, Arrays.asList(literal, upper));
+ }
+ }
+
+ private static Map<String, Object> base(long row) {
+ return base(row, (int) row);
+ }
+
+ private static Map<String, Object> base(long row, int b) {
+ Map<String, Object> values = new HashMap<>();
+ values.put("id", row);
+ values.put("a", (int) row);
+ values.put("b", b);
+ values.put("s", String.format("s%06d", row));
+ return values;
+ }
+
+ private static Object updated(long row, String column, int rows) {
+ long value = rows - row;
+ return column.equals("s") ? String.format("u%06d", value) : (Object)
(int) value;
+ }
+
+ // -------------------------------------------------------------- the model
+
+ /** The values of every row of a table, by row id, and its row id ranges.
*/
+ private static class Model {
+ private final String name;
+ private final boolean partitioned;
+ private final LinkedHashMap<String, DataType> columns = new
LinkedHashMap<>();
+ private final List<Map<String, Object>> rows = new ArrayList<>();
+ private final List<long[]> ranges = new ArrayList<>();
+ private final List<String> rangePartitions = new ArrayList<>();
+ private final Set<Long> deleted = new HashSet<>();
+ private FileStoreTable table;
+
+ private Model(String name, boolean partitioned) {
+ this.name = name;
+ this.partitioned = partitioned;
+ }
+ }
+
+ private Model create(
+ String name, String format, boolean partitioned, Map<String,
String> options)
+ throws Exception {
+ Model m = new Model(name, partitioned);
+ if (partitioned) {
+ m.columns.put("pt", DataTypes.STRING());
+ }
+ m.columns.put("id", DataTypes.BIGINT());
+ m.columns.put("a", DataTypes.INT());
+ m.columns.put("b", DataTypes.INT());
+ m.columns.put("s", DataTypes.STRING());
+ Schema.Builder builder = Schema.newBuilder();
+ m.columns.forEach(builder::column);
+ if (partitioned) {
+ builder.partitionKeys("pt");
+ }
+ builder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), format)
+ // several row groups and pages per file
+ .option("parquet.block.size", String.valueOf(16 * 1024))
+ .option("parquet.page.size", String.valueOf(1024));
+ options.forEach(builder::option);
+ catalog.createTable(identifier(name), builder.build(), false);
+ m.table = getTable(identifier(name));
+ return m;
+ }
+
+ private void writeRange(
+ Model m, int count, String partition, Function<Long, Map<String,
Object>> values)
+ throws Exception {
+ long firstRowId = m.rows.size();
+ BatchWriteBuilder builder = m.table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (long row = firstRowId; row < firstRowId + count; row++) {
+ Map<String, Object> rowValues = new
HashMap<>(values.apply(row));
+ if (partition != null) {
+ rowValues.put("pt", partition);
+ }
+ m.rows.add(rowValues);
+ write.write(toRow(m, rowValues, new
ArrayList<>(m.columns.keySet())));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ m.table = getTable(identifier(m.name));
+ m.ranges.add(new long[] {firstRowId, count});
+ m.rangePartitions.add(partition);
+ assertThat(
+ m.table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .anyMatch(f -> f.nonNullFirstRowId() ==
firstRowId))
+ .as("the new base file starts at row id %s", firstRowId)
+ .isTrue();
+ }
+
+ private void update(
+ Model m, int range, List<String> columns, BiFunction<Long, String,
Object> values)
+ throws Exception {
+ long firstRowId = m.ranges.get(range)[0];
+ long count = m.ranges.get(range)[1];
+ String partition = m.rangePartitions.get(range);
+ List<String> writeColumns = new ArrayList<>();
+ if (m.partitioned) {
+ writeColumns.add("pt");
+ }
+ writeColumns.addAll(columns);
+ BatchWriteBuilder builder = m.table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+
builder.newWrite().withWriteType(m.table.rowType().project(writeColumns));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (long row = firstRowId; row < firstRowId + count; row++) {
+ Map<String, Object> rowValues = m.rows.get((int) row);
+ for (String column : columns) {
+ rowValues.put(column, values.apply(row, column));
+ }
+ if (partition != null) {
+ rowValues.put("pt", partition);
+ }
+ write.write(toRow(m, rowValues, writeColumns));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ commit.commit(messages);
+ }
+ m.table = getTable(identifier(m.name));
+ }
+
+ private void addColumn(Model m, String column, DataType type) throws
Exception {
+ catalog.alterTable(identifier(m.name), SchemaChange.addColumn(column,
type), false);
+ m.columns.put(column, type);
+ m.table = getTable(identifier(m.name));
+ }
+
+ private void alterType(Model m, String column, DataType type) throws
Exception {
+ catalog.alterTable(identifier(m.name),
SchemaChange.updateColumnType(column, type), false);
+ m.columns.put(column, type);
+ m.table = getTable(identifier(m.name));
+ }
+
+ private void delete(Model m, List<Long> rowIds) throws Exception {
+ TreeMap<Long, List<DataFileMeta>> groups = new TreeMap<>();
+ for (ManifestEntry entry : m.table.store().newScan().plan().files()) {
+ groups.computeIfAbsent(entry.file().nonNullFirstRowId(), k -> new
ArrayList<>())
+ .add(entry.file());
+ }
+ Map<String, DeletionVector> vectors = new HashMap<>();
+ for (long rowId : rowIds) {
+ Map.Entry<Long, List<DataFileMeta>> group =
groups.floorEntry(rowId);
+ DataFileMeta anchor = retrieveAnchorFile(group.getValue(), f -> f);
+ vectors.computeIfAbsent(anchor.fileName(), k -> new
BitmapDeletionVector())
+ .delete(rowId - group.getKey());
+ m.deleted.add(rowId);
+ }
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ m.table.store().newIndexFileHandler(),
+ m.table.latestSnapshot().get(),
+ BinaryRow.EMPTY_ROW);
+ vectors.forEach(maintainer::notifyNewDeletionVector);
+ List<IndexFileMeta> added = new ArrayList<>();
+ List<IndexFileMeta> removed = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ (entry.kind() == FileKind.ADD ? added :
removed).add(entry.indexFile());
+ }
+ try (BatchTableCommit commit =
m.table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ added,
+ removed),
+ CompactIncrement.emptyIncrement())));
+ }
+ m.table = getTable(identifier(m.name));
+ }
+
+ private static void assignFirstRowId(List<CommitMessage> messages, long
firstRowId) {
+ for (CommitMessage message : messages) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ List<DataFileMeta> files = new
ArrayList<>(impl.newFilesIncrement().newFiles());
+ impl.newFilesIncrement().newFiles().clear();
+ files.forEach(
+ f ->
impl.newFilesIncrement().newFiles().add(f.assignFirstRowId(firstRowId)));
+ }
+ }
+
+ private static GenericRow toRow(Model m, Map<String, Object> values,
List<String> columns) {
+ Object[] fields = new Object[columns.size()];
+ for (int i = 0; i < columns.size(); i++) {
+ fields[i] = internal(values.get(columns.get(i)),
m.columns.get(columns.get(i)));
+ }
+ return GenericRow.of(fields);
+ }
+
+ private static Object internal(Object value, DataType type) {
+ if (value == null) {
+ return null;
+ }
+ if (type.getTypeRoot() == DataTypeRoot.VARCHAR) {
+ return BinaryString.fromString((String) value);
+ }
+ if (type.getTypeRoot() == DataTypeRoot.BIGINT) {
+ return ((Number) value).longValue();
+ }
+ return value;
+ }
+
+ // -------------------------------------------------------------------
reads
+
+ private static class Result {
+ private final List<String> matched = new ArrayList<>();
+ private final List<String> all = new ArrayList<>();
+ private long returned;
+ // rows returned by the same read without the pushdown
+ private long baseline;
+ }
+
+ private Result assertMatchesModel(Model m, Predicate filter) throws
Exception {
+ List<String> expected = expected(m, filter);
+ Result disabled = read(pushDown(m.table, false), filter, null);
+ Result enabled = read(pushDown(m.table, true), filter, null);
+ assertThat(disabled.matched).as("%s, disabled, %s", m.name,
filter).isEqualTo(expected);
+ assertThat(enabled.matched).as("%s, enabled, %s", m.name,
filter).isEqualTo(expected);
+ assertThat(enabled.returned).isLessThanOrEqualTo(disabled.returned);
+ enabled.baseline = disabled.returned;
+ return enabled;
+ }
+
+ private static List<String> expected(Model m, Predicate filter) {
+ RowType rowType = m.table.rowType();
+ List<String> expected = new ArrayList<>();
+ for (int row = 0; row < m.rows.size(); row++) {
+ if (m.deleted.contains((long) row)) {
+ continue;
+ }
+ GenericRow genericRow = toRow(m, m.rows.get(row), new
ArrayList<>(m.columns.keySet()));
+ if (filter.test(genericRow)) {
+ expected.add(format(genericRow, rowType));
+ }
+ }
+ Collections.sort(expected);
+ return expected;
+ }
+
+ private static Result read(FileStoreTable table, Predicate filter, RowType
projection)
+ throws Exception {
+ RowType readType = projection == null ? table.rowType() : projection;
+ ReadBuilder builder =
table.newReadBuilder().withReadType(readType).withFilter(filter);
+ Result result = new Result();
+ try (RecordReader<InternalRow> reader =
+
builder.newRead().createReader(builder.newScan().plan().splits())) {
+ reader.forEachRemaining(
+ row -> {
+ result.returned++;
+ String formatted = format(row, readType);
+ result.all.add(formatted);
+ if (projection == null && filter.test(row)) {
+ result.matched.add(formatted);
+ }
+ });
+ }
+ Collections.sort(result.matched);
+ Collections.sort(result.all);
+ return result;
+ }
+
+ private static List<String> matched(
+ InnerTableRead read, List<Split> splits, Predicate filter, Model
m) throws Exception {
+ List<String> matched = new ArrayList<>();
+ RowType rowType = m.table.rowType();
+ try (RecordReader<InternalRow> reader = read.createReader(splits)) {
+ reader.forEachRemaining(
+ row -> {
+ if (filter.test(row)) {
+ matched.add(format(row, rowType));
+ }
+ });
+ }
+ Collections.sort(matched);
+ return matched;
+ }
+
+ private static String format(InternalRow row, RowType rowType) {
+ InternalRow.FieldGetter[] getters =
+ InternalRowUtils.createFieldGetters(rowType.getFieldTypes());
+ List<String> values = new ArrayList<>(getters.length);
+ for (InternalRow.FieldGetter getter : getters) {
+ values.add(String.valueOf(getter.getFieldOrNull(row)));
+ }
+ return String.join(",", values);
+ }
+
+ private static FileStoreTable pushDown(FileStoreTable table, boolean
enabled) {
+ return copy(
+ table,
+ DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED.key(),
+ String.valueOf(enabled));
+ }
+
+ private static FileStoreTable copy(FileStoreTable table, String key,
String value) {
+ return table.copy(Collections.singletonMap(key, value));
+ }
+
+ private static PredicateBuilder builder(Model m) {
+ return new PredicateBuilder(m.table.rowType());
+ }
+
+ private static int index(Model m, String column) {
+ List<String> names =
+ m.table.rowType().getFields().stream()
+ .map(DataField::name)
+ .collect(Collectors.toList());
+ return names.indexOf(column);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownTest.java
new file mode 100644
index 0000000000..465745f8a4
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionStatsPushDownTest.java
@@ -0,0 +1,627 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+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.ReadBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.IntUnaryOperator;
+import java.util.stream.Collectors;
+
+import static
org.apache.paimon.CoreOptions.DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests pushing filters into merged data evolution groups by file
statistics. */
+public class DataEvolutionStatsPushDownTest extends TableTestBase {
+
+ private static final int ROWS = 20_000;
+
+ private int[] a;
+ private int[] b;
+ private final Set<Long> deleted = new HashSet<>();
+
+ @Override
+ public Schema schemaDefault() {
+ return schema(false, ROWS, Collections.emptyMap());
+ }
+
+ private Schema schema(boolean deletionVectors, int rows, Map<String,
String> options) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.INT())
+ .column("c", DataTypes.STRING())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ // several row groups and pages per file
+ .option("parquet.block.size", String.valueOf(rows * 2))
+ .option("parquet.page.size", String.valueOf(1024));
+ if (deletionVectors) {
+ builder.option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ }
+ options.forEach(builder::option);
+ return builder.build();
+ }
+
+ @Test
+ public void testStaleCopyDoesNotExcludeMatches() throws Exception {
+ FileStoreTable table = create("stale", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+
+ Predicate filter = builder(table).lessThan(1, 200);
+ Result result = assertMatchesModel(table, filter);
+ assertThat(result.matched).hasSize(200);
+ assertThat(result.returned).isLessThan(ROWS / 2);
+ }
+
+ @Test
+ public void testNewestUpdateWins() throws Exception {
+ FileStoreTable table = create("newest", false, ROWS);
+ update(table, "a", i -> i + ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+
+ Result result = assertMatchesModel(table, builder(table).lessThan(1,
200));
+ assertThat(result.matched).hasSize(200);
+ assertThat(result.returned).isLessThan(ROWS / 2);
+ }
+
+ @Test
+ public void testFiltersWonByDifferentFiles() throws Exception {
+ FileStoreTable table = create("different", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ update(table, "b", i -> i * 2);
+
+ PredicateBuilder builder = builder(table);
+ Predicate filter =
+ PredicateBuilder.and(
+ builder.lessThan(1, 2000), builder.greaterOrEqual(2,
(ROWS - 500) * 2));
+ Result result = assertMatchesModel(table, filter);
+ assertThat(result.matched).hasSize(500);
+ assertThat(result.returned).isLessThan(ROWS / 2);
+ }
+
+ @Test
+ public void testOrAcrossFiles() throws Exception {
+ FileStoreTable table = create("or", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ update(table, "b", i -> i);
+
+ PredicateBuilder builder = builder(table);
+ Result result =
+ assertMatchesModel(
+ table,
+ PredicateBuilder.or(builder.lessThan(1, 100),
builder.lessThan(2, 100)));
+ assertThat(result.matched).hasSize(200);
+ }
+
+ @Test
+ public void testWithDeletionVectors() throws Exception {
+ FileStoreTable table = create("dv", true, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ List<Long> toDelete = new ArrayList<>();
+ for (long id = ROWS - 1; id >= ROWS - 100; id -= 3) {
+ toDelete.add(id);
+ }
+ toDelete.add(5L);
+ deleteRows(table, toDelete);
+
+ Result result = assertMatchesModel(table, builder(table).lessThan(1,
200));
+ assertThat(result.matched).hasSize(200 - 34);
+ assertThat(result.returned).isLessThan(ROWS / 2);
+ }
+
+ @Test
+ public void testWithRowRanges() throws Exception {
+ FileStoreTable table = create("ranges", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+
+ Predicate filter = builder(table).lessThan(1, 200);
+ List<Range> rowRanges = Collections.singletonList(new Range(ROWS -
150, ROWS - 1));
+ for (boolean enabled : new boolean[] {true, false}) {
+ Result result = read(withPushDown(table, enabled), filter,
rowRanges);
+
assertThat(result.matched).containsExactlyElementsOf(expected(filter,
rowRanges));
+ assertThat(result.matched).hasSize(150);
+ }
+ }
+
+ @Test
+ public void testRenamedColumn() throws Exception {
+ FileStoreTable table = create("renamed", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ catalog.alterTable(identifier("renamed"),
SchemaChange.renameColumn("a", "a2"), false);
+ table = getTable(identifier("renamed"));
+
+ Result result = assertMatchesModel(table, builder(table).lessThan(1,
200));
+ assertThat(result.matched).hasSize(200);
+ assertThat(result.returned).isLessThan(ROWS / 2);
+ }
+
+ @Test
+ public void testLostFileBehavesAsWithoutPushDown() throws Exception {
+ FileStoreTable table = create("lost", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ table = getTable(identifier("lost"));
+ table.fileIO().deleteQuietly(updateFilePath(table));
+
+ assertSameOutcome(table, CoreOptions.SCAN_IGNORE_LOST_FILE.key());
+ }
+
+ @Test
+ public void testCorruptFileBehavesAsWithoutPushDown() throws Exception {
+ FileStoreTable table = create("corrupt", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+ table = getTable(identifier("corrupt"));
+ table.fileIO().overwriteFileUtf8(updateFilePath(table), "not a parquet
file");
+
+ assertSameOutcome(table, CoreOptions.SCAN_IGNORE_CORRUPT_FILE.key());
+ }
+
+ @Test
+ public void testWithRowSidecar() throws Exception {
+ FileStoreTable table =
+ create(
+ "sidecar",
+ false,
+ ROWS,
+ Collections.singletonMap(
+
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"));
+ update(table, "a", i -> ROWS - 1 - i);
+ table = getTable(identifier("sidecar"));
+ // the few rows left by the candidate ranges are read from the sidecars
+ assertThat(
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .anyMatch(
+ f ->
+ f.extraFiles().stream()
+ .anyMatch(name ->
name.endsWith(".row"))))
+ .isTrue();
+
+ Result result = assertMatchesModel(table, builder(table).lessThan(1,
100));
+ assertThat(result.matched).hasSize(100);
+ assertThat(result.returned).isLessThan(ROWS / 10);
+ }
+
+ /**
+ * The row sidecar of a table of these widths needs a wider null header
once the row tracking
+ * fields are appended to its schema, so reads with candidate rows must
not select it.
+ */
+ @ParameterizedTest
+ @ValueSource(ints = {6, 7, 8, 9, 15, 16})
+ public void testWideTableWithRowSidecar(int width) throws Exception {
+ String name = "wide_" + width;
+ Schema.Builder schema = Schema.newBuilder();
+ for (int c = 0; c < width; c++) {
+ schema.column("c" + c, DataTypes.INT());
+ }
+ schema.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(),
"true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option("parquet.block.size", String.valueOf(ROWS * 2))
+ .option("parquet.page.size", String.valueOf(1024));
+ catalog.createTable(identifier(name), schema.build(), false);
+
+ FileStoreTable table = getTable(identifier(name));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < ROWS; i++) {
+ Object[] values = new Object[width];
+ values[0] = i;
+ values[1] = i;
+ for (int c = 2; c < width; c++) {
+ values[c] = i + c;
+ }
+ write.write(GenericRow.of(values));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+
table.rowType().project(Collections.singletonList("c1")));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < ROWS; i++) {
+ write.write(GenericRow.of(ROWS - 1 - i));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages);
+ commit.commit(messages);
+ }
+
+ table = getTable(identifier(name));
+ RowType readType = table.rowType().project(Arrays.asList("c0", "c1"));
+ Predicate filter = new PredicateBuilder(table.rowType()).lessThan(1,
100);
+ List<String> expected = new ArrayList<>();
+ for (int i = ROWS - 100; i < ROWS; i++) {
+ expected.add(i + ":" + (ROWS - 1 - i));
+ }
+ for (boolean enabled : new boolean[] {true, false}) {
+ ReadBuilder read =
+ withPushDown(table, enabled)
+ .newReadBuilder()
+ .withReadType(readType)
+ .withFilter(filter);
+ List<String> matched = new ArrayList<>();
+ long[] returned = new long[1];
+ try (RecordReader<InternalRow> reader =
+
read.newRead().createReader(read.newScan().plan().splits())) {
+ reader.forEachRemaining(
+ row -> {
+ returned[0]++;
+ if (row.getInt(1) < 100) {
+ matched.add(row.getInt(0) + ":" +
row.getInt(1));
+ }
+ });
+ }
+ assertThat(matched)
+ .as("width %s, pushdown %s", width, enabled)
+ .containsExactlyInAnyOrderElementsOf(expected);
+ if (enabled) {
+ assertThat(returned[0]).isLessThan(ROWS / 10);
+ }
+ }
+ }
+
+ private Path updateFilePath(FileStoreTable table) {
+ DataFileMeta update =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(f -> f.writeCols() != null &&
f.writeCols().contains("a"))
+ .findFirst()
+ .get();
+ return table.store()
+ .pathFactory()
+ .createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0)
+ .toPath(update);
+ }
+
+ /** The pushdown must not change what a read does, with the file option on
and off. */
+ private void assertSameOutcome(FileStoreTable table, String ignoreOption) {
+ Predicate filter = builder(table).lessThan(1, 200);
+ for (boolean ignore : new boolean[] {false, true}) {
+ FileStoreTable options =
+ table.copy(Collections.singletonMap(ignoreOption,
String.valueOf(ignore)));
+ assertThat(outcome(withPushDown(options, true), filter))
+ .as("%s=%s", ignoreOption, ignore)
+ .isEqualTo(outcome(withPushDown(options, false), filter));
+ }
+ }
+
+ private String outcome(FileStoreTable table, Predicate filter) {
+ try {
+ return "rows " + read(table, filter, null).matched;
+ } catch (Exception e) {
+ return "error " + e.getClass().getName();
+ }
+ }
+
+ @Test
+ public void testDisabled() throws Exception {
+ FileStoreTable table = create("disabled", false, ROWS);
+ update(table, "a", i -> ROWS - 1 - i);
+
+ Predicate filter = builder(table).lessThan(1, 200);
+ Result result = read(withPushDown(table, false), filter, null);
+ assertThat(result.matched).containsExactlyElementsOf(expected(filter,
null));
+ assertThat(result.returned).isEqualTo(ROWS);
+ }
+
+ @Test
+ public void testRandomUpdatesAndFilters() throws Exception {
+ int rows = 3000;
+ Random random = new Random(20260929L);
+ for (int round = 0; round < 30; round++) {
+ FileStoreTable table = create("random_" + round, round % 3 == 0,
rows);
+ int updates = 1 + random.nextInt(3);
+ for (int u = 0; u < updates; u++) {
+ List<String> columns = new ArrayList<>();
+ if (random.nextBoolean()) {
+ columns.add("a");
+ }
+ if (columns.isEmpty() || random.nextBoolean()) {
+ columns.add("b");
+ }
+ IntUnaryOperator values = randomValues(random, rows);
+ update(table, columns, values);
+ }
+ if (round % 3 == 0) {
+ List<Long> toDelete = new ArrayList<>();
+ for (int d = 0; d < 20; d++) {
+ toDelete.add((long) random.nextInt(rows));
+ }
+ deleteRows(table, toDelete);
+ }
+ for (int f = 0; f < 5; f++) {
+ assertMatchesModel(table, randomFilter(random, builder(table),
rows));
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------ model
+
+ private static IntUnaryOperator randomValues(Random random, int rows) {
+ int shift = random.nextInt(rows);
+ switch (random.nextInt(4)) {
+ case 0:
+ return i -> i + shift;
+ case 1:
+ return i -> rows - 1 - i + shift;
+ case 2:
+ long seed = random.nextLong();
+ return i -> new Random(seed + i).nextInt(rows * 2);
+ default:
+ return i -> (i / 97) * 3 + shift;
+ }
+ }
+
+ private static Predicate randomFilter(Random random, PredicateBuilder
builder, int rows) {
+ Predicate first = randomLeaf(random, builder, rows);
+ if (random.nextBoolean()) {
+ return first;
+ }
+ Predicate second = randomLeaf(random, builder, rows);
+ return random.nextBoolean()
+ ? PredicateBuilder.and(first, second)
+ : PredicateBuilder.or(first, second);
+ }
+
+ private static Predicate randomLeaf(Random random, PredicateBuilder
builder, int rows) {
+ int field = 1 + random.nextInt(2);
+ int value = random.nextInt(rows * 2);
+ switch (random.nextInt(4)) {
+ case 0:
+ return builder.lessThan(field, value);
+ case 1:
+ return builder.greaterThan(field, value);
+ case 2:
+ return builder.equal(field, value);
+ default:
+ return builder.between(field, value, value + rows / 20);
+ }
+ }
+
+ private Result assertMatchesModel(FileStoreTable table, Predicate filter)
throws Exception {
+ List<String> expected = expected(filter, null);
+ Result disabled = read(withPushDown(table, false), filter, null);
+ Result enabled = read(withPushDown(table, true), filter, null);
+ assertThat(disabled.matched).as("disabled, %s",
filter).isEqualTo(expected);
+ assertThat(enabled.matched).as("enabled, %s",
filter).isEqualTo(expected);
+ return enabled;
+ }
+
+ private List<String> expected(Predicate filter, List<Range> rowRanges) {
+ List<String> expected = new ArrayList<>();
+ for (int i = 0; i < a.length; i++) {
+ if (deleted.contains((long) i) || !inRanges(rowRanges, i)) {
+ continue;
+ }
+ GenericRow row = GenericRow.of((long) i, a[i], b[i],
BinaryString.fromString("c-" + i));
+ if (filter.test(row)) {
+ expected.add(format(row));
+ }
+ }
+ return expected;
+ }
+
+ private static boolean inRanges(List<Range> rowRanges, long rowId) {
+ return rowRanges == null
+ || rowRanges.stream().anyMatch(r -> r.from <= rowId && rowId
<= r.to);
+ }
+
+ // --------------------------------------------------------------- table io
+
+ private FileStoreTable create(String name, boolean deletionVectors, int
rows) throws Exception {
+ return create(name, deletionVectors, rows, Collections.emptyMap());
+ }
+
+ private FileStoreTable create(
+ String name, boolean deletionVectors, int rows, Map<String,
String> options)
+ throws Exception {
+ catalog.createTable(identifier(name), schema(deletionVectors, rows,
options), false);
+ FileStoreTable table = getTable(identifier(name));
+ a = new int[rows];
+ b = new int[rows];
+ deleted.clear();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < rows; i++) {
+ a[i] = i;
+ b[i] = i;
+ write.write(GenericRow.of((long) i, i, i,
BinaryString.fromString("c-" + i)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ return getTable(identifier(name));
+ }
+
+ private void update(FileStoreTable table, String column, IntUnaryOperator
values)
+ throws Exception {
+ update(table, Collections.singletonList(column), values);
+ }
+
+ private void update(FileStoreTable table, List<String> columns,
IntUnaryOperator values)
+ throws Exception {
+ table = getTable(identifier(table.name()));
+ RowType writeType = table.rowType().project(columns);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < a.length; i++) {
+ int value = values.applyAsInt(i);
+ Object[] fields = new Object[columns.size()];
+ for (int c = 0; c < columns.size(); c++) {
+ fields[c] = value + c;
+ if (columns.get(c).equals("a")) {
+ a[i] = value + c;
+ } else {
+ b[i] = value + c;
+ }
+ }
+ write.write(GenericRow.of(fields));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages);
+ commit.commit(messages);
+ }
+ }
+
+ /** Updates cover the row id range of the single base file, which starts
at 0. */
+ private static void assignFirstRowId(List<CommitMessage> messages) {
+ for (CommitMessage message : messages) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ List<DataFileMeta> files = new
ArrayList<>(impl.newFilesIncrement().newFiles());
+ impl.newFilesIncrement().newFiles().clear();
+ files.forEach(f ->
impl.newFilesIncrement().newFiles().add(f.assignFirstRowId(0)));
+ }
+ }
+
+ private void deleteRows(FileStoreTable table, List<Long> rowIds) throws
Exception {
+ table = getTable(identifier(table.name()));
+ List<DataFileMeta> files =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ DataFileMeta anchor = retrieveAnchorFile(files, f -> f);
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ table.store().newIndexFileHandler(),
+ table.latestSnapshot().get(),
+ BinaryRow.EMPTY_ROW);
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ for (long rowId : rowIds) {
+ deletionVector.delete(rowId - anchor.nonNullFirstRowId());
+ deleted.add(rowId);
+ }
+ maintainer.notifyNewDeletionVector(anchor.fileName(), deletionVector);
+
+ List<IndexFileMeta> newIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ if (entry.kind() == FileKind.ADD) {
+ newIndexFiles.add(entry.indexFile());
+ } else {
+ deletedIndexFiles.add(entry.indexFile());
+ }
+ }
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ newIndexFiles,
+ deletedIndexFiles),
+ CompactIncrement.emptyIncrement())));
+ }
+ }
+
+ private static FileStoreTable withPushDown(FileStoreTable table, boolean
enabled) {
+ return table.copy(
+ Collections.singletonMap(
+
DATA_EVOLUTION_MERGED_READ_STATS_PUSHDOWN_ENABLED.key(),
+ String.valueOf(enabled)));
+ }
+
+ private static PredicateBuilder builder(FileStoreTable table) {
+ return new PredicateBuilder(table.rowType());
+ }
+
+ private Result read(FileStoreTable table, Predicate filter, List<Range>
rowRanges)
+ throws Exception {
+ ReadBuilder builder = table.newReadBuilder().withFilter(filter);
+ if (rowRanges != null) {
+ builder.withRowRanges(rowRanges);
+ }
+ Result result = new Result();
+ try (RecordReader<InternalRow> reader =
+
builder.newRead().createReader(builder.newScan().plan().splits())) {
+ reader.forEachRemaining(
+ row -> {
+ result.returned++;
+ if (filter.test(row)) {
+ result.matched.add(format(row));
+ }
+ });
+ }
+ result.matched.sort(null);
+ return result;
+ }
+
+ private static String format(InternalRow row) {
+ return String.format(
+ "%010d,%d,%d,%s", row.getLong(0), row.getInt(1),
row.getInt(2), row.getString(3));
+ }
+
+ private static class Result {
+ private final List<String> matched = new ArrayList<>();
+ private long returned;
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
index 5fcee88b50..b04d7f3755 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.columnar.VectorizedColumnBatch;
import org.apache.paimon.data.columnar.writable.WritableColumnVector;
import org.apache.paimon.data.shredding.MapSharedShreddingReadPlanFactory;
import org.apache.paimon.data.shredding.ShreddingReadPlan;
+import org.apache.paimon.format.FileMetadataCache;
import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatReaderFactory;
import
org.apache.paimon.format.parquet.ParquetListLayoutResolver.LayoutContext;
@@ -50,6 +51,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VectorType;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.Range;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.filter2.compat.FilterCompat;
@@ -186,6 +188,58 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
return requestedSchemaCache;
}
+ @Nullable
+ @Override
+ public List<Range> candidateRowRanges(FormatReaderFactory.Context context)
throws IOException {
+ if (predicates == null || predicates.isEmpty()) {
+ return null;
+ }
+
+ ParquetInputFile inputFile =
+ ParquetInputFile.fromPath(context.fileIO(),
context.filePath(), context.fileSize());
+ ParquetReadOptions.Builder readOptionsBuilder =
+ ParquetUtil.getParquetReadOptionsBuilder(conf).withRange(0,
context.fileSize());
+ ParquetInputStream inputStream = inputFile.newStream();
+ ParquetFileReader reader = null;
+ try {
+ ParquetMetadata footer =
+ readFooter(context, inputFile, readOptionsBuilder.build(),
inputStream);
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter =
+ ParquetFilters.convert(predicates, fileSchema,
caseSensitive);
+ if (!FilterCompat.isFilteringRequired(filter)) {
+ return null;
+ }
+ ParquetReadOptions readOptions =
readOptionsBuilder.withRecordFilter(filter).build();
+ reader = new ParquetFileReader(inputFile, footer, readOptions,
inputStream, null);
+ // only the filtered columns, the column index store reads the
indexes of all of them
+
reader.setRequestedSchema(getOrCreateRequestedSchema(fileSchema).messageType);
+ return reader.candidateRowRanges();
+ } finally {
+ // the reader owns the stream once it is created
+ if (reader != null) {
+ reader.close();
+ } else {
+ inputStream.close();
+ }
+ }
+ }
+
+ private static ParquetMetadata readFooter(
+ FormatReaderFactory.Context context,
+ ParquetInputFile inputFile,
+ ParquetReadOptions options,
+ ParquetInputStream inputStream)
+ throws IOException {
+ FileMetadataCache cache = context.metadataCache();
+ if (cache == null) {
+ return ParquetFileReader.readFooter(inputFile, options,
inputStream, true);
+ }
+ return cache.getOrLoad(
+ context.filePath(),
+ path -> ParquetFileReader.readFooter(inputFile, options,
inputStream, true));
+ }
+
@Override
public FileRecordReader<InternalRow>
createReader(FormatReaderFactory.Context context)
throws IOException {
@@ -195,8 +249,7 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
ParquetUtil.getParquetReadOptionsBuilder(conf).withRange(0,
context.fileSize());
ParquetInputStream inputStream = inputFile.newStream();
ParquetMetadata footer =
- ParquetFileReader.readFooter(
- inputFile, readOptionsBuilder.build(), inputStream,
true);
+ readFooter(context, inputFile, readOptionsBuilder.build(),
inputStream);
MessageType fileSchema = footer.getFileMetaData().getSchema();
ParquetFileReader reader;
diff --git
a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 9f56c21769..7f138aa33d 100644
---
a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++
b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -22,6 +22,7 @@ import org.apache.paimon.format.parquet.ParquetInputFile;
import org.apache.paimon.format.parquet.ParquetInputStream;
import org.apache.paimon.fs.FileRange;
import org.apache.paimon.fs.VectoredReadable;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringBitmap32;
import org.apache.hadoop.fs.Path;
@@ -377,6 +378,39 @@ public class ParquetFileReader implements Closeable {
return total;
}
+ /**
+ * Rows of the row groups left by the statistics, dictionary and bloom
filters which the column
+ * indexes of the requested columns can not exclude, as sorted file row
positions. Returns null
+ * when the row positions are unknown. No data page is read, only the
metadata of the file and,
+ * when those filters are enabled, the dictionaries and bloom filters of
the row groups.
+ */
+ @Nullable
+ public List<Range> candidateRowRanges() {
+ boolean pageFiltering =
+ options.useColumnIndexFilter()
+ &&
(FilterCompat.isFilteringRequired(options.getRecordFilter())
+ || selection != null);
+ List<Range> ranges = new ArrayList<>();
+ for (int i = 0, n = blocks.size(); i < n; ++i) {
+ BlockMetaData block = blocks.get(i);
+ long offset = block.getRowIndexOffset();
+ if (offset < 0) {
+ return null;
+ }
+ if (block.getRowCount() == 0) {
+ continue;
+ }
+ if (!pageFiltering) {
+ ranges.add(new Range(offset, offset + block.getRowCount() -
1));
+ continue;
+ }
+ for (RowRanges.Range range : getRowRanges(i).getRanges()) {
+ ranges.add(new Range(offset + range.from, offset + range.to));
+ }
+ }
+ return ranges;
+ }
+
/**
* @return the path for this file
* @deprecated will be removed in 2.0.0; use {@link #getFile()} instead
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCandidateRowRangesTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCandidateRowRangesTest.java
new file mode 100644
index 0000000000..86bc5e0d3e
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCandidateRowRangesTest.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.format.parquet;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FileMetadataCache;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOFunction;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringBitmap32;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ParquetReaderFactory#candidateRowRanges}. */
+public class ParquetCandidateRowRangesTest {
+
+ private static final RowType ROW_TYPE =
+ RowType.of(DataTypes.INT(), DataTypes.STRING(),
DataTypes.BIGINT());
+ private static final int ROWS = 100_000;
+
+ @TempDir public File folder;
+
+ private final LocalFileIO fileIO = new LocalFileIO();
+ private Path path;
+
+ @BeforeEach
+ public void before() throws Exception {
+ path = new Path(folder.getPath(), UUID.randomUUID() + ".parquet");
+ Options options = new Options();
+ // several row groups of several pages each
+ options.set("parquet.block.size", String.valueOf(256 * 1024));
+ options.set("parquet.page.size", String.valueOf(8 * 1024));
+ ParquetWriterFactory factory =
+ new ParquetWriterFactory(new RowDataParquetBuilder(ROW_TYPE,
options));
+ FormatWriter writer = factory.create(fileIO.newOutputStream(path,
false), "snappy");
+ for (int i = 0; i < ROWS; i++) {
+ writer.addElement(
+ GenericRow.of(i, BinaryString.fromString("value-" + i),
(long) (i % 7)));
+ }
+ writer.close();
+ }
+
+ @Test
+ public void testRangesCoverMatchesAndPrune() throws Exception {
+ List<Range> ranges = candidateRowRanges(builder().lessThan(0, 1000),
null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(covered(ranges)).isLessThan(ROWS / 10);
+ assertThat(contains(ranges, 0, 999)).isTrue();
+ }
+
+ @Test
+ public void testRangesAreFilePositionsAcrossRowGroups() throws Exception {
+ List<Range> ranges =
+ candidateRowRanges(
+ PredicateBuilder.and(
+ builder().greaterOrEqual(0, 60_000),
+ builder().lessOrEqual(0, 60_010)),
+ null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(contains(ranges, 60_000, 60_010)).isTrue();
+ assertThat(contains(ranges, 0, 0)).isFalse();
+ assertThat(covered(ranges)).isLessThan(ROWS / 10);
+ assertThat(readSelected(ranges, 0)).contains(60_000, 60_005, 60_010);
+ }
+
+ @Test
+ public void testNoMatchAndNoFilter() throws Exception {
+ assertThat(candidateRowRanges(builder().greaterThan(0, ROWS),
null)).isEmpty();
+ assertThat(
+ new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
null)
+ .candidateRowRanges(context(null)))
+ .isNull();
+ }
+
+ @Test
+ public void testSelectedRowsContainAllMatches() throws Exception {
+ List<Range> ranges = candidateRowRanges(builder().lessThan(0, 1000),
null);
+ List<Integer> ids = readSelected(ranges, 0);
+
+ List<Integer> matched = new ArrayList<>();
+ for (int id : ids) {
+ if (id < 1000) {
+ matched.add(id);
+ }
+ }
+ // the format reader skips whole pages only, rows are cut exactly by
the caller
+ assertThat(matched).hasSize(1000);
+ assertThat((long) ids.size()).isGreaterThanOrEqualTo(covered(ranges));
+ }
+
+ @Test
+ public void testFooterIsReadOnceWithCache() throws Exception {
+ int[] loads = new int[1];
+ FileMetadataCache cache =
+ new FileMetadataCache() {
+ @Override
+ public <T> T getOrLoad(Path path, IOFunction<Path, T>
loader)
+ throws IOException {
+ return super.getOrLoad(
+ path,
+ p -> {
+ loads[0]++;
+ return loader.apply(p);
+ });
+ }
+ };
+
+ List<Range> ranges = candidateRowRanges(builder().lessThan(0, 1000),
cache);
+ ParquetReaderFactory reader = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
+ int rows = 0;
+ try (RecordReader<InternalRow> recordReader =
+ reader.createReader(context(cache, toSelection(ranges)))) {
+ RecordReader.RecordIterator<InternalRow> batch;
+ while ((batch = recordReader.readBatch()) != null) {
+ while (batch.next() != null) {
+ rows++;
+ }
+ batch.releaseBatch();
+ }
+ }
+
+ assertThat(loads[0]).isEqualTo(1);
+ assertThat((long) rows).isGreaterThanOrEqualTo(covered(ranges));
+ assertThat(rows).isLessThan(ROWS / 10);
+ }
+
+ private static PredicateBuilder builder() {
+ return new PredicateBuilder(ROW_TYPE);
+ }
+
+ private List<Range> candidateRowRanges(Predicate predicate,
FileMetadataCache cache)
+ throws IOException {
+ return new ParquetReaderFactory(
+ new Options(), ROW_TYPE, 500,
Collections.singletonList(predicate))
+ .candidateRowRanges(context(cache));
+ }
+
+ private FormatReaderContext context(FileMetadataCache cache) throws
IOException {
+ return context(cache, null);
+ }
+
+ private FormatReaderContext context(FileMetadataCache cache,
RoaringBitmap32 selection)
+ throws IOException {
+ return new FormatReaderContext(
+ fileIO, path, fileIO.getFileSize(path), selection, null,
cache);
+ }
+
+ private List<Integer> readSelected(List<Range> ranges, int field) throws
IOException {
+ List<Integer> values = new ArrayList<>();
+ ParquetReaderFactory reader = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
+ try (RecordReader<InternalRow> recordReader =
+ reader.createReader(context(null, toSelection(ranges)))) {
+ recordReader.forEachRemaining(row ->
values.add(row.getInt(field)));
+ }
+ return values;
+ }
+
+ private static RoaringBitmap32 toSelection(List<Range> ranges) {
+ RoaringBitmap32 selection = new RoaringBitmap32();
+ for (Range range : ranges) {
+ for (long row = range.from; row <= range.to; row++) {
+ selection.add((int) row);
+ }
+ }
+ return selection;
+ }
+
+ private static long covered(List<Range> ranges) {
+ return ranges.stream().mapToLong(Range::count).sum();
+ }
+
+ private static boolean contains(List<Range> ranges, long from, long to) {
+ return ranges.stream().anyMatch(range -> range.from <= from &&
range.to >= to);
+ }
+}
