Layout and pruning
A table's data files are partitioned by time and carry min/max bounds per column. Engines use both to skip files a filter rules out. Once a time bucket has ended, its small files are assembled into larger ones. Tables with granularity none are not assembled.
Time partitioning
| Granularity | Partition fields |
|---|---|
hour, the default | _kafka_partition, _kafka_timestamp_hour |
day | _kafka_partition, _kafka_timestamp_day |
month | _kafka_partition, _kafka_timestamp_month |
none | _kafka_partition |
Buckets are UTC hours, days or months of _kafka_timestamp. Set a table's granularity with mink table add <topic> --granularity day, or change it later with mink granularity <topic> <granularity>. Files already written keep their granularity, and new files use the new one.
Statistics
Every data file records, per column:
| Statistic | Columns |
|---|---|
| Min and max | _kafka_partition, _kafka_offset, _kafka_timestamp, and every typed primitive column, except binary, fixed and columns inside lists and maps. String bounds keep their first 16 characters. |
| Null count | Every leaf column |
| NaN count | Every float and double leaf column |
The Parquet footer also holds min and max per row group. Each table also publishes distinct-value estimates per column for cost-based planners. They do not skip files.
What prunes
| Filter | Skips |
|---|---|
_kafka_timestamp range | Time buckets and files outside the range |
_kafka_partition = p | Files of other partitions |
_kafka_offset range | Files outside the range |
A typed column, such as id = 1234 | Files whose min and max exclude the value. That needs values that grow with offset or time, such as sequential ids or event times. |
Raw _kafka_key or _kafka_value, or JSON decoded from them | Nothing. Binary columns carry no bounds. |
Measured on a table of 3,600 records in 36 files, 3 partitions, 100 consecutive ids per file, before assembly:
| Query | Files read |
|---|---|
WHERE id = 1234 | 1 of 36 |
WHERE id BETWEEN 2000 AND 2400 | 5 of 36 |
WHERE _kafka_partition = 1 AND id < 600 | 2 of 36 |
WHERE _kafka_offset BETWEEN 1000 AND 1050 | 3 of 36 |
WHERE amount > 990, amounts scattered | 12 of 36 |
The console's SQL page shows these numbers with Profile. See Console.
Assembly
A time bucket's files are assembled once the bucket ended assembly_delay ago, 10 minutes by default. For each partition, consecutive files are joined into one file of at most assembled_bytes, 1 GiB by default. Assembly reads no Kafka records: the new file's row groups are the old files' row groups, byte for byte.
After assembly, a filter on a typed column reads the assembled file. The file keeps each original file's min and max as its row group's statistics, for engines that skip row groups. On the table above, once its hour was assembled:
| Query | Files read |
|---|---|
WHERE id = 1234 | 3 of 3 |
WHERE _kafka_partition = 1 | 1 of 3 |
Manifests
Each manifest holds the files of one time bucket, so engines skip whole manifests outside a time range. An assembled bucket has one manifest. A bucket still open has at most 16.