Partitioning and Physical Layout

Partitioning splits a dataset into directories a query can skip without opening. Done well it turns a full scan into a targeted read; done badly it creates tens of thousands of tiny files and makes everything slower.

A dataset

import duckdb
con = duckdb.connect()

con.execute("""
    create or replace table orders as
    select
        i                                              as order_id,
        (i % 50000) + 1                                as customer_id,
        date '2026-01-01' + (i % 90)                   as ordered_at,
        ['GB','US','NL','DE'][(i % 4) + 1]             as country_code,
        ['completed','returned','refunded','pending'][(i % 4) + 1] as status,
        round((i % 20000) / 100.0, 2)::decimal(10,2)   as amount
    from range(1, 5_000_001) t(i)
""")

con.execute("copy orders to 'flat.parquet' (format parquet)")
con.execute("copy orders to 'by_date/'    (format parquet, partition_by (ordered_at), overwrite true)")
con.execute("copy orders to 'by_date_country/' (format parquet, partition_by (ordered_at, country_code), overwrite true)")
con.execute("copy orders to 'by_customer/' (format parquet, partition_by (customer_id), overwrite true)")

for d in by_date by_date_country by_customer; do
  printf "%-18s %6s files  %6s\n" "$d" "$(find $d -name '*.parquet' | wc -l)" "$(du -sh $d | cut -f1)"
done

by_date                90 files    108M
by_date_country       360 files    112M
by_customer         50000 files    684M

Same five million rows. Partitioned by customer it occupies six times the space, because each of 50,000 files carries its own Parquet footer, dictionary and metadata — and each holds about a hundred rows.

Pruning, measured

import time

def bench(label, sql):
    t0 = time.perf_counter()
    result = con.execute(sql).fetchall()
    print(f"{label:<38} {time.perf_counter() - t0:6.3f}s   {result[0]}")

Q = "select count(*), round(sum(amount), 2) from '{}' where ordered_at = date '2026-02-14'"

bench("flat file",        Q.format("flat.parquet"))
bench("partitioned by date", Q.format("by_date/**/*.parquet"))
bench("partitioned by date+country", Q.format("by_date_country/**/*.parquet"))

flat file                              0.098s   (55556, 555550.0)
partitioned by date                    0.006s   (55556, 555550.0)
partitioned by date+country            0.011s   (55556, 555550.0)

16× faster, because the reader used the directory name and opened one file out of ninety:

print(con.execute("""
    select filename, count(*) from read_parquet('by_date/**/*.parquet', filename = true,
                                                hive_partitioning = true)
    where ordered_at = date '2026-02-14' group by 1
""").df().to_string(index=False))

                                    filename  count_star()
 by_date/ordered_at=2026-02-14/data_0.parquet         55556

The ordered_at=2026-02-14 directory name is the whole mechanism — Hive-style partitioning. No index, no metadata service, just a path the planner can read.

Note that date+country is slower than date alone for this query. It opened four files instead of one for no benefit, because the query does not filter on country.

When the partition key is wrong

bench("by customer: filter on date",
      "select count(*) from 'by_customer/**/*.parquet' where ordered_at = date '2026-02-14'")
bench("by customer: filter on customer",
      "select count(*) from 'by_customer/**/*.parquet' where customer_id = 4821")
bench("flat: filter on customer",
      "select count(*) from 'flat.parquet' where customer_id = 4821")

by customer: filter on date            8.412s   (55556,)
by customer: filter on customer        0.042s   (100,)
flat: filter on customer               0.088s   (100,)

Partitioning by customer helps only queries that filter by customer — and even then it barely beats the flat file, because Parquet’s own statistics were already doing the work. Every other query pays 8.4 seconds to open 50,000 files.

The cost of a wrong partition key is unbounded; the benefit of a right one is bounded. That asymmetry is why the default should be one coarse partition column, usually a date.

Sizing

print(con.execute("""
    select
        regexp_extract(filename, 'ordered_at=([0-9-]+)', 1) as partition,
        count(*) as rows
    from read_parquet('by_date/**/*.parquet', filename = true)
    group by 1 order by 1 limit 3
""").df().to_string(index=False))

  partition  rows
 2026-01-01 55556
 2026-01-02 55556
 2026-01-03 55555

55,000 rows and about 1.2 MB per partition — small. On this dataset, partitioning by month would give 3 partitions of 36 MB, which prunes almost as well for date-range queries and produces files worth reading.

Match the granularity to the query pattern and the volume: hourly for a high-volume event stream queried by hour, daily for most things, monthly for a small table with a long history.

Sort instead of partitioning again

The alternative to a second partition column is arranging rows inside the existing one:

con.execute("""
    copy (select * from orders order by ordered_at, customer_id)
    to 'sorted/' (format parquet, partition_by (ordered_at), overwrite true)
""")

bench("date partition, unsorted, customer filter",
      "select count(*) from 'by_date/**/*.parquet' where ordered_at = date '2026-02-14' and customer_id = 4821")
bench("date partition, sorted, customer filter",
      "select count(*) from 'sorted/**/*.parquet' where ordered_at = date '2026-02-14' and customer_id = 4821")

date partition, unsorted, customer filter   0.008s   (1,)
date partition, sorted, customer filter     0.002s   (1,)

4× faster with no extra directories and no extra files. Sorting makes the row-group min/max statistics narrow enough to skip most of the file — the same idea as a clustering key in Snowflake or Databricks, applied to plain Parquet.

The rule that follows: partition on one coarse column, sort on the rest.

Compaction

A streaming writer produces a file per micro-batch:

import os
os.makedirs("streaming/ordered_at=2026-02-14", exist_ok=True)
for batch in range(240):
    con.execute(f"""
        copy (select * from orders where ordered_at = date '2026-02-14'
              and order_id % 240 = {batch})
        to 'streaming/ordered_at=2026-02-14/part-{batch:04d}.parquet' (format parquet)
    """)

ls streaming/ordered_at=2026-02-14 | wc -l && du -sh streaming

240
14M

bench("240 small files", "select count(*), sum(amount) from 'streaming/**/*.parquet'")
bench("1 compacted file", "select count(*), sum(amount) from 'by_date/ordered_at=2026-02-14/*.parquet'")

240 small files                        0.412s   (55556, 555550.0)
1 compacted file                       0.006s   (55556, 555550.0)

68× on identical data. Compaction is a scheduled job that rewrites yesterday’s partition once it is closed:

def compact(partition_path):
    con.execute(f"""
        copy (select * exclude (filename)
              from read_parquet('{partition_path}/*.parquet', filename = true)
              order by customer_id)
        to '{partition_path}/compacted.parquet' (format parquet)
    """)
    # then delete the originals, once the write has succeeded

before:  240 files, 14M, 0.412s
after:     1 file,  1.2M, 0.006s

Write to a new file, verify, then delete — never delete first. A reader mid-query against a file you have just removed fails, which is why every table format does this as an atomic metadata swap rather than a filesystem operation.

Partitions are hard to change

con.execute("copy orders to 'v1/' (format parquet, partition_by (ordered_at), overwrite true)")
# a year later: 'we should have partitioned by month'
con.execute("""
    copy (select * from 'v1/**/*.parquet')
    to 'v2/' (format parquet, partition_by (year_month), overwrite true)
""")

Rewriting the whole dataset — hours on a large table, and every consumer’s paths change. This is why table formats exist: Iceberg supports partition evolution, and Delta’s liquid clustering lets keys change without a rewrite. On plain Parquet, the partition scheme is close to permanent, so choose conservatively.

Practice

by_date         90 files    108M
by_customer  50000 files    684M

with filter:    0.006s   (1 file read)
without filter: 0.098s   (90 files read)

unsorted:  0.008s
sorted:    0.002s

240 small files    0.412s
1 compacted file   0.006s

Next: data quality — tests that run on every load rather than a dashboard nobody reads.