How to Query Hive-Partitioned GeoParquet in DuckDB
On this page
Problem statement
One 446 MB Parquet file is convenient until every query filters on the same column. Then the engine reads the file, prunes what it can from the row-group statistics, and still touches more than it needs.
Partitioning splits the data into a directory tree whose path encodes a column value:
places/
country=AD/data_0.parquet
country=AE/data_0.parquet
country=US/data_0.parquet
…
A query with where country = 'US' then reads one directory and ignores every other one — before opening anything. Measured: partitioning 13,472,195 rows by country produced 271 files totalling 490 MB, 10% more than the single 446 MB file, and 190 of the files were under 1 MB (Example 1).
The cost is that partitioning is a commitment. It optimises one access pattern and can make every other one worse — and on this data it did not even win its own: a filtered count took 1.9 ms on the single file and 6.0 ms across the partitions (Example 2).
Quick answer
Write with partition_by, read with hive_partitioning:
import duckdb
con = duckdb.connect()
# write
con.execute("""
copy (select * from read_parquet('full.parquet'))
to 'places_by_country'
(format parquet, partition_by (country), compression zstd, overwrite_or_ignore)
""")
# read: the partition column comes back as a normal column
print(con.execute("""
select country, count(*)
from read_parquet('places_by_country/**/*.parquet', hive_partitioning = 1)
where country = 'NO'
group by 1
""").fetchall())[('NO', 608564)]
The partition column does not exist inside the files — it is reconstructed from the directory names, which is why filtering on it costs nothing at all.
Step-by-step solution
1. Choose the partition column from the queries, not from the data
A good partition column is one that almost every query filters on, and whose values divide the data into manageable pieces.
Good candidates: country, region, year, month, product line. Bad candidates: anything with thousands of distinct values (one file per value means thousands of tiny files), anything with two values (the pruning saves half at best), and anything queries rarely filter on.
2. Aim for the right file size
Partitioning by a high-cardinality column produces the small-files problem: thousands of files, each a few kilobytes, where per-file overhead dominates.
Target roughly 100 MB to 1 GB per file. Measured on the 13.5-million-row dataset partitioned by country, the largest partition was the United States and the smallest were tiny territories — a spread of several orders of magnitude, which is typical and usually acceptable when the small ones are rarely queried.
3. Write the partitioned dataset
copy (select * from source)
to 'output_directory'
(format parquet, partition_by (country, year), compression zstd,
overwrite_or_ignore);Two partition columns nest: country=US/year=2025/data_0.parquet. Each additional level multiplies the file count, so two levels is usually the practical limit.
Measured: partitioning 13.5 million rows by country took 5.5 s and produced 271 files — more than the 253 countries, because DuckDB splits a large partition across several files (the United States got three).
4. Read with hive_partitioning = 1
select * from read_parquet('output/**/*.parquet', hive_partitioning = 1);Without the flag, the partition column is missing from the result — the values live only in the paths. With it, DuckDB parses the directory names into columns and uses them for pruning.
5. Sort within each partition too
Partitioning prunes at the file level; row-group statistics prune within a file. Both are worth having:
copy (select * from source order by country, lat)
to 'output' (format parquet, partition_by (country), compression zstd);The measured effect of within-file sorting is large: on a single file, a filtered query read 0.26 MB from a sorted file and 10.25 MB from a shuffled one — 39× the bytes for the same answer (measured here).
6. Know what partitioning costs
- Queries that do not filter on the partition column now read many files instead of one, with per-file overhead on each.
- Listing the directory is itself work, and on object storage with tens of thousands of prefixes it can dominate a small query.
- Rewriting is expensive. Changing the partition column means rewriting the whole dataset.
- The partition column disappears from the file contents, so a file read without
hive_partitioningis missing a column.
Code examples
The data used on this page
The points are GeoNames' full allCountries dump — every feature, about 13.5 million, rebuilt daily at download.geonames.org, so counts drift a little between runs — kept both as the 1.79 GB text file and as places.parquet (also linked as geonames.parquet) with name, latitude, longitude, feature class, country and population. The polygons are Natural Earth's 1:10m admin-1 states and provinces, 4,596 of them, as provinces.shp.
import os
import zipfile
import duckdb
import requests
def fetch(url, path):
if not os.path.exists(path):
with requests.get(url, stream=True, timeout=600) as reply:
reply.raise_for_status()
with open(path, "wb") as out:
for chunk in reply.iter_content(1 << 20):
out.write(chunk)
fetch("https://download.geonames.org/export/dump/allCountries.zip", "allCountries.zip") # about 420 MB
if not os.path.exists("allCountries.txt"):
zipfile.ZipFile("allCountries.zip").extract("allCountries.txt")
if not os.path.exists("places.parquet"):
duckdb.sql("""
copy (select column01 as name, column04 as lat, column05 as lon, column06 as feature_class,
column08 as country, column14 as population
from read_csv('allCountries.txt', delim = '\\t', header = false, quote = '',
all_varchar = false, sample_size = -1))
to 'places.parquet' (format parquet)""")
if not os.path.exists("geonames.parquet"):
os.link("places.parquet", "geonames.parquet")
fetch("https://naciscdn.org/naturalearth/10m/cultural/ne_10m_admin_1_states_provinces.zip", "provinces.zip")
if not os.path.exists("provinces.shp"):
with zipfile.ZipFile("provinces.zip") as z:
for member in z.namelist():
if member.startswith("ne_10m_admin_1_states_provinces."):
open("provinces" + os.path.splitext(member)[1], "wb").write(z.read(member))
n = duckdb.sql("select count(*) from read_parquet('places.parquet')").fetchone()[0]
print(f"{n:,} GeoNames points; places.parquet {os.path.getsize('places.parquet') / 1e6:.0f} MB, "
f"allCountries.txt {os.path.getsize('allCountries.txt') / 1e9:.2f} GB")13,472,195 GeoNames points; places.parquet 306 MB, allCountries.txt 1.79 GB
The single-file baseline is all 19 GeoNames columns written with zstd, full.parquet, in the dump's own order.
COLUMNS = ("geonameid BIGINT, name VARCHAR, asciiname VARCHAR, alternatenames VARCHAR, lat DOUBLE, lon DOUBLE, "
"feature_class VARCHAR, feature_code VARCHAR, country VARCHAR, cc2 VARCHAR, admin1 VARCHAR, "
"admin2 VARCHAR, admin3 VARCHAR, admin4 VARCHAR, population BIGINT, elevation VARCHAR, dem INTEGER, "
"timezone VARCHAR, modified DATE")
TSV = ("read_csv('allCountries.txt', delim = '\\t', header = false, quote = '', columns = {"
+ ", ".join(f"'{c.split()[0]}': '{c.split()[1]}'" for c in COLUMNS.split(", ")) + "})")
if not os.path.exists("full.parquet"):
duckdb.sql(f"copy (select * from {TSV}) to 'full.parquet' (format parquet, compression zstd)")
print(f"full.parquet {os.path.getsize('full.parquet') / 1e6:.0f} MB")full.parquet 446 MB
Example 1 — writing a partitioned dataset with a report
import duckdb
import os
import time
def write_partitioned(con, source, target, partition_columns, sort_within=None,
compression="zstd"):
"""Partition, then report the file-size distribution — the thing that
decides whether the layout will work."""
order = f"order by {', '.join(list(partition_columns) + list(sort_within or []))}"
started = time.perf_counter()
con.execute(f"""
copy (select * from {source} {order})
to '{target}'
(format parquet, partition_by ({', '.join(partition_columns)}),
compression {compression}, overwrite_or_ignore)
""")
elapsed = time.perf_counter() - started
sizes = []
for root, _, files in os.walk(target):
for name in files:
sizes.append(os.path.getsize(os.path.join(root, name)))
sizes.sort()
total = sum(sizes)
print(f"wrote {len(sizes):,} files, {total / 1e6:,.0f} MB, in {elapsed:.1f}s")
print(f" smallest {sizes[0] / 1e6:8.3f} MB")
print(f" median {sizes[len(sizes) // 2] / 1e6:8.3f} MB")
print(f" largest {sizes[-1] / 1e6:8.3f} MB")
tiny = sum(1 for s in sizes if s < 1e6)
if tiny > len(sizes) * 0.5:
print(f" ! {tiny} files under 1 MB — the partition column may be too "
f"high-cardinality")
import shutil
shutil.rmtree("sorted_by_country", ignore_errors=True)
write_partitioned(con, "read_parquet('full.parquet')", "sorted_by_country", ["country"], sort_within=["lat"])wrote 271 files, 490 MB, in 5.5s
smallest 0.004 MB
median 0.389 MB
largest 39.950 MB
! 190 files under 1 MB — the partition column may be too high-cardinality
Example 2 — measuring how much a partition actually prunes
def pruning_benefit(con, partitioned_glob, single_file, filter_sql):
"""Same query, two layouts. The difference is what partitioning bought."""
import time
def timed(sql):
best = None
for _ in range(3):
started = time.perf_counter()
con.execute(sql).fetchall()
elapsed = time.perf_counter() - started
best = elapsed if best is None else min(best, elapsed)
return best
single = timed(f"select count(*) from read_parquet('{single_file}') "
f"where {filter_sql}")
partitioned = timed(f"select count(*) from read_parquet('{partitioned_glob}', "
f"hive_partitioning = 1) where {filter_sql}")
print(f"single file {single * 1000:8.1f} ms")
print(f"partitioned {partitioned * 1000:8.1f} ms "
f"({single / max(partitioned, 1e-9):.1f}×)")
print(" a small difference means the single file's row-group statistics "
"were already doing the work")
pruning_benefit(con, "sorted_by_country/**/*.parquet", "full.parquet", "country = 'NO'")single file 1.9 ms
partitioned 6.0 ms (0.3×)
a small difference means the single file's row-group statistics were already doing the work
That last line is the important one. Measured here, the single file won: 1.9 ms against 6.0 ms, because the GeoNames dump is already grouped by country — only 6 of the file's 109 row groups have a country range that includes NO — so its row-group statistics prune almost as precisely as the directories, and listing and opening a partition costs more than the groups it skips. Partitioning pays when the single file is not clustered on the filter column, when the data is remote, or when whole partitions are rewritten independently.
Example 3 — listing what the partitions contain, without reading them
def partition_summary(con, glob, partition_column):
"""Row counts per partition, from metadata rather than from the data."""
frame = con.execute(f"""
with files as (select file_name, num_rows from parquet_file_metadata('{glob}')),
sizes as (select file_name, sum(total_compressed_size) as bytes
from parquet_metadata('{glob}') group by 1)
select regexp_extract(file_name, '{partition_column}=([^/]+)', 1) as {partition_column},
count(*) as files,
sum(num_rows) as rows,
round(sum(bytes) / 1e6, 1) as mb
from files join sizes using (file_name)
group by 1 order by rows desc
""").df()
print(frame.head(10).to_string(index=False))
return frame
summary = partition_summary(con, "sorted_by_country/**/*.parquet", "country")country files rows mb
US 3 2241404.0 78.3
CN 1 961888.0 37.3
IN 2 660026.0 17.7
NO 1 608564.0 19.5
FI 1 552800.0 17.4
MX 2 521520.0 18.2
ID 1 452093.0 13.9
RU 1 412812.0 18.1
CA 1 317026.0 8.5
TH 1 264777.0 12.0
Reading the metadata rather than the rows makes this instant even on a large dataset, and it is the fastest way to spot a partition that has grown out of proportion. Two details make it correct: the partition value is not in the metadata — it lives only in the path, so it is parsed from the file name — and parquet_metadata has one row per column chunk, so sizes are summed per file before joining, or every row count would be multiplied by the number of columns.
Explanation
Why partition pruning is free
The partition column's values are in the directory names. DuckDB parses country=US from the path and evaluates where country = 'US' against it before opening the file.
That is qualitatively different from row-group pruning, which requires reading the footer of each file. On a dataset of thousands of files, skipping most of them without any I/O is the whole point.
Why the small-files problem is real
Every Parquet file has a footer, a schema and per-column metadata. At 100 MB per file that overhead is negligible; at 50 kB per file it can exceed the data.
Worse, each file is a separate open — and on object storage, a separate HTTP request with its latency. A query over ten thousand tiny files spends its time on round trips.
The guidance to target 100 MB to 1 GB per file comes from that arithmetic, not from any property of Parquet itself.
Why sorting inside partitions still matters
Partitioning answers "which files do I need?" Sorting answers "which parts of this file do I need?" They compose, and neither replaces the other.
The measured single-file numbers show how much the second is worth: a filter on a sorted file read 0.26 MB in 4 requests; on a shuffled file with identical rows, 10.25 MB in 222 requests. If your partitions are large, sorting within them recovers most of that.
Why partitioning is a commitment
A partitioned layout is optimised for one filter. Queries that do not use it pay the per-file cost of every partition, and changing the scheme means rewriting the dataset.
That is why the first step is to look at the queries. If half the queries filter by country and half by year, partitioning by country makes half of them faster and half of them slower — and a single sorted file may serve both better than either partitioning would.
Edge cases or notes
hive_partitioning = 1is required or the partition column is missing from the result.- The partition column is not stored inside the files — it exists only in the paths.
overwrite_or_ignoreis needed to write into a directory that already exists.- Nesting multiplies files. Two levels is usually the practical maximum.
- Partition values are strings in the path — a numeric column comes back needing a cast.
- Directory listing costs time on object storage, especially with many prefixes.
parquet_file_metadata()andparquet_metadata()read only footers, which makes dataset inspection instant.- A single well-sorted file is often enough. Measure before committing to a layout.
Internal links
- How to query remote GeoParquet over HTTP with DuckDB — where pruning matters most
- How to write partitioned GeoParquet — the GeoPandas route
- Columnar or row storage: why DuckDB is fast on wide tables — row-group statistics
- How DuckDB uses (and does not use) a spatial index — sorting as the file-backed index
- GeoParquet and columnar storage explained — the format
- How to choose chunk and tile sizes — the same sizing argument for rasters
- Cloud-native geospatial explained — why layout is a performance decision
- How to read shapefiles, GeoJSON and GeoParquet in DuckDB — the readers
FAQ
What is Hive partitioning?
A directory layout where the path encodes a column value — country=US/data_0.parquet. Queries filtering on that column skip whole directories without opening any file.
How do I read a partitioned dataset in DuckDB?
read_parquet('dir/**/*.parquet', hive_partitioning = 1). Without the flag the partition column is missing, because its values live only in the paths.
What should I partition by?
A column almost every query filters on, with enough distinct values to divide the data and few enough to keep files large. Country and year are typical; an id column is not.
How large should each partition file be?
Roughly 100 MB to 1 GB. Thousands of small files spend their time on per-file overhead and, on object storage, on round trips.
Do I still need to sort within partitions?
Yes. Partitioning prunes files; sorting prunes row groups inside them. On a single file the measured difference was 0.26 MB against 10.25 MB for the same filter.
Is partitioning always worth it?
No. It optimises one filter and makes other queries read more files. A single well-sorted Parquet file is frequently enough — measure both before committing.