Storage Formats, Partitioning, and Serialization Questions

Direct answer. A slow Parquet scan when other tables read fine is a process-of-elimination problem: check column projection (is the query actually reading only the columns it needs), predicate pushdown (is the filter reaching the scan node), rowgroup settings and file sizes (is this specific table poorly organized), compression codec and dictionary encoding (is decompression itself the bottleneck), and executor memory (is the cluster spilling to disk or garbage-collecting excessively for this specific table's row/column shape); the fact that OTHER tables read fine points strongly at something specific to THIS table's physical layout rather than a cluster-wide issue.

Structured elaboration, as a diagnostic checklist.

1. Column projection. Check the actual query for a SELECT * or an unnecessarily wide column list; verify (via EXPLAIN) that projection pushdown is actually narrowing which columns are read. Impact if wrong: reading every column when only a few are needed multiplies I/O directly by the ratio of unused-to-used columns.
2. Predicate pushdown. Verify (via EXPLAIN, checking for a pushed-down filter at the scan node) that the query's WHERE clause is reaching the file-scan layer rather than being applied only after a full read; a function-wrapped filter column is a common, specific cause.
3. Rowgroup settings. Check this specific table's actual row-group size (via the file's own metadata) against a healthy target; a table written with an unusually large or small row-group size (compared to other, healthy tables) directly explains a scan-speed difference other tables don't share.
4. Compression codec. Check which codec this table was written with; if it differs from other, faster tables (say, this one uses Gzip while others use Snappy or Zstandard), the extra decompression CPU cost, measurable directly, is a legitimate, isolated explanation.
5. Dictionary encoding settings. Check whether this table's high-selectivity columns are dictionary-encoded appropriately; a HIGH-cardinality column that got dictionary-encoded anyway (or a writer configuration forcing it) can actually hurt rather than help, a real, measurable, and non-obvious failure mode.
6. File sizes. Check this table's actual file-count and average-file-size; a table specifically affected by the small-files problem (a real, measured 10x file-count difference between well- and poorly-organized versions of otherwise identical data) explains a scan-speed regression isolated to this one table, since the per-file overhead scales with file count, not data volume.
7. Executor memory. Check for spill-to-disk or excessive garbage collection specifically during this table's scan (via the Spark UI's stage metrics, or your engine's equivalent); a table with unusually wide rows, deeply nested columns, or a poorly-chosen row-group/page size relative to available executor memory can trigger memory pressure that other, more modestly-shaped tables don't.

How each check's outcome changes the diagnosis. Ruling OUT projection and pushdown first (checking the query plan) narrows the search to the table's own physical organization; if row-group size and file count both look healthy, compression codec and dictionary-encoding choice become the prime suspects; if those also look normal, executor memory pressure specific to this table's row/column width is the remaining likely explanation, at which point the fix shifts from "reorganize the table" to "tune the read-side memory configuration for this specific table's shape."

Trade-offs & pitfalls. A common mistake is jumping straight to "increase executor memory" as a first response to any slow scan, without first ruling out the cheaper, more common explanations (a defeated pushdown, a badly-organized file layout) that memory tuning wouldn't actually fix; work through the checklist roughly in the order above, cheapest and most likely to check first, before reaching for a resource-scaling fix.

Direct answer. A metadata service that avoids expensive object-store LIST operations for a table with millions of partitions and billions of files needs to maintain its own authoritative, queryable index of the table's file layout (partition values, file paths, and ideally per-file statistics), updated transactionally as part of every write, so that "which files satisfy this partition predicate" is answered by querying a small, fast index rather than by listing the object store; this is the same problem Apache Iceberg's manifest-tree metadata layer solves natively, and building a custom version of it (relevant when working with raw files, no table format) means replicating that same design.

Structured elaboration.

• APIs. A minimal useful surface: list_partitions(predicate) returning matching partition identifiers without touching the object store; lookup_files(partition_predicate, column_predicate) returning candidate file paths (optionally with per-file statistics) for a given filter combination; a register_write(file_paths, partition_values) API that writers call to atomically register newly-written files, this is the API surface that keeps the index authoritative rather than best-effort.
• Caching. A hot, in-memory (or fast-key-value-store-backed) cache of the most frequently queried partition ranges (recent dates, in most real workloads) avoids re-querying the service's own backing store for every request; cache invalidation on new writes needs to be precise (invalidate exactly the affected partition's cache entry, not the whole cache) to keep the service both fast and correct under continuous ingestion.
• Consistency model with writers. The service needs to guarantee that once register_write returns success for a batch of files, ANY subsequent query through the service sees those files (a form of read-your-writes / linearizable consistency for the registration operation itself), while tolerating writers that are still IN PROGRESS (a file not yet registered simply isn't visible yet, which is the correct, safe behavior, not a bug); this is the same "commit only after everything's confirmed written" discipline that makes object-store writes safe.
• Integration with Hive metastore or Iceberg. Rather than building this from scratch, integrating as a caching/acceleration LAYER in front of an existing catalog (Hive Metastore, or Iceberg's own catalog) is the more common real-world pattern: the underlying catalog remains the authoritative source of partition/file registration, while this service provides a fast, horizontally-scalable read path in front of it, specifically to handle query-planning load at high concurrency without hitting the underlying catalog's own scaling limits directly.
• Horizontal scaling. Since the service's read path (partition/file lookup) is far more frequent than its write path (new file registration) in most analytical workloads, the natural scaling shape is many read replicas behind a load balancer, backed by a smaller number of consistently-ordered write appliers (or a single write path with the reads served from replicated, eventually-but-boundedly-consistent copies, accepting a small, bounded staleness window in exchange for read scalability, a real trade-off against the strict consistency goal above that should be made deliberately, not by accident).

Worked example. This is architecturally very close to describing Apache Iceberg's own manifest-list-plus-manifest-file metadata tree reimplemented as a standalone service rather than as files colocated with the table: the manifest list is this service's cached, fast-to-query partition/file index; the atomic snapshot swap is this service's register_write-then-visible guarantee; the whole point, avoiding object-store LIST calls at query-planning time, is identical in both designs. A team facing exactly this problem should seriously consider adopting Iceberg (or Delta Lake) directly rather than building a custom equivalent, unless there's a specific reason (a legacy system, a non-Parquet format, an unusual latency requirement) the existing table formats genuinely don't serve.

Trade-offs & pitfalls. A common mistake in a custom-built version of this service is under-designing the write-time consistency guarantee (treating registration as fire-and-forget rather than confirmed-before-visible), which reintroduces exactly the eventual-consistency and partial-write risks that motivated building the service in the first place.

Direct answer. Predicate pushdown eliminates ROWS (or whole row groups of rows) that cannot match a filter condition. Projection pushdown eliminates COLUMNS that a query never references. They solve different halves of the same I/O-reduction problem, filter what you don't need vs. select what you don't need, and a well-optimized columnar read applies both simultaneously.

Structured elaboration.

• Predicate pushdown operates on the WHERE clause. The engine pushes the filter condition down to the storage-reading layer (ideally as close to the disk as possible) instead of reading every row and filtering in a later processing step. In a columnar file, this is implemented via the min/max and null-count statistics recorded per row group: a row group whose statistics prove it cannot contain a matching row is skipped entirely.
• Projection pushdown operates on the SELECT clause (or more precisely, on the actual set of columns referenced anywhere in the query, including in a WHERE or GROUP BY that doesn't appear in the final output). The engine reads only the column chunks for columns actually needed, never touching the bytes belonging to unreferenced columns at all. This only makes sense in a columnar layout, where columns are physically separate on disk; a row-based format cannot skip a column without still reading past it inside each row.

Worked example. Consider a Parquet table events(event_id, event_date, user_id, event_type, payload_json, device_fingerprint) with six columns, and a query:

SELECT user_id, event_type
FROM events
WHERE event_date = DATE '2024-06-15'

• Projection pushdown determines the query only references user_id, event_type (in SELECT), and event_date (in WHERE), three of six columns. The engine never reads the event_id, payload_json, or device_fingerprint column chunks at all, for any row, regardless of whether those rows match the filter, cutting I/O roughly in half before the filter is even considered.
• Predicate pushdown, applied within the three columns actually being read, uses event_date's row-group statistics to skip whole row groups whose date range excludes 2024-06-15 (demonstrated concretely, with real numbers: one row group out of twenty needed to be opened for an equivalent date filter on a comparable dataset).

Combined, the two techniques mean this query reads roughly three columns' worth of data, and within those columns, only the row groups that could possibly contain matching rows, a small fraction of the table's total bytes, without either optimization requiring the query to be rewritten or hinted manually; both are automatic consequences of a columnar layout with embedded statistics. The memory footprint reduction follows directly from the same mechanism, not a separate one: a column chunk that projection pushdown never reads is never decompressed or decoded into a buffer at all, and a row group that predicate pushdown skips never has its values materialized into memory for filtering, so the engine's peak working set during the scan is bounded by the columns and row groups it actually touches rather than by the table's full size.

Trade-offs & pitfalls. Projection pushdown can be silently defeated by a query pattern that looks selective but isn't: SELECT * FROM events WHERE ... forces every column to be read regardless of how selective the filter is, since every column is part of the projection. Predicate pushdown can be silently defeated by wrapping the filtered column in a function or expression (WHERE CAST(event_date AS VARCHAR) = '2024-06-15' or WHERE YEAR(event_date) = 2024), because many engines cannot invert an arbitrary function to map the literal back onto the column's own value range; some modern engines have gotten better at pushing simple, well-known functions through this barrier, but it should never be assumed, write the filter directly against the raw column when pruning matters.

Direct answer. Good partition-key rules of thumb, in order of importance: (1) partition by a column your queries actually filter on, ideally the MOST commonly filtered one; (2) prefer a column with natural ordering or a small number of stable, roughly-even-sized values over a high-cardinality one; (3) size partitions so each one holds a meaningfully large amount of data (avoiding both too-few-huge-partitions and too-many-tiny-partitions); (4) avoid partitioning by a column whose value changes after a row is written (forces a costly move between partitions on update); (5) consider write patterns, not just read patterns, since the partition currently receiving writes can become a hotspot.

Structured elaboration with worked examples.

1. Partition by what you actually filter on. A good partition key for event logs is event_date: if the overwhelming majority of queries filter or aggregate by date (a very common real pattern), partitioning on it directly enables pruning for most traffic. A bad partition key for the same table would be event_id (a unique identifier): almost no query filters by a specific event ID range, so partitioning on it provides no pruning benefit for realistic query patterns while still incurring the operational overhead of having a partition scheme at all.
2. Prefer natural ordering / bounded cardinality over high cardinality. A good partition key for sales data is region (a small, stable set: a handful to a few dozen values), because each partition is a meaningful, roughly-even-sized chunk and the total partition count stays manageable. A bad partition key for the same table would be customer_id (potentially millions of distinct values): this produces either one partition per customer (an enormous small-files problem) or requires an additional hashing/bucketing layer to make it usable at all, at which point it's no longer really "partitioning by customer_id" in the simple sense.
3. Size partitions meaningfully. Too coarse (say, partitioning a modest table by YEAR when it only has a few years of data) means each partition is enormous and pruning barely narrows the scan. Too fine (partitioning the same table by MINUTE) means most partitions are nearly empty, and the small-files and metadata overhead this creates was demonstrated directly and measurably: an unsorted parallel write to a 365-partition (daily) scheme still produced over 2,600 files from a 2,000,000-row table, and a finer partitioning granularity would only make that worse per partition.
4. Avoid a column whose value changes after write. A bad partition key for a sales table would be order_status if orders transition through statuses over time (pending -> shipped -> delivered): every status change would require physically moving the row's data from one partition to another, an expensive and error-prone operation compared to updating a value in place within a stable partition.
5. Consider write patterns. Even a well-chosen read-favorable key like event_date concentrates ALL new writes on whichever partition represents "now," which is the hot-partition pitfall covered in the general partitioning-strategy discussion; for very high-ingestion-rate tables, this sometimes argues for a secondary bucketing dimension specifically to spread write load within the current time partition, a genuine tension with rule 1's "partition by what you filter on" when the highest-value read pattern and the highest-value write pattern point in different directions.

Worked example: choosing among three real candidates. For a large web-events table, comparing event_date, a hash of user_id, and geographic_region as partition-key candidates: event_date wins if the dominant query pattern is "recent activity" (the common case for most analytics/dashboard workloads); a hash of user_id wins specifically if the dominant pattern is per-user lookups across all history with no time-scoping; geographic_region wins if regional-compliance or regional-reporting queries dominate and the region count stays small and stable. The right choice is not abstract, it is whichever candidate's cardinality and access-pattern profile most closely matches the rules above for YOUR actual, measured query traffic, not a generic best practice applied without checking.

Trade-offs & pitfalls. A common mistake is choosing a partition key based on the SCHEMA's most prominent-looking column (a natural key, a customer identifier) rather than the ACTUAL, measured query-filter distribution; the right partition key is discovered by looking at query logs, not by looking at the table's schema in isolation.

Direct answer. Dictionary encoding replaces repeated values with small integer codes pointing into a per-column dictionary, and wins big on low-cardinality columns (a handful to a few thousand distinct values repeated across many rows) but actively hurts on high-cardinality columns, where the dictionary itself grows almost as large as the data. Run-length encoding (RLE) collapses consecutive repeats of the same value into a (value, count) pair, and wins on columns with long runs of identical values, which is common after sorting. Delta encoding stores the difference between consecutive values rather than the values themselves, and wins on monotonic or slowly-changing numeric sequences like timestamps or auto-incrementing IDs. Bit-packing stores small-range integers using only as many bits as their range needs rather than a fixed word size, and wins on any bounded-range integer column, flags, small enums, percentages.

Structured elaboration and worked example (measured, not asserted).

A real, measured demonstration of the dictionary-encoding case specifically:

import duckdb, os
con = duckdb.connect()
con.execute("CREATE TABLE low_card AS SELECT i AS id, "
            "(['US','GB','DE','IN','BR'])[1 + (i % 5)] AS country, i AS ts_val "
            "FROM range(500000) t(i)")
con.execute("CREATE TABLE high_card AS SELECT i AS id, "
            "'uuid-' || md5(i::VARCHAR) AS customer_uuid, i AS ts_val "
            "FROM range(500000) t(i)")
con.execute("COPY low_card TO 'low_card.parquet' (FORMAT PARQUET)")
con.execute("COPY high_card TO 'high_card.parquet' (FORMAT PARQUET)")
print('low-card:', os.path.getsize('low_card.parquet'), 'bytes')
print('high-card:', os.path.getsize('high_card.parquet'), 'bytes')
enc = con.execute("SELECT DISTINCT path_in_schema, encodings "
                   "FROM parquet_metadata('low_card.parquet')").fetchall()
print(enc)

Writing two 500,000-row Parquet files that are identical except for one string column, one with 5 distinct repeated values (country), the other with 500,000 distinct near-unique values (a UUID-like customer_uuid), produced file sizes of 4,014,775 bytes for the low-cardinality table versus 20,197,769 bytes for the high-cardinality table on this run, roughly a 5x difference for the same row count and the same second BIGINT column in both tables. DuckDB's own writer confirmed it selected PLAIN_DICTIONARY encoding for the low-cardinality country column, and fell back to PLAIN for the high-cardinality customer_uuid column (inspected directly via Parquet's embedded row-group metadata). This is the concrete, measured version of the general finding: on a high-cardinality dimension like a customer ID, dictionary encoding is typically the LEAST effective encoding, because the dictionary has to hold nearly as many distinct entries as there are rows, eliminating almost all of the compression benefit while adding a lookup-table's worth of overhead; a general-purpose byte-level compressor (Zstandard, Gzip) applied without dictionary encoding, or delta-encoding if the high-cardinality values happen to be sortable and close together, typically does better on that shape of column.

Practical guidance by type.

• Strings: dictionary-encode when cardinality is low relative to row count; for genuinely high-cardinality strings (customer IDs, UUIDs, free text), skip dictionary encoding and rely on the general compressor, or consider whether a surrogate integer key could replace the string entirely at the modeling layer.
• Integers: delta-encode when the column is sorted or near-sorted (timestamps in an append-only log, auto-incrementing IDs); bit-pack when the range is small and bounded regardless of sort order (a status code with 8 possible values).
• Timestamps: delta encoding is close to a default-correct choice for timestamps in time-ordered data, since consecutive events are usually close together in time.
• Sparse or high-null columns: a wide, mostly-null VARCHAR column with many repeated values (a common real pattern: an optional field populated for only a fraction of rows) benefits from dictionary encoding on its non-null values plus efficient null-bitmap storage; if the sparse column also has extreme cardinality among its non-null values, consider a schema-level fix (a surrogate ID referencing a separate lookup table) rather than relying on file-level encoding alone.

Trade-offs & pitfalls. Most modern columnar writers (Parquet's writer, in particular) choose per-column encodings automatically and adaptively (falling back from dictionary to plain encoding if the dictionary grows past a size threshold), so a common mistake is manually forcing an encoding that the writer's own heuristics would have avoided for good reason; understanding these trade-offs is more valuable for diagnosing WHY a column compressed poorly than for manually overriding the writer in most modern engines.