You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "zhuxiangyi (via GitHub)" <gi...@apache.org> on 2026/10/05 12:47:01 UTC
[PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
zhuxiangyi opened a new pull request, #10394:
URL: https://github.com/apache/paimon/pull/10394
### Purpose
**Problem.** A partial-column update of a data evolution table (for example
`MERGE INTO ... UPDATE SET score = ...`) is stored as a separate file that covers the same row id
range as the base file. Reading such a range stitches its files together by row position, so
`DataEvolutionSplitRead` pushes no filter into the format readers of a merged group: dropping rows
in only one of the readers would misalign them (see the class javadoc, and `createUnionReader`,
which builds the format reader with no filters). The file index, and since #10289 a bitmap
selection, can skip rows of a merged group, but the row group statistics and the page indexes of
the Parquet files are not used at all.
As a result, once a query reads an updated column together with any other column of a range, every
filter of the query, including filters on columns that were never updated, is evaluated only after
all rows of the range have been read. The same range stored in one file reads just the pages which
can match. In the benchmark below a 1% range filter reads 29.2 MB when the ranges are merged and
1.8 MB when they are single files.
**Why not the existing options.**
- Compacting a range into one file brings the pushdown back, but rewrites every row of every
updated range, and has to be repeated after the next update. Producing the single file layout of
the benchmark rewrites 408 MB.
- #10289 needs a bitmap file index on each filtered column, and the bitmap index only answers `=`,
`<>`, `IN`, `NOT IN` and null checks, not the range predicates on time or id columns which are
the common selective filters.
**What this PR does.** The rows which may match are computed for the whole group from the
metadata of the files holding the latest values of the filtered columns (Parquet row group
statistics, dictionaries, bloom filters and page indexes), and every file of the group then reads
the same rows through the existing row range path. Every file skips the same pages and cuts the
same rows, so the readers stay aligned.
- `FormatReaderFactory#candidateRowRanges` returns the rows which may match from file metadata
only, or null for formats which can not tell. `ParquetReaderFactory` implements it on top of the
row group and column index filtering of `ParquetFileReader`.
- Only the latest copy of a field may prune: an older copy is stale and would exclude rows which
do match. The winning files are resolved by the logic of the file index pushdown, extracted into
`DataEvolutionSplitRead#winningFieldIds`. A filter is evaluated on one file, so a filter whose
fields are won by different files, such as an OR across them, is not used.
- The candidate ranges are intersected with the row ranges of the split (for example from a global
index) and with the bitmap selection of #10289, which is applied to the same readers. When they
cover the whole group nothing changes, when they are empty the group is skipped.
- The footer read for the candidate rows is reused when the readers of the group open the same
file, through `FormatReaderFactory.Context#metadataCache`.
- Failing to compute the candidate rows reads the whole group, so a lost or corrupt file is still
reported by opening it, respecting `scan.ignore-lost-file` and `scan.ignore-corrupt-file`.
New option `data-evolution.merged-read.stats-pushdown.enabled`, default true.
**Scope and limitations.**
- Parquet only, and groups of normal data files sharing one row id range. Groups with blob or
vector-store files, and tables with nested field evolution, keep the current behavior.
- It pays off only when the filtered column is clustered in write order, as time or id columns
usually are, because pruning is by statistics. Otherwise nothing is pruned, see the benchmark.
- Computing the candidate rows opens each file which wins a filtered column once more and reads
the column index of the filtered columns. See the I/O pattern below.
### Tests
- `ParquetCandidateRowRangesTest`: the ranges cover all matches and prune, file positions across
row groups, no match and no filter, reading the selected rows, the footer read once with a cache.
- `DataEvolutionStatsPushDownTest`, each case compared with the pushdown disabled and with an
in-memory model:
- a stale copy of the filtered column does not exclude matching rows
- the newest of several updates of a column wins
- filters whose fields are won by different files are intersected
- an OR across files
- deletion vectors
- intersection with the row ranges of the split
- a renamed column
- row sidecars enabled
- a lost or a corrupt update file behaves as without the pushdown, with and without
`scan.ignore-lost-file` / `scan.ignore-corrupt-file`
- the pushdown disabled
- 30 rounds of random updates, deletions and filters
- The existing data evolution and Parquet tests pass with the option enabled by default, including
the bitmap pushdown tests of #10289.
**Benchmark.** A table of 4,000,000 rows in 8 row id ranges of 500,000 rows (11 columns, Parquet
with default settings). Every range has one base file and one update file rewriting `ts2`, `rnd2`
and `d0`: `ts2` and `ts` increase with the row id, `rnd2` is random. The "single file" layout is
the same table after compacting every range into one file. Queries read `id`, the filtered column,
`s0` and `d0` with a single thread, and the filter is a range on one column. Apple M4, 16 GB,
macOS, JDK 17, local SSD with a warm page cache. The time is the median of 7 runs after a warm-up,
reported as the median of 3 JVM runs, which differ by a few percent. A counting `FileIO` reports
what is read from the Parquet data files. Matching rows and a checksum of their content are the
same in all three layouts for every query.
| filter | merged, off | merged, on | single file |
|---|---|---|---|
| `ts2` range, 0.1% of rows (updated, clustered) | 59 ms, 29.2 MB | 9 ms, 1.2 MB | 9 ms, 1.2 MB |
| `ts2` range, 1% | 63 ms, 29.2 MB | 10 ms, 1.8 MB | 9 ms, 1.8 MB |
| `ts2` range, 10% | 72 ms, 29.2 MB | 50 ms, 12.3 MB | 38 ms, 12.3 MB |
| `ts2` range, 50% | 199 ms, 73.0 MB | 175 ms, 59.0 MB | 152 ms, 59.0 MB |
| `ts` range, 1% (not updated, clustered) | 29 ms, 14.6 MB | 8 ms, 1.8 MB | 7 ms, 1.8 MB |
| `rnd2` range, 1% (updated, random) | 224 ms, 121.8 MB | 228 ms, 121.8 MB | 196 ms, 121.8 MB |
| no filter | 327 ms, 116.8 MB | 327 ms, 116.8 MB | 284 ms, 116.8 MB |
- The scan already skips the row id ranges whose statistics exclude the filter, so "merged, off"
reads only the ranges overlapping the clustered filters (1 to 5 of the 8). This PR prunes inside
them.
- The gain follows the selectivity and the clustering: up to 6x faster and 24x fewer bytes for a
selective filter on a clustered column, nothing for a random column or without a filter.
- The `ts` row shows that a filter on a column which was never updated also benefits.
I/O pattern, number of reads on the input stream, seeks and opened streams, for the 1% `ts2`
filter and the random filter:
| | merged, off | merged, on | single file |
|---|---|---|---|
| `ts2` 1%: reads / seeks / opens | 34 / 16 / 4 | 3896 / 28 / 6 | 3268 / 22 / 2 |
| `rnd2` 1%: reads / seeks / opens | 128 / 56 / 16 | 4264 / 72 / 24 | 13072 / 80 / 8 |
Skipping pages turns a few large reads into many small ones, as it does for the single file layout,
and computing the candidate rows opens each winning file once more (6 vs 4 streams for 2 groups).
When nothing can be pruned (`rnd2`) the cost is these extra opens and reads, within the run to run
noise here (about 2% of the time). Both are cheap on a local disk. Their effect on an object store depends on the buffering of
its `FileIO` and was not measured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10394:
URL: https://github.com/apache/paimon/pull/10394#issuecomment-6081695274
**[P1] Fix the physical row-sidecar schema before statistics ranges can select it**
I reproduced incorrect query results on `859e1e6392096607f1c295c14e5e51b403cf20cd`. The new `groupRowRanges = candidates` path in `DataEvolutionSplitRead` makes an ordinary filtered merged read eligible for `readTarget`'s row sidecar. Previously, the same query had no explicit row ranges and used Parquet.
This HEAD's `formatBuilder` still adds the virtual row-ID/sequence fields to the physical data schema (`rowTypeWithRowTracking(schema.logicalRowType(), true, true)`). A `.row` file with seven physical INT fields has a one-byte null header; its reader is built as nine fields and consumes a two-byte header, shifting the decoded values.
Actual table reproduction on JDK 8:
1. Create seven nullable INT columns `c0` … `c6`, with row tracking, data evolution and row sidecars enabled, Parquet block size 40000 and page size 1024. Keep the default sidecar selection thresholds.
2. Write 20,000 base rows (`c0=i`, `c1=i`, remaining columns `i+columnIndex`). Commit a projected update of only `c1=19999-i`, covering the original row-ID range, as a partial-column MERGE does.
3. Read the projection `c0,c1` with predicate `c1 < 100`, without supplying any row ranges.
With statistics pushdown disabled, all 100 rows match the oracle: `19900:99`, `19901:98`, `19902:97`, etc. With the new default enabled, 100 rows are returned but `c0` is corrupted (for example `-100663219:37`, `-1006632883:91`). Setting the sidecar selection ratio to `0.000001`, so the same statistics ranges are read from Parquet, restores the exact expected results with pushdown still enabled.
The existing sidecar test uses four physical columns, so adding two virtual fields does not cross the null-header size boundary. Please build row-sidecar readers from the actual physical schema, or keep this new path from selecting them until that fix is present, and add this wider-table regression. The optimization has clear value; this wrong-result path needs to be resolved before enabling it by default.
Validation: the seven targeted existing/new core and Parquet test classes passed all 131 tests with standard Maven checks. The additional table reproduction failed only with the new path enabled; its Parquet control passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10394:
URL: https://github.com/apache/paimon/pull/10394#issuecomment-6217104910
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on PR #10394:
URL: https://github.com/apache/paimon/pull/10394#issuecomment-6160253309
Thanks @JingsongLi, confirmed and fixed.
The candidate rows were passed down as row ranges, which also decide whether `readTarget` reads the row sidecar, so filtered reads started to select it. They now restrict the selection the group already shares for the bitmap index (#10289), and the row ranges are unchanged, so this PR no longer affects sidecar selection. `testWideTableWithRowSidecar` covers your setup and fails for 7, 8, 15 and 16 columns before the fix.
The decoding bug itself also exists on master: a `withRowRanges` read with the pushdown disabled returns the same shifted values. #10415 fixes it (verified on 1369458e78), so I left it there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10394:
URL: https://github.com/apache/paimon/pull/10394#discussion_r4291197779
##########
paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java:
##########
@@ -305,14 +324,41 @@ private RecordReader<InternalRow> createReader(
groupSelection =
buildGroupSelection(needMergeFiles, fileIndexResults);
}
- return createUnionReader(
- needMergeFiles,
- partition,
- dataFilePathFactory,
- rowRanges,
- readRowType,
- deletionVector,
- groupSelection);
+ FileMetadataCache metadataCache = new FileMetadataCache();
+ List<Range> candidates =
+ candidateRanges(
+ filters,
+ needMergeFiles,
+ dataFilePathFactory,
+ metadataCache);
+ // Candidates restrict the group selection, not the row ranges of the
+ // split, which also decide whether a row sidecar is read instead.
+ 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<>();
+ }
+ }
+ // the readers of a group are created right away, see #candidateRanges
+ groupMetadataCache = metadataCache;
Review Comment:
Could we keep this cache local to the merged-group construction and pass it explicitly through `createUnionReader` → `createFieldBunchReader` → `createFileReader`? It is created for one group and cleared immediately after its readers are constructed, so it does not need to be instance state. Using the field as a temporary parameter makes `createFileReader` depend on the caller having set it and on reader creation remaining synchronous. Explicit parameter passing would make the ownership and lifetime clear and remove the set/reset `try/finally`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #10394:
URL: https://github.com/apache/paimon/pull/10394
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Push filters into merged data evolution groups by file statistics [paimon]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on code in PR #10394:
URL: https://github.com/apache/paimon/pull/10394#discussion_r4291829205
##########
paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java:
##########
@@ -305,14 +324,41 @@ private RecordReader<InternalRow> createReader(
groupSelection =
buildGroupSelection(needMergeFiles, fileIndexResults);
}
- return createUnionReader(
- needMergeFiles,
- partition,
- dataFilePathFactory,
- rowRanges,
- readRowType,
- deletionVector,
- groupSelection);
+ FileMetadataCache metadataCache = new FileMetadataCache();
+ List<Range> candidates =
+ candidateRanges(
+ filters,
+ needMergeFiles,
+ dataFilePathFactory,
+ metadataCache);
+ // Candidates restrict the group selection, not the row ranges of the
+ // split, which also decide whether a row sidecar is read instead.
+ 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<>();
+ }
+ }
+ // the readers of a group are created right away, see #candidateRanges
+ groupMetadataCache = metadataCache;
Review Comment:
Done in 65d77e2fd7: the cache is created per merged group and passed explicitly through `createUnionReader` → `createFieldBunchReader` → `createFileReader`, so the instance field and the `try/finally` are gone. Only the normal-file path uses it; the single-file, blob and vector paths pass null. `testFooterReadOncePerFile` checks that each file of a merged group reads its footer once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org