parthchandra commented on code in PR #6125:
URL: https://github.com/apache/datafusion-comet/pull/6125#discussion_r4403836214
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -347,13 +384,98 @@ struct EagerPageIndexReader {
metadata_cache: Arc<FileMetadataCache>,
metadata_size_hint: Option<usize>,
spark_variant_schema: bool,
+ /// Empty handle whose registration every fetched buffer's reservation
shares.
+ reservation: Option<MemoryReservation>,
+ /// Holds the decode-buffer estimate while this file is being read.
+ _decode_reservation: Option<MemoryReservation>,
require_field_ids: bool,
stamp_int96_leaves: bool,
conversion_check: Option<(SchemaRef, SparkParquetOptions, bool)>,
deferred_rejections: Arc<Mutex<HashMap<Path, (ObjectMeta,
RejectOnNonEmpty)>>>,
deferred_rejection: Option<RejectOnNonEmpty>,
}
+/// Data page size limit that both parquet-mr and parquet-rs default to.
+const DEFAULT_PAGE_SIZE: usize = 1024 * 1024;
+
+/// Estimates what a scan holds while decoding beyond the compressed column
chunks, which
+/// [`ReservedBytes`] charges exactly. arrow-rs allocates the rest internally
with no hook to
+/// observe it: per projected leaf column, a decompressed data page bounded by
the writer's page
+/// size, plus the output batch being assembled. Dictionary pages are left
out: they are often
+/// absent or small, and counting them at the same bound doubled the estimate
against TPC-H
+/// measurements. Files written with larger pages or large dictionaries are
underestimated.
+pub(crate) fn decode_buffer_estimate(projected: &Schema, batch_size: usize) ->
usize {
+ fn leaves(data_type: &DataType) -> usize {
+ match data_type {
+ DataType::Struct(fields) => fields.iter().map(|f|
leaves(f.data_type())).sum(),
+ DataType::List(child)
+ | DataType::LargeList(child)
+ | DataType::FixedSizeList(child, _)
+ | DataType::ListView(child)
+ | DataType::LargeListView(child)
+ | DataType::Map(child, _) => leaves(child.data_type()),
+ _ => 1,
+ }
+ }
+ // Variable-width and nested values have no fixed width; assume a short
string.
+ let row_width: usize = projected
Review Comment:
Small thing on the estimate: `columns` counts leaf columns recursively (via
`leaves`), but `row_width` here sums `primitive_width().unwrap_or(32)` over
only the top-level fields. So a struct adds its real leaf count to the page
term but a flat 32 to the batch term, no matter how many leaves it has. It's
just an estimate and `try_grow` makes a miss harmless, so not a bug — but the
batch-size term under-counts wide nested rows. Worth either walking leaves for
`row_width` too, or a one-line comment saying the batch term is deliberately
coarse for nested types.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -347,13 +384,98 @@ struct EagerPageIndexReader {
metadata_cache: Arc<FileMetadataCache>,
metadata_size_hint: Option<usize>,
spark_variant_schema: bool,
+ /// Empty handle whose registration every fetched buffer's reservation
shares.
+ reservation: Option<MemoryReservation>,
+ /// Holds the decode-buffer estimate while this file is being read.
+ _decode_reservation: Option<MemoryReservation>,
require_field_ids: bool,
stamp_int96_leaves: bool,
conversion_check: Option<(SchemaRef, SparkParquetOptions, bool)>,
deferred_rejections: Arc<Mutex<HashMap<Path, (ObjectMeta,
RejectOnNonEmpty)>>>,
deferred_rejection: Option<RejectOnNonEmpty>,
}
+/// Data page size limit that both parquet-mr and parquet-rs default to.
+const DEFAULT_PAGE_SIZE: usize = 1024 * 1024;
+
+/// Estimates what a scan holds while decoding beyond the compressed column
chunks, which
+/// [`ReservedBytes`] charges exactly. arrow-rs allocates the rest internally
with no hook to
+/// observe it: per projected leaf column, a decompressed data page bounded by
the writer's page
+/// size, plus the output batch being assembled. Dictionary pages are left
out: they are often
+/// absent or small, and counting them at the same bound doubled the estimate
against TPC-H
+/// measurements. Files written with larger pages or large dictionaries are
underestimated.
+pub(crate) fn decode_buffer_estimate(projected: &Schema, batch_size: usize) ->
usize {
+ fn leaves(data_type: &DataType) -> usize {
+ match data_type {
+ DataType::Struct(fields) => fields.iter().map(|f|
leaves(f.data_type())).sum(),
+ DataType::List(child)
Review Comment:
The `leaves` recursion handles
`List`/`LargeList`/`FixedSizeList`/`ListView`/`LargeListView`/`Map`, but the
decode-estimate test only covers `Struct`. A case like `List<Struct<..>>` or a
`Map` would lock in these arms and cover the nested/complex-type path.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -347,13 +384,98 @@ struct EagerPageIndexReader {
metadata_cache: Arc<FileMetadataCache>,
metadata_size_hint: Option<usize>,
spark_variant_schema: bool,
+ /// Empty handle whose registration every fetched buffer's reservation
shares.
+ reservation: Option<MemoryReservation>,
+ /// Holds the decode-buffer estimate while this file is being read.
+ _decode_reservation: Option<MemoryReservation>,
require_field_ids: bool,
stamp_int96_leaves: bool,
conversion_check: Option<(SchemaRef, SparkParquetOptions, bool)>,
deferred_rejections: Arc<Mutex<HashMap<Path, (ObjectMeta,
RejectOnNonEmpty)>>>,
deferred_rejection: Option<RejectOnNonEmpty>,
}
+/// Data page size limit that both parquet-mr and parquet-rs default to.
+const DEFAULT_PAGE_SIZE: usize = 1024 * 1024;
+
+/// Estimates what a scan holds while decoding beyond the compressed column
chunks, which
+/// [`ReservedBytes`] charges exactly. arrow-rs allocates the rest internally
with no hook to
+/// observe it: per projected leaf column, a decompressed data page bounded by
the writer's page
+/// size, plus the output batch being assembled. Dictionary pages are left
out: they are often
+/// absent or small, and counting them at the same bound doubled the estimate
against TPC-H
+/// measurements. Files written with larger pages or large dictionaries are
underestimated.
+pub(crate) fn decode_buffer_estimate(projected: &Schema, batch_size: usize) ->
usize {
+ fn leaves(data_type: &DataType) -> usize {
+ match data_type {
+ DataType::Struct(fields) => fields.iter().map(|f|
leaves(f.data_type())).sum(),
+ DataType::List(child)
+ | DataType::LargeList(child)
+ | DataType::FixedSizeList(child, _)
+ | DataType::ListView(child)
+ | DataType::LargeListView(child)
+ | DataType::Map(child, _) => leaves(child.data_type()),
+ _ => 1,
+ }
+ }
+ // Variable-width and nested values have no fixed width; assume a short
string.
+ let row_width: usize = projected
+ .fields()
+ .iter()
+ .map(|f| f.data_type().primitive_width().unwrap_or(32))
+ .sum();
+ let columns: usize = projected
+ .fields()
+ .iter()
+ .map(|f| leaves(f.data_type()))
+ .sum();
+ columns * DEFAULT_PAGE_SIZE + batch_size * row_width
Review Comment:
`columns * DEFAULT_PAGE_SIZE` assumes a full 1 MiB decompressed page per
leaf column. On a wide table — say 200 columns — that reserves ~200 MiB per
open reader even when the actual pages are small. Under a tight pool that could
throttle task concurrency or push spill pressure onto other operators. The
TPC-H numbers look great, but they're narrow-ish tables; could you sanity-check
the behavior on a very wide schema? Not fatal since the grow is refusable, just
flagging the worst case.
--
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]
