jayzhan211 commented on code in PR #26407:
URL: https://github.com/apache/datafusion/pull/26407#discussion_r4370478290
##########
datafusion/datasource-parquet/src/file_format.rs:
##########
@@ -521,22 +549,51 @@ impl FileFormat for ParquetFormat {
.await?;
let file_metadata_cache =
state.runtime_env().cache_manager.get_file_metadata_cache();
- let metadata = DFParquetMetadata::new(store, object)
- .with_metadata_size_hint(self.metadata_size_hint())
- .with_decryption_properties(file_decryption_properties)
- .with_file_metadata_cache(Some(file_metadata_cache))
- .fetch_metadata()
- .await?;
- let statistics = DFParquetMetadata::statistics_from_parquet_metadata(
- &metadata,
- &table_schema,
- )?;
- let ordering =
- crate::metadata::ordering_from_parquet_metadata(&metadata,
&table_schema)?;
- Ok(
- datafusion_datasource::file_format::FileMeta::new(statistics)
- .with_ordering(ordering),
- )
+ let metadata_size_hint = self.metadata_size_hint();
+
+ // Decoding the footer and converting its statistics is CPU bound and
+ // grows with the number of columns. Callers such as `ListingTable` run
+ // this for many files concurrently from a single planning task, so run
+ // it in its own task to spread that work over the runtime's threads
+ // instead of decoding every file on the thread polling the planner.
+ // As in `spawn_buffered`, only spawn from a multi-thread tokio
runtime.
+ match tokio::runtime::Handle::try_current() {
Review Comment:
nit: the two arms repeat the same `fetch_stats_and_ordering` call. Building
the future once and choosing whether to spawn it keeps a single copy (this
compiles and passes clippy/fmt locally):
```diff
- match tokio::runtime::Handle::try_current() {
- Ok(handle)
- if handle.runtime_flavor()
- == tokio::runtime::RuntimeFlavor::MultiThread =>
- {
- let store = Arc::clone(store);
- let object = object.clone();
- SpawnedTask::spawn(async move {
- fetch_stats_and_ordering(
- store.as_ref(),
- &object,
- metadata_size_hint,
- file_decryption_properties,
- file_metadata_cache,
- &table_schema,
- )
- .await
- })
- .join_unwind()
- .await
- .map_err(|e| DataFusionError::ExecutionJoin(Box::new(e)))?
- }
- // Otherwise do the work on this task, where the caller polls it
- // together with the other files (for example in
`ListingTable`'s
- // `buffer_unordered`).
- _ => {
- fetch_stats_and_ordering(
- store.as_ref(),
- object,
- metadata_size_hint,
- file_decryption_properties,
- file_metadata_cache,
- &table_schema,
- )
- .await
- }
- }
+ let store = Arc::clone(store);
+ let object = object.clone();
+ let fetch = async move {
+ fetch_stats_and_ordering(
+ store.as_ref(),
+ &object,
+ metadata_size_hint,
+ file_decryption_properties,
+ file_metadata_cache,
+ &table_schema,
+ )
+ .await
+ };
+ match tokio::runtime::Handle::try_current() {
+ Ok(handle)
+ if handle.runtime_flavor()
+ == tokio::runtime::RuntimeFlavor::MultiThread =>
+ {
+ SpawnedTask::spawn(fetch)
+ .join_unwind()
+ .await
+ .map_err(|e|
DataFusionError::ExecutionJoin(Box::new(e)))?
+ }
+ // Otherwise do the work on this task, where the caller polls it
+ // together with the other files.
+ _ => fetch.await,
+ }
```
--
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]
