viirya commented on PR #58895:
URL: https://github.com/apache/spark/pull/58895#issuecomment-6064730845
Thanks for landing this, @peter-toth. A few follow-ups after reading the
merged version. None of them is a correctness concern.
**Where the feature can pay off in practice**
I think the set of real workloads this helps is narrower than the benchmark
suggests, and it would help to say so, or to measure it, before anyone
considers changing the default.
- **The bloom has to exist first.** `InjectRuntimeFilter` only injects one
when the join is probably a shuffle join, or the application side already has a
shuffle below it, and the application-side scan is at least 10GB by default. A
typical fact table broadcast-joined with a filtered dimension gets no bloom, so
it gets nothing from this PR.
- **The survivors need physical locality.** A page is the unit that can be
skipped, and with default writer settings a narrow column's page holds up to
20,000 rows. With survivors spread at random at rate p, a page has no survivor
with probability about e^(-p·20000), so page skipping needs p well below
1/20,000, and the bloom's false positives count toward p. In practice the gain
comes from row groups or runs of pages with no survivor at all. That needs the
surviving keys to form a range in the table's physical order, or each key to
own whole pages. Hash bucketing does not give that. The most natural candidate,
a date key, is usually handled already: by DPP when the table is partitioned on
it, and by broadcast when the date dimension is small.
- **The tables most likely to be clustered on a join key are out of reach.**
Delta (Z-order, liquid clustering) reads through `DeltaParquetFileFormat`, a
subclass that the exact-class check excludes. Iceberg (sort orders) goes
through DSv2. The conf doc says "V1 Parquet file source", and a Delta user will
likely read that as covering them. It may be worth stating the subclass
exclusion there, and later offering subclasses an explicit opt-in.
- **The benchmark sits at the favorable end of all of this, and on local
storage.**
- The fact table is `spark.range` written as a single file, so it is fully
sorted on `k`.
- The "clustered" dimension selects one contiguous key range.
- Row groups are capped at 1,500 and 42,000 rows (~3MB). With the default
128MB the whole ~100MB table would be a single row group, so the clustered
cases' 33-of-34 row-group skips would not happen.
- The bloom is sized at 13 bits per item, injection is forced (broadcast
off, application-side threshold 0), and it runs on `local[1]` against a local
filesystem.
The representative rows are "scattered" with default pages: the join is
1.0–1.4x on the wide table and 0.9–1.0x on the narrow one.
My main worry is object storage, where most of these tables live. There, a
plain read fetches a row group's columns in one vectored read. This reader
issues two, phase 1 for the keys and phase 2 for the rest, and with no pushed
data filter phase 2 also reads the offset indexes first. For a row group where
the filter prunes little, that is extra request latency with nothing saved, and
a local SSD hides it. So the scattered rows above may look worse there, and the
clustered ones may gain less than the byte counts suggest. If you get a chance,
a run on object storage would tell us a lot, ideally TPC-DS (or similar) at a
realistic scale with default write settings and default injection thresholds,
noting which queries get `StorageFilters` and how they change.
**Survivor buffer memory**
The survivor buffer (up to 64MB per reader, so N times that per executor)
sits outside any `MemoryConsumer`. Registering it would let an allocation be
denied, and `spill()` could simply release the buffer and fall back to
re-reading the keys.
**Comments**
About 1,000 of the ~2,400 added non-test lines are comments, much of it
narrative (alternatives considered, past measurements, the same point restated
across `SQLConf`, `ParquetStorageFilter` and the reader). A cleanup that keeps
only the invariants and non-obvious trade-offs would make the reader much
easier to maintain.
**The public conf doc**
`spark.sql.parquet.storageFilterPushdown.enabled` is user-facing, but its
doc is ~25 lines of internals: phases, reading key columns twice, when the
reader gives up. I'd keep only what changes a user's decision or answer.
Something like:
> If true, the vectorized reader of the V1 Parquet data source may use a
runtime filter on join keys, such as a runtime bloom filter, to skip reading
the pages and row groups of other columns that hold no matching rows. The
filter is still applied after the scan, so results do not change, with two
exceptions: with spark.sql.files.ignoreCorruptFiles=true the rows returned from
a corrupt file can differ, and a file with an incorrect Parquet page index can
return wrong results (set parquet.filter.columnindex.enabled=false to stop
relying on it). Only ParquetFileFormat itself is supported, not its subclasses.
It can be slower when matching rows are spread across most pages. Matching key
values are buffered per row group, up to
spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes, outside
Spark's memory manager.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
