Inside DataFusion's ListingTable: Listing an Object Store in Parallel and Turning Files into FileGroups

TL;DR: Listing a table is a four-stage pipeline: (1) turn the filter into a path prefix so the object store can skip whole directories, (2) list the prefix and decode Hive-style partition values out of each path, (3) fetch per-file metadata (statistics, ordering) concurrently with meta_fetch_concurrency, and (4) pack the surviving files into FileGroups — one per target_partitions. A LIMIT can cut the pipeline short after stage 3, which is the trick that makes SELECT * FROM t LIMIT 1 cheap on a million-file table.

Everything above the physical plan is a lie of omission. When you write SELECT * FROM 's3://bucket/events/' WHERE dt = '2024-01-01', the planner needs to know which files that resolves to before it can build a plan. That question is answered by the listing layer, and it is where a surprising amount of DataFusion’s real-world performance is won or lost.

The relevant code lives in three crates: datafusion/catalog/ (the TableProvider abstraction), datafusion/catalog-listing/ (ListingTable, the Hive-partitioned file-backed table), and datafusion/datasource/ (PartitionedFile, FileGroup, FileScanConfig).

The contract: `TableProvider`

The catalog is a trait, not a class hierarchy. Everything a table must be able to do is expressed in TableProvider:

rust
// datafusion/session/src/table.rs:55-57, 193-199
pub trait TableProvider: Any + Debug + Sync + Send {
    /// Get a reference to the schema for this table
    fn schema(&self) -> SchemaRef;
    ...
    /// Create an [`ExecutionPlan`] for scanning the table with optional
    /// `projection`, `filter`, and `limit`, described below.
    async fn scan(
        &self,
        state: &dyn Session,
        projection: Option<&[usize]>,
        filters: &[Expr],
        limit: Option<usize>,
    ) -> Result<Arc<dyn ExecutionPlan>>;
}

The scan method’s contract is unusually well specified — the doc comment spells out the evaluation order (filters, then limit, then projection), warns that filters may reference columns absent from the projection, and notes that “if any pushed-down filters are Inexact, the LIMIT cannot be pushed down” because inexact filters don’t guarantee that every filtered row was removed. ListingTable overrides the richer scan_with_args variant so it also gets the skip (OFFSET) argument.

Stage 0: split the filter

Before listing anything, ListingTable sorts the incoming filters into two buckets:

rust
// datafusion/catalog-listing/src/table.rs:646-651
// If the filters can be resolved using only partition cols, there is no need to
// pushdown it to TableScan, otherwise, `unhandled` pruning predicates will be generated
let (partition_filters, filters): (Vec<_>, Vec<_>) =
    filters.iter().cloned().partition(|filter| {
        can_be_evaluated_for_partition_pruning(&table_partition_col_names, filter)
    });

Partition-only filters go to the listing stage (they can prune whole directories and whole files). Everything else stays as a data filter for the scan, handled by the pruning machinery.

There’s also a subtlety about LIMIT:

rust
// datafusion/catalog-listing/src/table.rs:629-662
// The scan must read enough rows to satisfy `skip + limit`, not
// just `limit`, before any rows are skipped below.
let inflated_limit = limit.map(|l| l.saturating_add(skip.unwrap_or(0)));
...
// We should not limit files before assigning declared output partitions
// or before applying non-partition filters.
let statistic_file_limit =
    if filters.is_empty() && declared_output_partitioning.is_none() {
        inflated_limit
    } else {
        None
    };

The file limit is only pushed into the listing when there are no data filters to apply — a file that fails a data filter still has to be counted against the limit’s row budget. Get this wrong and LIMIT 10 returns fewer than 10 rows.

Stage 1: turn the filter into a path prefix

This is the highest-leverage trick in the whole listing layer. Hive-partitioned tables encode partition values as directory names:

text

s3://bucket/events/dt=2024-01-01/region=us-east-1/part-0.parquet
s3://bucket/events/dt=2024-01-02/region=us-west-2/part-0.parquet

A filter dt = '2024-01-01' can be turned into the prefix dt=2024-01-01, and then the object store’s LIST operation never returns the other day’s files at all. No request, no parsing, no filtering.

rust
// datafusion/catalog-listing/src/helpers.rs:262-305
pub fn evaluate_partition_prefix<'a>(
    partition_cols: &'a [(String, DataType)],
    filters: &'a [Expr],
) -> Option<Path> {
    let mut partition_values = HashMap::new();
    for filter in filters {
        populate_partition_values(&mut partition_values, filter);
    }
    if partition_values.is_empty() {
        return None;
    }
    let mut parts = Vec::with_capacity(partition_cols.len());
    for (p, _) in partition_cols {
        match partition_values.get(p.as_str()) {
            Some(PartitionValue::Single(val)) => {
                // if a partition only has a single literal value, then it can be added to the
                // prefix
                let encoded = encode_partition_value(val);
                if encoded != val.as_str() {
                    // The same decoded value can be represented by both raw and
                    // percent-encoded partition directories. Prefix pruning is
                    // an optimization, so stop before this partition rather
                    // than listing only one spelling and potentially skipping
                    // valid rows.
                    break;
                }
                parts.push(format!("{p}={encoded}"));
            }
            _ => {
                // break on the first unconstrainted partition to create a common prefix
                // for all covered partitions.
                break;
            }
        }
    }
    if parts.is_empty() { None } else { Some(Path::from_iter(parts)) }
}

Three details worth pausing on:

Only equality constraints count. populate_partition_values only recognises Operator::Eq against a literal, and only AND compositions. A dt > '2024-01-01' filter yields no prefix, because there’s no way to express “all directories after this one” as a LIST prefix.

The prefix is a common prefix, so it stops at the first unconstrained column. If partition_cols is [dt, region] and you filter only on dt, you get prefix dt=2024-01-01 and then stop — because region is unconstrained, all regions under that date must be listed.

The percent-encoding guard is a correctness fix. The same decoded value can be spelled two ways in the store (dt=2024-01-01 and dt=2024-01-01 with a %-encoded variant). If the literal’s encoding doesn’t round-trip, prefixing would list only one spelling and silently drop files. So it bails out. “Prefix pruning is an optimization, so stop before this partition rather than listing only one spelling and potentially skipping valid rows.”

The recursive directory walk

For genuinely hierarchical layouts, there’s a parallel directory walker with an explicit concurrency limit:

rust
// datafusion/catalog-listing/src/helpers.rs:142-143
/// The maximum number of concurrent listing requests
const CONCURRENCY_LIMIT: usize = 100;

rust
// datafusion/catalog-listing/src/helpers.rs:156-226 (excerpt)
impl Partition {
    /// List the direct children of this partition updating `self.files` with
    /// any child files, and returning a list of child "directories"
    async fn list(mut self, store: &dyn ObjectStore) -> Result<(Self, Vec<Path>)> {
        trace!("Listing partition {}", self.path);
        let prefix = Some(&self.path).filter(|p| !p.as_ref().is_empty());
        let result = store.list_with_delimiter(prefix).await?;
        self.files = Some(
            result.objects.into_iter()
                .filter(|object_meta| object_meta.size > 0)
                .collect(),
        );
        Ok((self, result.common_prefixes))
    }
}

/// Returns a recursive list of the partitions in `table_path` up to `max_depth`
pub async fn list_partitions(
    store: &dyn ObjectStore,
    table_path: &ListingTableUrl,
    max_depth: usize,
    partition_prefix: Option<Path>,
) -> Result<Vec<Partition>> {
    ...
    let mut pending = vec![];
    let mut futures = FuturesUnordered::new();
    futures.push(partition.list(store));

    while let Some((partition, paths)) = futures.next().await.transpose()? {
        // If pending contains a future it implies prior to this iteration
        // `futures.len == CONCURRENCY_LIMIT`. We can therefore add a single
        // future from `pending` to the working set
        if let Some(next) = pending.pop() {
            futures.push(next)
        }
        let depth = partition.depth;
        out.push(partition);
        for path in paths {
            let child = Partition { path, depth: depth + 1, files: None };
            match depth < max_depth {
                true => match futures.len() < CONCURRENCY_LIMIT {
                    true => futures.push(child.list(store)),
                    false => pending.push(child.list(store)),
                },
                false => out.push(child),
            }
        }
    }
    Ok(out)
}

This is a textbook breadth-first walk with a bounded fan-out: at most 100 in-flight LIST requests, a pending stack for overflow, and FuturesUnordered so that completed requests are reaped immediately rather than in submission order. The list_with_delimiter call is the object store’s “give me the files and the subdirectories of this one directory” operation.

Note the filter(|object_meta| object_meta.size > 0) — zero-byte files are dropped. That is a detail that bites people who write placeholder files into a data directory.

This helper is public API; the default scan path uses the object store’s own recursive LIST instead, but the hierarchical walk is what you’d use for genuinely deep partition trees.

Stage 2: list the prefix and decode partitions

The actual listing goes through ListingTableUrl:

rust
// datafusion/datasource/src/url.rs:247-304 (excerpt)
pub async fn list_prefixed_files<'a>(
    &'a self,
    ctx: &'a dyn Session,
    store: &'a dyn ObjectStore,
    prefix: Option<Path>,
    file_extension: &'a str,
) -> Result<BoxStream<'a, Result<ObjectMeta>>> {
    let exec_options = &ctx.config_options().execution;
    let ignore_subdirectory = exec_options.listing_table_ignore_subdirectory;
    ...
    let list: BoxStream<'a, Result<ObjectMeta>> = if self.is_collection() {
        list_with_cache(ctx, store, self.table_ref.as_ref(), &self.prefix, prefix.as_ref()).await?
    } else {
        match store.head(&full_prefix).await {
            Ok(meta) => futures::stream::once(async { Ok(meta) })
                .map_err(|e| DataFusionError::ObjectStore(Box::new(e)))
                .boxed(),
            // If the head command fails, it is likely that object doesn't exist.
            // Retry as though it were a prefix (aka a collection)
            Err(object_store::Error::NotFound { .. }) => {
                list_with_cache(ctx, store, self.table_ref.as_ref(), &self.prefix, prefix.as_ref()).await?
            }
            Err(e) => return Err(e.into()),
        }
    };

    Ok(list
        .try_filter(move |meta| {
            let path = &meta.location;
            let extension_match = path.as_ref().ends_with(file_extension);
            let glob_match = self.contains(path, ignore_subdirectory);
            futures::future::ready(extension_match && glob_match)
        })
        .boxed())
}

Two things to notice. First, the HEAD-then-LIST dance: if the URL points at a single file, HEAD succeeds and the stream is one object; if it 404s, it’s treated as a collection and listed. Second, the listing cache:

rust
// datafusion/datasource/src/url.rs:390-423 (excerpt)
match ctx.runtime_env().cache_manager.get_list_files_cache() {
    None => Ok(store.list(Some(&full_prefix))
        .map(|res| res.map_err(|e| DataFusionError::ObjectStore(Box::new(e))))
        .boxed()),
    Some(cache) => {
        ...
        let vec = if let Some(cached) = cache.get(&table_scoped_base_path) {
            debug!("Hit list files cache");
            cached.files_matching_prefix(&filter_prefix)
        } else if let Some(cached) = cache.get(&table_scoped_list_path) {
            debug!("Hit list files cache for requested path");
            cached.files_matching_prefix(&None)
        } else {
            let mut vec = store.list(Some(&full_prefix))
                .try_collect::<Vec<ObjectMeta>>()
                .await?;
            vec.shrink_to_fit(); // Right-size before caching
            let cached: CachedFileList = vec.into();
            let result = cached.files_matching_prefix(&None);
            cache.put(&table_scoped_list_path, cached);
            result
        };

The cache is keyed by {table, path}, so a full-table listing can satisfy a prefix-filtered request by filtering the cached list. The shrink_to_fit() before caching is a nice touch: the vector came from a stream and is over-allocated.

Now the partition values. Each listed object’s path is decoded into partition column values:

rust
// datafusion/catalog-listing/src/helpers.rs:466-495 (excerpt)
for (part, expected_partition) in subpath.zip(table_partition_cols) {
    match part.split_once('=') {
        Some((name, val)) if name == expected_partition => {
            // Preserve the original value if percent-decoding produces invalid UTF-8.
            let decoded = percent_decode_str(val)
                .decode_utf8()
                .unwrap_or(Cow::Borrowed(val));
            part_values.push(decoded);
        }
        _ => {
            debug!("Ignoring file: file_path='{file_path}', ...");
            return None;
        }
    }
}

Note the strictness: the path segment’s name= must match the expected partition column, in order. A file that doesn’t fit the declared partition layout is dropped, not guessed at.

Then the partition filters are applied per file, by evaluating the predicate against a one-row batch:

rust
// datafusion/catalog-listing/src/helpers.rs:311-330 (excerpt)
pub fn filter_partitioned_file(
    pf: PartitionedFile,
    filters: &[Expr],
    df_schema: &DFSchema,
) -> Result<Option<PartitionedFile>> {
    if pf.partition_values.is_empty() && !filters.is_empty() {
        return Ok(None);
    } else if filters.is_empty() {
        return Ok(Some(pf));
    }
    let arrays = pf.partition_values.iter()
        .map(|v| v.to_array()).collect::<Result<_, _>>()?;
    let batch = RecordBatch::try_new(Arc::clone(df_schema.inner()), arrays)?;
    let filter = utils::conjunction(filters.iter().cloned()).unwrap_or_else(|| lit(true));
    ...

This is a nice trick: the partition values are materialised as a one-row Arrow batch and the ordinary physical expression evaluator is reused to test the filter. No special-case interpreter for partition predicates.

Stage 3: fetch metadata concurrently

Now the interesting concurrency. Multiple table paths are listed in parallel, then per-file metadata (statistics + ordering) is fetched concurrently:

rust
// datafusion/catalog-listing/src/table.rs:936-986
async fn collect_files_for_scan<'a>(
    &'a self,
    ctx: &'a dyn Session,
    store: &'a Arc<dyn ObjectStore>,
    listing_time_filters: &'a [Expr],
    file_limit: Option<usize>,
) -> datafusion_common::Result<(FileGroup, bool)> {
    // list files (with partitions)
    let file_list = future::try_join_all(self.table_paths.iter().map(|table_path| {
        pruned_partition_list(
            ctx, store.as_ref(), table_path, listing_time_filters,
            &self.options.file_extension, &self.options.table_partition_cols,
        )
    }))
    .await?;
    let meta_fetch_concurrency =
        ctx.config_options().execution.meta_fetch_concurrency.get();
    // Table paths can overlap, for example when one path is a directory and
    // another names a file inside it. A ListingTable uses one object store,
    // so the object path uniquely identifies a file within this scan.
    let mut seen_files = HashSet::new();
    let file_list = stream::iter(file_list)
        .flatten_unordered(meta_fetch_concurrency)
        .try_filter(move |file| {
            future::ready(seen_files.insert(file.object_meta.location.clone()))
        });
    // collect the statistics and ordering if required by the config
    let files = file_list
        .map(|part_file| async {
            let part_file = part_file?;
            let (statistics, ordering) = if ctx.config().collect_statistics() {
                self.do_collect_statistics_and_ordering(ctx, store, &part_file).await?
            } else {
                (Arc::new(Statistics::new_unknown(&self.file_schema)), None)
            };
            Ok(part_file.with_statistics(statistics).with_ordering(ordering))
        })
        .boxed()
        .buffer_unordered(
            ctx.config_options().execution.meta_fetch_concurrency.get(),
        );

    get_files_with_limit(files, file_limit, ctx.config().collect_statistics()).await
}

Several things to unpack:

try_join_all over table_paths. A ListingTable can point at multiple paths; each is listed concurrently.

flatten_unordered + buffer_unordered with meta_fetch_concurrency. This is the key parallelism knob. The default is 32:

rust
// datafusion/common/src/config.rs:1060
pub meta_fetch_concurrency: ConfigNonZeroUsize, default = non_zero_usize_default(32)

flatten_unordered merges the per-path streams while preserving up to N in-flight items; buffer_unordered(N) then bounds the number of concurrent metadata fetches. Both are “unordered” — results come back in completion order, not listing order, which matters because metadata fetch latency is dominated by network round-trips and the natural completion order is skewed.

Deduplication via seen_files. Two table paths can overlap (a directory plus a file inside it), and listing the same file twice would double-count its rows and its statistics. The comment spells out the assumption: “A ListingTable uses one object store, so the object path uniquely identifies a file within this scan.”

Statistics and ordering come from one metadata read. This is an explicit optimization:

rust
// datafusion/catalog-listing/src/table.rs:1135-1186 (excerpt)
async fn do_collect_statistics_and_ordering(
    &self,
    ctx: &dyn Session,
    store: &Arc<dyn ObjectStore>,
    part_file: &PartitionedFile,
) -> datafusion_common::Result<(Arc<Statistics>, Option<LexOrdering>)> {
    let path = TableScopedPath {
        table: part_file.table_reference.clone(),
        path: part_file.object_meta.location.clone(),
    };
    let meta = &part_file.object_meta;

    // Check cache first. The key stays `{table, path}` for cheap lookups;
    // the cached value carries the schema fingerprint to prevent reusing
    // stats computed under a different file schema.
    if let Some(cache) = &self.collected_statistics
        && let Some(cached) = cache.get(&path)
        && cached.is_valid_for(meta, &self.file_schema_fingerprint)
    {
        return Ok((Arc::clone(&cached.statistics), cached.ordering.clone()));
    }

    // Cache miss or invalid: fetch both statistics and ordering in a single metadata read
    let file_meta = self.options.format
        .infer_stats_and_ordering(ctx, store, Arc::clone(&self.file_schema), meta)
        .await?;
    ...

For Parquet, “one metadata read” means one footer read that yields both the column statistics and the sorting_columns metadata. Reading the footer twice would double the I/O for every file in the table.

The schema fingerprint in the cache key is a subtle correctness measure: if the file schema changed (schema evolution!), the cached statistics are invalid and must not be reused.

Stage 3.5: the `LIMIT` shortcut

Here’s the trick I mentioned at the top. get_files_with_limit stops pulling files as soon as the accumulated row count exceeds the limit:

rust
// datafusion/catalog-listing/src/table.rs:1209-1260 (excerpt)
async fn get_files_with_limit(
    files: impl Stream<Item = datafusion_common::Result<PartitionedFile>>,
    limit: Option<usize>,
    collect_stats: bool,
) -> datafusion_common::Result<(FileGroup, bool)> {
    let mut file_group = FileGroup::default();
    // Fusing the stream allows us to call next safely even once it is finished.
    let mut all_files = Box::pin(files.fuse());
    enum ProcessingState {
        ReadingFiles,
        ReachedLimit,
    }
    let mut state = ProcessingState::ReadingFiles;
    let mut num_rows = Precision::Absent;

    while let Some(file_result) = all_files.next().await {
        // Early exit if we've already reached our limit
        if matches!(state, ProcessingState::ReachedLimit) {
            break;
        }
        let file = file_result?;
        // Update file statistics regardless of state
        if collect_stats && let Some(file_stats) = &file.statistics {
            num_rows = if file_group.is_empty() {
                // For the first file, just take its row count
                file_stats.num_rows
            } else {
                // For subsequent files, accumulate the counts
                num_rows.add(&file_stats.num_rows)
            };
        }
        // Always add the file to our group
        file_group.push(file);
        // Check if we've hit the limit (if one was specified)
        if let Some(limit) = limit
            && let Precision::Exact(row_count) = num_rows
            && row_count > limit
        {
            state = ProcessingState::ReachedLimit;
        }
    }
    // If we still have files in the stream, it means the limit kicked
    // in, and the statistic could have been different had we processed the
    // files in a different order.
    let inexact_stats = all_files.next().await.is_some();
    Ok((file_group, inexact_stats))
}

The Precision::Exact guard is essential: you can only stop early if the row counts are exact. An estimated row count of “100 rows” for a file that actually has 10 million would let you stop after one file and return zero rows.

And inexact_stats is the honesty flag: if the stream still had files when we stopped, then the aggregate statistics for the table are wrong (we didn’t see all the files), so they get downgraded:

rust
// datafusion/datasource/src/statistics.rs:408-410
if inexact_stats {
    statistics = statistics.to_inexact()
}

This is exactly the mechanism that makes SELECT * FROM huge_table LIMIT 1 cheap: you list the prefix, you fetch metadata for the first few files, and you stop. The object store’s LIST is still O(prefix) in the worst case, but you avoid the per-file footer reads for the rest of the table.

Stage 4: pack files into `FileGroup`s

Finally the files are distributed across partitions. The FileGroup is a deliberately thin abstraction — a bag of files plus merged statistics:

rust
// datafusion/datasource/src/file_groups.rs:379-389
#[derive(Debug, Clone)]
pub struct FileGroup {
    /// The files in this group
    files: Vec<PartitionedFile>,
    /// Optional statistics for the data across all files in the group.
    ///
    /// These statistics cover the full table schema: file columns plus partition columns.
    statistics: Option<Arc<Statistics>>,
}

The split is a chunked round-robin, with a sort first for determinism:

rust
// datafusion/datasource/src/file_groups.rs:453-484
/// Partition the list of files into `n` groups
pub fn split_files(mut self, n: usize) -> Vec<FileGroup> {
    if self.is_empty() {
        return vec![];
    }
    // ObjectStore::list does not guarantee any consistent order and for some
    // implementations such as LocalFileSystem, it may be inconsistent. Thus
    // Sort files by path to ensure consistent plans when run more than once.
    self.files.sort_by(|a, b| a.path().cmp(b.path()));

    // effectively this is div with rounding up instead of truncating
    let chunk_size = self.len().div_ceil(n);
    let mut chunks = Vec::with_capacity(n);
    ...

The sort is there for plan reproducibility: object stores don’t promise a stable listing order, so without sorting, the same query would produce different file→partition assignments on different runs, which makes plans non-deterministic and makes performance regressions impossible to bisect.

There’s a second, cleverer grouping mode:

rust
// datafusion/datasource/src/file_groups.rs:486-543 (excerpt)
pub fn group_by_partition_values(
    self,
    max_target_partitions: usize,
) -> Vec<FileGroup> {
    ...
    let mut partition_groups: HashMap<Vec<ScalarValue>, Vec<PartitionedFile>> = HashMap::new();
    for file in self.files {
        partition_groups.entry(file.partition_values.clone()).or_default().push(file);
    }
    let num_unique_partitions = partition_groups.len();
    // Sort for deterministic bucket assignment across query executions.
    let mut sorted_partitions: Vec<_> = partition_groups.into_iter().collect();
    ...
    sorted_partitions.sort_by(|a, b| {
        compare_rows(&a.0, &b.0, &sort_options).unwrap_or(Ordering::Equal)
    });
    if num_unique_partitions <= max_target_partitions {
        sorted_partitions.into_iter().map(|(_, files)| FileGroup::new(files)).collect()
    } else {
        // Merge into max_target_partitions buckets using round-robin.
        let mut target_groups = vec![vec![]; max_target_partitions];
        for (idx, (_, files)) in sorted_partitions.into_iter().enumerate() {
            let bucket = idx % max_target_partitions;
            target_groups[bucket].extend(files);
        }
        target_groups.into_iter().map(FileGroup::new).collect()
    }
}

The idea: if all files with the same dt=2024-01-01 land in the same file group, then the scan can declare an output partitioning on dt, and the optimizer can skip repartitioning for an aggregate or join on dt. That’s a real win, and it’s why there’s a threshold:

rust
// datafusion/catalog-listing/src/table.rs:1014-1036
// Threshold: 0 = disabled, N > 0 = enabled when distinct_keys >= N
//
// When enabled, files are grouped by their Hive partition column values, allowing
// FileScanConfig to declare output partitioning. This enables the optimizer to
// skip repartitioning for aggregates and joins on partition columns.
let threshold = ctx.config_options().optimizer.preserve_file_partitions;

let (file_groups, grouped_by_partition) =
    if threshold > 0 && !self.options.table_partition_cols.is_empty() {
        let grouped = file_group.group_by_partition_values(file_group_count);
        if grouped.len() >= threshold {
            (grouped, true)
        } else {
            let all_files: Vec<_> = grouped.into_iter().flat_map(|g| g.into_inner()).collect();
            (FileGroup::new(all_files).split_files(file_group_count), false)
        }
    } else {
        (file_group.split_files(file_group_count), false)
    };

datafusion.optimizer.preserve_file_partitions defaults to 0 (disabled). Set it to 1 to always group by partition values, or to N to only do it when there are at least N distinct partition values.

Stage 5: build the scan

Finally the whole thing becomes a FileScanConfig:

rust
// datafusion/catalog-listing/src/table.rs:769-786
let file_source = self.create_file_source();
let scan_config = FileScanConfigBuilder::new(object_store_url, file_source)
    .with_file_groups(partitioned_file_lists)
    .with_constraints(self.constraints.clone())
    .with_statistics(statistics)
    .with_projection_indices(projection)?
    .with_limit(inflated_limit)
    .with_output_ordering(output_ordering)
    .with_output_partitioning(output_partitioning)
    .with_expr_adapter(self.expr_adapter_factory.clone())
    .build();

// create the execution plan
let plan = self.options.format
    .create_physical_plan(state, scan_config)
    .await?;

FileScanConfig is the hand-off point between the catalog and the physical plan:

rust
// datafusion/datasource/src/file_scan_config/mod.rs:153-181 (excerpt)
pub struct FileScanConfig {
    /// Object store URL, used to get an [`ObjectStore`] instance from
    /// [`RuntimeEnv::object_store`]
    pub object_store_url: ObjectStoreUrl,
    /// List of files to be processed, grouped into partitions
    ///
    /// DataFusion may attempt to read each partition of files
    /// concurrently, however files *within* a partition will be read
    /// sequentially, one after the next.
    ///
    /// Note that when `datafusion.execution.enable_file_stream_work_stealing`
    /// is enabled (the default), files may be reassigned to a different
    /// partition at runtime unless `preserve_order` or
    /// `partitioned_by_file_group` is set, so a file is not guaranteed to be
    /// read by the partition it is grouped under here.
    pub file_groups: Vec<FileGroup>,
    ...

“Files within a partition will be read sequentially, one after the next” — that’s the reason FileGroup exists at all. It’s the unit of sequential work that is distributed in parallel across partitions.

The statistics attached to the config are the merged table-level ones:

rust
// datafusion/datasource/src/statistics.rs:380-413 (excerpt)
pub fn compute_all_files_statistics(
    file_groups: Vec<FileGroup>,
    table_schema: SchemaRef,
    collect_stats: bool,
    inexact_stats: bool,
) -> Result<(Vec<FileGroup>, Statistics)> {
    let file_groups_with_stats = file_groups.into_iter()
        .map(|file_group| compute_file_group_statistics(file_group, Arc::clone(&table_schema), collect_stats))
        .collect::<Result<Vec<_>>>()?;
    // Then summary statistics across all file groups
    let file_groups_statistics = file_groups_with_stats.iter()
        .filter_map(|file_group| file_group.file_statistics(None));
    let mut statistics = Statistics::try_merge_iter_with_ndv_fallback(
        file_groups_statistics, &table_schema, NdvFallback::Max,
    )?;
    if inexact_stats {
        statistics = statistics.to_inexact()
    }
    Ok((file_groups_with_stats, statistics))
}

Two levels: per-file-group statistics (merged from the files in the group), then a table-level merge across groups. NdvFallback::Max is the conservative choice for NDV merging — for files from the same table, the NDV of the union is bounded by the larger NDV, not the sum.

The whole pipeline

text

 ListingTable::scan_with_args
        │
        ├─ split filters: partition-only vs data
        ├─ inflate limit by skip
        │
        ▼ list_files_for_scan
        │
        ├─ evaluate_partition_prefix  →  "dt=2024-01-01"
        ├─ list_prefixed_files (LIST, with cache)
        ├─ parse_partitions_for_path   →  partition values per file
        ├─ filter_partitioned_file     →  drop non-matching partitions
        │
        ▼ collect_files_for_scan
        │
        ├─ try_join_all over table_paths        (parallel listing)
        ├─ flatten_unordered / buffer_unordered   (meta_fetch_concurrency = 32)
        │     └─ infer_stats_and_ordering        (one footer read per file)
        │
        ▼ get_files_with_limit  ── early exit when Exact rows > limit
        │
        ▼ split_files / group_by_partition_values → Vec<FileGroup>
        │
        ▼ compute_all_files_statistics → table Statistics
        │
        ▼ FileScanConfigBuilder → FileScanConfig → DataSourceExec

Trade-offs and failure modes

Listing is O(files). For a table with a million small files, the LIST operation dominates query latency, and no amount of parallelism in the scan helps. This is the “small files problem”; the mitigations are compaction upstream, or the listing cache.

The listing cache is a correctness hazard if the table changes. The cache key is {table, path} plus a schema fingerprint and the object metadata, and is_valid_for checks those. But if a file is added to the table, a cached listing will not see it.

collect_statistics = false kills everything downstream. Look at the branch again:

rust
let (statistics, ordering) = if ctx.config().collect_statistics() {
    self.do_collect_statistics_and_ordering(ctx, store, &part_file).await?
} else {
    (Arc::new(Statistics::new_unknown(&self.file_schema)), None)
};

No statistics means no file-level pruning, no file ordering, no NDV-based join estimates, no CollectLeft decision — and, crucially, no early exit on LIMIT, because the row count would be Absent and the Precision::Exact guard would never fire. So SELECT * FROM t LIMIT 1 would list every file in the table.

Zero-byte files are silently dropped, and so are files whose paths don’t line up with the declared partition layout. If you use zero-byte data files as placeholders, or typo your partition column list, data disappears without an error.

preserve_file_partitions can produce fewer groups than partitions. If you have 3 distinct partition values and target_partitions = 16, you get 3 file groups, the scan has 3 partitions, and the rest of the machine idles.

The LIMIT early exit changes the statistics’ meaning. The table’s reported row count is the sum over the files that were examined, not the whole table. That’s why inexact_stats exists and why the result is downgraded to Precision::Inexact.

meta_fetch_concurrency is a double-edged sword. 32 concurrent footer reads is great for latency and terrible for a store with strict request limits.

Configuration

Option	Default	Effect
`datafusion.execution.collect_statistics`	`true`	Read file footers for statistics. Off ⇒ no pruning, no ordering, no `LIMIT` early exit.
`datafusion.execution.meta_fetch_concurrency`	`32`	Concurrent per-file metadata fetches.
`datafusion.execution.listing_table_ignore_subdirectory`	`true`	Whether the table’s glob may match files in subdirectories.
`datafusion.optimizer.preserve_file_partitions`	`0`	`0` disables grouping by partition values; `1` always groups; `N` groups only when there are ≥ `N` distinct partition values.
`datafusion.execution.enable_file_stream_work_stealing`	`true`	Allows the runtime to reassign files to different partitions than the `FileGroup` they were grouped into.

Bottom line

The listing layer is where DataFusion’s “query engine” part ends and its “data lake client” part begins. Three ideas do most of the work:

Push the filter into the path. A partition equality becomes a LIST prefix, and the object store does the pruning for free. Getting the prefix wrong in the “too narrow” direction loses data, so the code bails out conservatively at every ambiguity.
Amortize the metadata reads. One footer read per file yields both statistics and ordering; the reads happen concurrently; the results are cached by {table, path, schema fingerprint}.
Stop early when you can prove it’s safe. A LIMIT with exact row counts lets the whole pipeline terminate after a handful of files, and the statistics are honestly downgraded to Inexact to reflect that.

The one thing to internalize: the file listing is a full pass over the table’s metadata. Everything downstream — pruning, join selection, limit pushdown — is cheap by comparison. If your query is slow and the table has many files, look at the listing before you look at the plan.

The contract: TableProvider