You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by al...@apache.org on 2023/12/22 19:28:53 UTC
(arrow-datafusion) branch main updated: Add new configuration item `listing_table_ignore_subdirectory` (#8565)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new ba46434f83 Add new configuration item `listing_table_ignore_subdirectory` (#8565)
ba46434f83 is described below
commit ba46434f839d612be01ee0d00e0d826475ce5f10
Author: Asura7969 <14...@qq.com>
AuthorDate: Sat Dec 23 03:28:47 2023 +0800
Add new configuration item `listing_table_ignore_subdirectory` (#8565)
* init
* test
* add config
* rename
* doc
* fix doc
* add sqllogictests & rename
* fmt & fix test
* clippy
* test read partition table
* simplify testing
* simplify testing
---
datafusion/common/src/config.rs | 5 +++
datafusion/core/src/datasource/listing/helpers.rs | 4 +--
datafusion/core/src/datasource/listing/url.rs | 26 ++++++++++++---
datafusion/core/src/execution/context/parquet.rs | 9 +++--
.../sqllogictest/test_files/information_schema.slt | 2 ++
datafusion/sqllogictest/test_files/parquet.slt | 38 +++++++++++++++++++++-
docs/source/user-guide/configs.md | 1 +
7 files changed, 75 insertions(+), 10 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index 03fb5ea320..dedce74ff4 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -273,6 +273,11 @@ config_namespace! {
/// memory consumption
pub max_buffered_batches_per_output_file: usize, default = 2
+ /// When scanning file paths, whether to ignore subdirectory files,
+ /// ignored by default (true), when reading a partitioned table,
+ /// `listing_table_ignore_subdirectory` is always equal to false, even if set to true
+ pub listing_table_ignore_subdirectory: bool, default = true
+
}
}
diff --git a/datafusion/core/src/datasource/listing/helpers.rs b/datafusion/core/src/datasource/listing/helpers.rs
index be74afa1f4..68de55e1a4 100644
--- a/datafusion/core/src/datasource/listing/helpers.rs
+++ b/datafusion/core/src/datasource/listing/helpers.rs
@@ -375,10 +375,10 @@ pub async fn pruned_partition_list<'a>(
store.list(Some(&partition.path)).try_collect().await?
}
};
-
let files = files.into_iter().filter(move |o| {
let extension_match = o.location.as_ref().ends_with(file_extension);
- let glob_match = table_path.contains(&o.location);
+ // here need to scan subdirectories(`listing_table_ignore_subdirectory` = false)
+ let glob_match = table_path.contains(&o.location, false);
extension_match && glob_match
});
diff --git a/datafusion/core/src/datasource/listing/url.rs b/datafusion/core/src/datasource/listing/url.rs
index 3ca7864f7f..766dee7de9 100644
--- a/datafusion/core/src/datasource/listing/url.rs
+++ b/datafusion/core/src/datasource/listing/url.rs
@@ -20,6 +20,7 @@ use std::fs;
use crate::datasource::object_store::ObjectStoreUrl;
use crate::execution::context::SessionState;
use datafusion_common::{DataFusionError, Result};
+use datafusion_optimizer::OptimizerConfig;
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use glob::Pattern;
@@ -184,14 +185,27 @@ impl ListingTableUrl {
}
/// Returns `true` if `path` matches this [`ListingTableUrl`]
- pub fn contains(&self, path: &Path) -> bool {
+ pub fn contains(&self, path: &Path, ignore_subdirectory: bool) -> bool {
match self.strip_prefix(path) {
Some(mut segments) => match &self.glob {
Some(glob) => {
- let stripped = segments.join("/");
- glob.matches(&stripped)
+ if ignore_subdirectory {
+ segments
+ .next()
+ .map_or(false, |file_name| glob.matches(file_name))
+ } else {
+ let stripped = segments.join("/");
+ glob.matches(&stripped)
+ }
+ }
+ None => {
+ if ignore_subdirectory {
+ let has_subdirectory = segments.collect::<Vec<_>>().len() > 1;
+ !has_subdirectory
+ } else {
+ true
+ }
}
- None => true,
},
None => false,
}
@@ -223,6 +237,8 @@ impl ListingTableUrl {
store: &'a dyn ObjectStore,
file_extension: &'a str,
) -> Result<BoxStream<'a, Result<ObjectMeta>>> {
+ let exec_options = &ctx.options().execution;
+ let ignore_subdirectory = exec_options.listing_table_ignore_subdirectory;
// If the prefix is a file, use a head request, otherwise list
let list = match self.is_collection() {
true => match ctx.runtime_env().cache_manager.get_list_files_cache() {
@@ -246,7 +262,7 @@ impl ListingTableUrl {
.try_filter(move |meta| {
let path = &meta.location;
let extension_match = path.as_ref().ends_with(file_extension);
- let glob_match = self.contains(path);
+ let glob_match = self.contains(path, ignore_subdirectory);
futures::future::ready(extension_match && glob_match)
})
.map_err(DataFusionError::ObjectStore)
diff --git a/datafusion/core/src/execution/context/parquet.rs b/datafusion/core/src/execution/context/parquet.rs
index 5d649d3e6d..7825d9b882 100644
--- a/datafusion/core/src/execution/context/parquet.rs
+++ b/datafusion/core/src/execution/context/parquet.rs
@@ -80,6 +80,7 @@ mod tests {
use crate::dataframe::DataFrameWriteOptions;
use crate::parquet::basic::Compression;
use crate::test_util::parquet_test_data;
+ use datafusion_execution::config::SessionConfig;
use tempfile::tempdir;
use super::*;
@@ -103,8 +104,12 @@ mod tests {
#[tokio::test]
async fn read_with_glob_path_issue_2465() -> Result<()> {
- let ctx = SessionContext::new();
-
+ let config =
+ SessionConfig::from_string_hash_map(std::collections::HashMap::from([(
+ "datafusion.execution.listing_table_ignore_subdirectory".to_owned(),
+ "false".to_owned(),
+ )]))?;
+ let ctx = SessionContext::new_with_config(config);
let df = ctx
.read_parquet(
// it was reported that when a path contains // (two consecutive separator) no files were found
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 5c6bf6e2da..36876beb14 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -150,6 +150,7 @@ datafusion.execution.aggregate.scalar_update_factor 10
datafusion.execution.batch_size 8192
datafusion.execution.coalesce_batches true
datafusion.execution.collect_statistics false
+datafusion.execution.listing_table_ignore_subdirectory true
datafusion.execution.max_buffered_batches_per_output_file 2
datafusion.execution.meta_fetch_concurrency 32
datafusion.execution.minimum_parallel_output_files 4
@@ -224,6 +225,7 @@ datafusion.execution.aggregate.scalar_update_factor 10 Specifies the threshold f
datafusion.execution.batch_size 8192 Default batch size while creating new batches, it's especially useful for buffer-in-memory batches since creating tiny batches would result in too much metadata memory consumption
datafusion.execution.coalesce_batches true When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting
datafusion.execution.collect_statistics false Should DataFusion collect statistics after listing files
+datafusion.execution.listing_table_ignore_subdirectory true When scanning file paths, whether to ignore subdirectory files, ignored by default (true), when reading a partitioned table, `listing_table_ignore_subdirectory` is always equal to false, even if set to true
datafusion.execution.max_buffered_batches_per_output_file 2 This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption
datafusion.execution.meta_fetch_concurrency 32 Number of files to read in parallel when inferring schema and statistics
datafusion.execution.minimum_parallel_output_files 4 Guarantees a minimum level of output files running in parallel. RecordBatches will be distributed in round robin fashion to each parallel writer. Each writer is closed and a new file opened once soft_max_rows_per_output_file is reached.
diff --git a/datafusion/sqllogictest/test_files/parquet.slt b/datafusion/sqllogictest/test_files/parquet.slt
index 6c3bd68770..0f26c14f00 100644
--- a/datafusion/sqllogictest/test_files/parquet.slt
+++ b/datafusion/sqllogictest/test_files/parquet.slt
@@ -276,6 +276,39 @@ LIMIT 10;
0 2014-08-27T14:00:00Z Timestamp(Millisecond, Some("UTC"))
0 2014-08-27T14:00:00Z Timestamp(Millisecond, Some("UTC"))
+# Test config listing_table_ignore_subdirectory:
+
+query ITID
+COPY (SELECT * FROM src_table WHERE int_col > 6 LIMIT 3)
+TO 'test_files/scratch/parquet/test_table/subdir/3.parquet'
+(FORMAT PARQUET, SINGLE_FILE_OUTPUT true);
+----
+3
+
+statement ok
+CREATE EXTERNAL TABLE listing_table
+STORED AS PARQUET
+WITH HEADER ROW
+LOCATION 'test_files/scratch/parquet/test_table/*.parquet';
+
+statement ok
+set datafusion.execution.listing_table_ignore_subdirectory = true;
+
+# scan file: 0.parquet 1.parquet 2.parquet
+query I
+select count(*) from listing_table;
+----
+9
+
+statement ok
+set datafusion.execution.listing_table_ignore_subdirectory = false;
+
+# scan file: 0.parquet 1.parquet 2.parquet 3.parquet
+query I
+select count(*) from listing_table;
+----
+12
+
# Clean up
statement ok
DROP TABLE timestamp_with_tz;
@@ -303,7 +336,6 @@ NULL
statement ok
DROP TABLE single_nan;
-
statement ok
CREATE EXTERNAL TABLE list_columns
STORED AS PARQUET
@@ -319,3 +351,7 @@ SELECT int64_list, utf8_list FROM list_columns
statement ok
DROP TABLE list_columns;
+
+# Clean up
+statement ok
+DROP TABLE listing_table;
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index 6fb5cc4ca8..1f7fa7760b 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -82,6 +82,7 @@ Environment variables are read during `SessionConfig` initialisation so they mus
| datafusion.execution.minimum_parallel_output_files | 4 | Guarantees a minimum level of output files running in parallel. RecordBatches will be distributed in round robin fashion to each parallel writer. Each writer is closed and a new file opened once soft_max_rows_per_output_file is reached. [...]
| datafusion.execution.soft_max_rows_per_output_file | 50000000 | Target number of rows in output files when writing multiple. This is a soft max, so it can be exceeded slightly. There also will be one file smaller than the limit if the total number of rows written is not roughly divisible by the soft max [...]
| datafusion.execution.max_buffered_batches_per_output_file | 2 | This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption [...]
+| datafusion.execution.listing_table_ignore_subdirectory | true | When scanning file paths, whether to ignore subdirectory files, ignored by default (true), when reading a partitioned table, `listing_table_ignore_subdirectory` is always equal to false, even if set to true [...]
| datafusion.optimizer.enable_distinct_aggregation_soft_limit | true | When set to true, the optimizer will push a limit operation into grouped aggregations which have no aggregate expressions, as a soft limit, emitting groups once the limit is reached, before all rows in the group are read. [...]
| datafusion.optimizer.enable_round_robin_repartition | true | When set to true, the physical plan optimizer will try to add round robin repartitioning to increase parallelism to leverage more CPU cores [...]
| datafusion.optimizer.enable_topk_aggregation | true | When set to true, the optimizer will attempt to perform limit operations during aggregations, if possible [...]