sunchao commented on code in PR #7094:
URL: https://github.com/apache/datafusion-comet/pull/7094#discussion_r4315143637
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -750,6 +788,164 @@ impl AsyncFileReader for EagerPageIndexReader {
}
}
+/// Loads a file's metadata through `metadata_cache`, with the page index as
`page_index_policy`
+/// asks, counting the metadata and footer bytes it reads from storage.
Returns the metadata
+/// together with the number of storage reads it took, which is zero when the
cache already held
+/// the metadata.
+async fn load_metadata(
+ store: Arc<dyn ObjectStore>,
+ scan_io_metrics: Arc<ScanIoMetrics>,
+ metadata_cache: Arc<FileMetadataCache>,
+ object_meta: &ObjectMeta,
+ metadata_size_hint: Option<usize>,
+ file_decryption_properties: Option<Arc<FileDecryptionProperties>>,
+ page_index_policy: Option<PageIndexPolicy>,
+) -> ParquetResult<(Arc<ParquetMetaData>, usize)> {
+ let cache_enabled = file_decryption_properties.is_none();
+ let storage_reads = Arc::new(AtomicUsize::new(0));
+ let footer_payload_bytes = Arc::new(AtomicUsize::new(0));
+ let metadata_store = ScanIoObjectStore {
+ inner: store,
+ scan_io_metrics,
+ role: ScanIoStoreRole::Metadata {
+ storage_reads: Arc::clone(&storage_reads),
+ footer_payload_bytes: Arc::clone(&footer_payload_bytes),
+ footer_payload: OnceLock::new(),
+ footer_recorded: AtomicBool::new(false),
+ file_size: object_meta.size,
+ record_footer_immediately: !cache_enabled,
+ },
+ };
+
+ let metadata = DFParquetMetadata::new(&metadata_store, object_meta)
+ .with_decryption_properties(file_decryption_properties)
+ .with_file_metadata_cache(Some(metadata_cache))
+ .with_metadata_size_hint(metadata_size_hint)
+ .with_page_index_policy(page_index_policy)
+ .fetch_metadata()
+ .await
+ .map_err(|e| {
+ ParquetError::General(format!(
+ "Failed to fetch metadata for file {}: {e}",
+ object_meta.location,
+ ))
+ });
+
+ if metadata.is_ok() {
+ let footer_bytes = footer_payload_bytes.load(Ordering::Relaxed);
+ if footer_bytes > 0 && cache_enabled {
+ metadata_store.record_footer(footer_bytes);
+ }
+ } else if cache_enabled {
+ metadata_store.record_valid_footer();
+ }
+ Ok((metadata?, storage_reads.load(Ordering::Relaxed)))
+}
+
+/// A load of a file's metadata started before DataFusion opens the file.
Dropping it aborts the
+/// load.
+/// Its result carries the number of storage reads it took, for the open that
uses it to record.
+type MetadataLoad = SpawnedTask<Option<(Arc<ParquetMetaData>, usize)>>;
+
+/// The scan-wide state of `with_next_file_metadata_prefetch`.
+#[derive(Debug)]
+struct NextFileMetadataPrefetch {
+ store: Arc<dyn ObjectStore>,
+ scan_io_metrics: Arc<ScanIoMetrics>,
+ metadata_cache: Arc<FileMetadataCache>,
+ /// For each file of the group, keyed by path and byte range, the next
file with another path
+ /// and that file's metadata size hint. Further ranges of a file's own
path are skipped,
+ /// because opening the file already loads their metadata.
+ next_files: HashMap<(Path, Option<FileRange>), (ObjectMeta,
Option<usize>)>,
+ /// The load in flight, with the path it loads. Each open replaces it
unless it is for the
+ /// next file, so a load for a file DataFusion never opens, such as one
pruned before its
+ /// open, is dropped at the next open.
+ in_flight: Mutex<Option<(Path, MetadataLoad)>>,
+}
+
+impl NextFileMetadataPrefetch {
+ fn new(
+ files: &[PartitionedFile],
+ store: Arc<dyn ObjectStore>,
+ scan_io_metrics: Arc<ScanIoMetrics>,
+ metadata_cache: Arc<FileMetadataCache>,
+ ) -> Self {
+ let mut next_files = HashMap::new();
+ for (position, file) in files.iter().enumerate() {
+ let location = &file.object_meta.location;
+ let next = files[position + 1..]
Review Comment:
[P2] Could this successor map be built in one reverse pass or once per
contiguous path run? With prefetch enabled, Spark 3.5+ can repack thousands of
adjacent ranges of one file into a task using
`spark.sql.files.maxPartitionBytes=1m` and `spark.sql.files.maxPartitionNum=1`.
Each range here searches the remaining suffix again. For 40,000 ranges of the
same path, construction performs 799,980,000 path comparisons and produces an
empty map because there is no next file. The isolated reproduction took about
4.15 seconds, versus under 1 ms for a reverse pass. This adds seconds before
the first batch, including for a limited scan that consumes little data, while
providing no prefetch benefit. Linear construction would preserve the intended
adjacent-range behavior without this setup regression.
Evidence: Verified that `CometScanExec` splits files before calling Spark's
`FilePartition.getFilePartitions`, whose Spark 3.5.9/4.0.4/4.1.3/4.2.0
implementations repack existing ranges without merging them. A source-based
packing reproduction maps 40,000 1-MiB ranges to one partition with
maxPartitionNum=1 and the default 4-MiB open cost.
`/tmp/comet7094-actual-types-probe.rs` contains the byte-for-byte head loop
from lines 873–884, using actual DataFusion 55.2 `PartitionedFile` and
object-store types. The optimized harness measured 129.7 ms for 8,192 ranges,
866.9 ms for 20,000, and 4.149 s for 40,000, all yielding zero successors. A
reverse-pass comparison took 0.810 ms for 40,000. Dependencies were from the
local debug build, and no end-to-end Spark timing is claimed.
--
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]
