Your Parquet Column Indexes Are Being Ignored on EMR and Glue

I was benchmarking data-clustering layouts on Delta tables. Locally, everything was skipping beautifully, box queries touched about 10% of rows. Then I ran the identical benchmark on EMR 7.13 and those queries scanned 93% of the table. Same code, same data shape, same queries.

Either my layouts were broken on EMR, or something in the read path was. This is the investigation that followed. It ends with a four-environment A/B test proving that the AWS Spark runtime’s vectorized parquet reader, on both EMR 7.13 and Glue 5.0, performs row-group pruning but never applies parquet column indexes. Stock Apache Spark 3.5.6, the exact version EMR ships, applies them fully against the same files.

A Quick Refresher: The Three Tiers of Parquet Pruning

Column indexes are the finest tier, per-page min/max stored in the footer. On a well-clustered table this is where most of the win lives; pages are tight, so a 5% range query really does read ~5% of the data. My numbers said this tier was missing on EMR, but benchmark numbers come with a hundred confounders, so I needed to isolate it.

Ruling Out the Suspects

Is the writer broken? I copied the EMR-written parquet files to my laptop and read them with stock Spark. Page skipping worked (8.6% of rows scanned), so the files carry valid column indexes. Writer exonerated.

Is it a configuration problem? One cheap cluster, four variants of the same query:

No documented conf restores the behavior, but the last row is the evidence. Disabling the vectorized reader falls back to the slow row reader, which always evaluates column indexes, and full page skipping returns. At 9 to 14x the wall-clock, that is ground truth, not a workaround. (This sweep ran on my original table with file-sized row groups, which is why no-skipping degrades to 0.939; the probe below uses a rebuilt table with more files. Compare within each table, not across.)

Is it the environment? One probe script, run unmodified in three AWS environments against byte-identical files in the same bucket. It runs ten staggered range queries over the sort column and reads the scan node’s actual output-row count from the executed plan. A counter, not a timing, so hardware differences cannot move it.

The Test Environments

Every number in this post comes from one of these setups. Bookmark this table; the prose refers back to it instead of re-describing hardware.

The Verdict

Stock Spark’s vectorized reader hits the ground truth exactly. EMR and Glue read 3.3x more rows, byte-identical to each other, which points at a shared AWS runtime lineage. And 0.287 is exactly what row-group pruning alone predicts for this file geometry. As a control, 80 range queries over unsorted columns read ~1.000 in every environment under both readers; the probe is consistent.

The Fork You’re Forced Into

Wall-clock for the same probe queries, both reader modes (warm medians). Hardware differs across rows (see the environments table), so read each row’s ratio, not the columns:

On EMR and Glue you pick one, fast decode that reads 3.3x more rows than it needs, or full skipping at 11 to 14x the wall-clock. Note the perverse detail in the EMR row, where the row reader scans 3.3x fewer rows yet takes 14x longer. Decode efficiency dominates I/O at this scale, which is why “just disable vectorization” is not a workaround anyone would ship. On stock Spark the fork doesn’t exist.

The Fast Native Readers Drop It Too

The obvious alternative is a native accelerated reader. Apache DataFusion Comet 0.16.0 read 1.000 of the table on the MWE where the row reader read 0.051, no page skipping in either of its scan implementations. The twist came on the real S3 table. Comet full-scanned it in 2.2s, faster than stock Spark took to read just 8.6% of it (4.9s, same node). Its I/O path and Rust decode are that much stronger than JVM s3a.

So page-index support is not just missing from AWS’s proprietary reader; the fastest open-source reader drops it too. (Gluten/Velox’s reader did pass the page-skipping test, the only fast native reader I found that does, but its native S3 client never worked end to end, so it could not be timed on real data.) Of everything tested, stock Spark’s vectorized reader is the only one that both decodes in batches and honors column indexes. The boring default is quietly the best parquet reader for layout-sensitive workloads, on a dimension nobody benchmarks.

The machinery Comet needs already exists in its own foundation (arrow-rs implements page-index pruning; DataFusion exposes it as enable_page_index). A reader with Comet’s I/O and working page skipping would strictly dominate every number above. That is an upstream contribution worth making.

Why You Should Care

If you invest in layout, Z-ordering, Hilbert clustering, sorted tables, Delta OPTIMIZE ZORDER, the page-level tier is the biggest single term in your payoff, and on the AWS runtime it silently evaporates. Improvements that measure 10 to 13x under stock Spark shrink to ~3x. Your layout jobs still cost the same to run; they pay out a third of the pruning. Clustering still helps on EMR (2.2x on box queries in my benchmark, from row-group pruning and decode locality), but the largest term goes unrealized, and the waste grows with scale, reaching 22x the necessary I/O at 2.75 billion rows.

If you’re staying on EMR/Glue, two mitigations recover part of it. Shrink row groups (parquet.block.size of 16 to 32 MB instead of 128 MB) to move skip granularity up a tier the AWS reader does honor. And shrink files, since file-level skipping via table-format stats works everywhere. Or do what I did and run stock Spark on EC2, where the reader just works.

Verify It Yourself in Five Minutes

Claims about vendor runtimes deserve receipts, so the whole investigation distills to a self-contained MWE of about 120 lines. It generates its own table of 20 million rows sorted by ts, with parquet.block.size pinned above the file size so each file is exactly one row group. That pin makes the verdict unarguable. Row-group pruning can only reach file granularity (~0.25 for a mid-table window) while page skipping reaches ~0.05, so the tiers cannot be confused.

The measurement reads the scan’s numOutputRows metric off the executed plan, pure py4j, portable to environments you don’t control:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def scan_output_rows(df):
    """Sum 'numOutputRows' over every parquet scan node in df's
    executed plan, descending AQE wrappers."""
    def walk(node):
        name = node.getClass().getSimpleName()
        if name == "AdaptiveSparkPlanExec":
            return walk(node.executedPlan())
        total = 0
        if name in ("FileSourceScanExec", "BatchScanExec"):
            opt = node.metrics().get("numOutputRows")
            if not opt.isEmpty():
                total += opt.get().value()
        if name.endswith("QueryStageExec"):
            total += walk(node.plan())
        it = node.children().iterator()
        while it.hasNext():
            total += walk(it.next())
        return total
    return walk(df._jdf.queryExecution().executedPlan())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def probe(vectorized):
    spark.conf.set("spark.sql.parquet.enableVectorizedReader",
                   str(vectorized).lower())
    df = spark.read.parquet(DATA).where(
        (F.col("ts") >= lo) & (F.col("ts") <= hi))
    q = df.agg(F.sum("val"))
    q.collect()
    return scan_output_rows(q)

vec_frac = probe(vectorized=True) / N
row_frac = probe(vectorized=False) / N
if vec_frac <= row_frac * 1.5:
    print("VERDICT: vectorized reader APPLIES column indexes")
else:
    print(f"VERDICT: vectorized reader IGNORES column indexes — "
          f"{vec_frac / row_frac:.1f}x more rows than the row reader")

Run it as a bare spark-submit, an EMR step, or a Glue job. Stock Spark prints both fractions at ~0.05 and APPLIES. EMR 7.13 and Glue 5.0 show the vectorized reader scanning ~5x more than the row reader on the same files.

Epilogue: Spark 4.2 Ends the Fork With AWS’s Own Library

After publishing the numbers above I tested the newest stack. Same node, same files, same query, page skipping verified throughout:

4.3x faster than the 3.5 line, beating Comet’s blind full-scan outright, and in the ballpark of EMR’s 0.88s from a different hardware class. The gap that motivated this whole investigation has effectively closed, and on the clustered table it is better still, box queries in 0.70s where 3.5.5 took 6.6s.

The root cause is the best punchline of the piece. Hadoop 3.5.0 made AWS’s own open-source Analytics Accelerator Library the default S3A input stream (HADOOP-19559), and Spark 4.2.0 bundles it. A page-index scan issues a dependent chain of small GETs per file (footer, then column index, then the sparse chunks), unparallelizable because later byte ranges depend on earlier reads. AAL deletes the chain instead of parallelizing it, one GET caches the file tail on open and predictive prefetch covers the columns. AWS advertises ~1.1 to 1.3x on suite-level TPC-DS; my latency-dominated selective scan is the degenerate best case at 4.3x. (Spark 4.0.4’s regression is the unlucky middle rung, the SDK v2 migration’s per-GET overhead without the accelerator that pays it back.)

At Scale, Matched Hardware

The two fleets from the environments table, identical core for core, same tables. The prose stays short because the tables below carry the numbers. At 2.75 billion rows the box queries were a photo finish, 0.81s on EC2 against 0.84s on EMR while EMR scanned 17x the rows. At 27.5 billion rows the finish was not close, 0.87s against 2.51s. Ground truth held exactly at both scales; EMR’s own row reader agrees with stock Spark’s vectorized reader on rows scanned to four decimal places, on every query family, the fourth independent platform to land on the same figure. Pruning compounds with data volume while decode bandwidth stays constant, so somewhere around 100 GB the wasted I/O stops being hideable.

Layout is the other half of the story, since the reader can only skip what the layout concentrates. Four methods clustered the same three columns of the fact table at 27.5 billion rows, two space-filling curves (Hilbert and Morton interleave every clustered column into one sort order) and two trees (kd splits on the widest dimension, qd trains its splits on a sample workload). Cells are frac (fraction of rows the scan materialized, the pruning truth) and warm seconds:

The optimize column is the one-time price of admission, about an hour and seven dollars per terabyte on this fleet, scaling linearly from the 100 GB runs, with storage landing 19 to 35 percent above baseline. The trees are respectable but read 3x to 20x more rows than Hilbert everywhere. The mechanism is depth; a tree gets roughly four cuts per partition to divide among three dimensions, while a curve interleaves all three into every file. Curves won every column at every scale I ran.

Formats and platforms, same dataset (the Hilbert row above), warm medians. The parquet and Iceberg rows are byte-identical files; Iceberg registered the parquet directory via add_files, so between those rows the only variable is the reader:

* No Iceberg runtime compatible with Spark 4.2 exists yet, so the Iceberg leg runs Spark 4.1.3 with the Analytics Accelerator switched on explicitly. As close to fair as the ecosystem allows.

Three things jump out. The Delta row on EC2 is the only one under a percent on boxes; the clustering rewrite that built it produced the finest page structure of any writer here, and stock Spark’s reader is the only one that exploits it. The Iceberg rows and the parquet-EMR row match to four decimal places, three page-blind stacks fed the very same bytes converging on one scan at row-group granularity. And EMR’s bold cells are real; on moderate-selectivity single-column ranges its decode still buys the stopwatch, which is why the inflection point is workload-dependent. (ORC and Avro were measured at 100 GB and cut from this round on their own evidence. ORC pruned tightest, 0.0044 on boxes, but decoded ~2x slower than parquet. Avro clocked ~38.4s on every family, a flat full-decode constant.)

Page-blindness, in other words, is a pattern rather than an EMR quirk. Comet, EMR, Glue, Iceberg. Four readers in this investigation share the blind spot, and upstream Spark’s parquet source remains the only fast reader I’ve measured that honors the index. So the ending writes itself. AWS published their S3 I/O secret sauce as open source, the Hadoop community made it the default, and stock Spark 4.2 now delivers EMR-class scan latency with page-level skipping, the combination AWS’s managed runtime still doesn’t offer. The boring default didn’t just stay correct. It got fast, with AWS’s own code.

Which Setup, at Which Size

Extrapolating the Crossover

Two matched-hardware points per family fit a trend but should not decide a migration alone, so here are the model, the numbers, and the assumptions. The page-blind reader’s time tracks rows decoded, floored at row-group granularity. The index-aware reader’s time tracks rows the layout actually needs plus fixed pruning overhead. The needed fraction falls with scale (measured at every decade), so the ratio between the readers grows, and each family crosses over where decode advantage equals waste ratio.

The cost rows use list price for the two fleets, $6.73 per hour on EC2 against $8.08 on EMR, and that 20 percent premium is why cost parity leads time parity by about half a decade of data size. At 1 TB, single-column queries already cost about the same on either platform while EMR still holds the stopwatch.

Assumptions, stated plainly. The projection extends a two-point trend, assumes the layout keeps improving its needed-rows fraction at the measured decelerating rate, and assumes decode bandwidth per core stays flat. The first two columns are measurements; treat the third as the hypothesis the next run tests.

Where This Goes Next

Two engines I have not measured are the obvious sequels, because both replace the reader and the layout machinery at once.

• Databricks. Photon is a proprietary vectorized reader, a black box from the outside. Does it evaluate parquet column indexes where the open-source lineage it left behind does not? The comparison stays identical to this campaign, OPTIMIZE ZORDER BY on the same three columns against a curve layout of the same table, so the only new variables are the reader and the platform. The probe carries over directly since counters, not timings, make it portable.
• Snowflake. Micro-partitions with their own metadata instead of parquet files, so the column-index question becomes a pruning-granularity question. How close does its pruning get to page level, and what does the same clustered workload cost end to end? Rows scanned per query is readable from its query profile, so the frac metric survives the translation.

Same discipline as everything above. Identical data, identical queries, counters over stopwatches, and every number published with the harness that produced it.