peter-toth commented on code in PR #59639:
URL: https://github.com/apache/spark/pull/59639#discussion_r4339163557
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -730,12 +800,18 @@ private RowRanges evaluateStorageFilter(
splicedBytes = 0L;
}
}
- if (splicedBytes + rangeBytes > cap) {
+ if (splicedBytes + rangeCharge > cap) {
// The buffer goes first, which is a no-op once splicing is given up.
abandonSplicing();
splicedBytes = 0L;
// The ranges would be thrown away, so stop.
- if (rangeBytes > cap) return null;
+ if (rangeCharge > cap) return null;
+ }
+ if (!reserveSurvivorMemory((accountKeys ? splicedBytes : 0L) +
rangeCharge)) {
+ // A reservation the task's memory manager denies is handled as
passing the cap is.
+ abandonSplicing();
+ splicedBytes = 0L;
+ if (!reserveSurvivorMemory(rangeCharge)) return null;
Review Comment:
Done in
[778c14954a9](https://github.com/apache/spark/pull/59639/commits/778c14954a9ede1af0fc6357b0b5546322e69a23).
I kept the give-up, as viirya suggests, and made it visible:
- new metrics count the row groups the budget gave up, by the cap and by the
memory manager;
- the first of each kind per split is logged at INFO.
A denied request with no key values in it now gives the filter up at once,
rather than repeating the acquire. That covers off heap, and a row group that
already gave splicing up.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -1977,19 +1978,26 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
private def marked(keys: Seq[Long], checked: Boolean): Seq[(Long, Boolean)] =
keys.map(_ -> checked)
+ // 40 rows of a long key and a string value in two row groups of 20. A
filter of `k >= 10` keeps
+ // one range of 10 rows in the first and all 20 rows of the second. A
buffered long key costs 9
+ // bytes and a range 40, so the first row group holds 130 bytes and the
second 220.
+ private def twoRowGroupsOf20(dir: File): String = {
Review Comment:
Done in
[778c14954a9](https://github.com/apache/spark/pull/59639/commits/778c14954a9ede1af0fc6357b0b5546322e69a23):
- "a key vector that cannot grow, or a codec's InternalError, gives the row
group up" now calls `twoRowGroupsOf20`;
- `readWithSpill` went with the spill;
- the denial test's `bytes` reads through `readUnderCap`, which now takes an
optional task context.
--
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]
