schenksj opened a new issue, #4842: URL: https://github.com/apache/datafusion-comet/issues/4842
> Built with Anthropic Fable # Comet Scan Performance: Root-Cause Analysis vs Plain Spark and Gluten --- ## Executive summary **The report "Comet scans are slower than plain Spark" pertains to both plain Parquet and Iceberg, but the causes are different — and a large share of the plain-Parquet reports trace to bugs that were diagnosed and fixed across Comet 0.15.0–0.17.0 (Apr–Jun 2026), with two scan-IO fixes merged to main but still unreleased as of 2026-07 (0.17.0 tagged 2026-06-16).** The canonical public case is the AWS EKS 3TB TPC-DS benchmark: Comet 0.14.0 was **11% slower** than vanilla Spark, and after the fixes 0.16.0 was **32% faster** (37% on Iceberg). So the first diagnostic question for any new report is *which Comet version and which storage backend*. **The report "much slower than Gluten" is structural and still true**: Comet's own published comparison (0.16.0 vs Gluten 1.6.0) shows Gluten ahead on TPC-H (2.8× vs 2.4× speedup), and the scan-side mechanisms responsible are identifiable in the Velox source and absent (or default-off) in Comet. Current-code causes, one line each (detail + evidence below): | # | Cause | Format | vs Spark | vs Gluten | |---|-------|--------|----------|-----------| | 1 | No IO/decode overlap by default: data cache + async prefetch exist but default **off**; demand-driven object-store reads | both | ✗ (loses on cold S3) | ✗✗ | | 2 | Row-level filter pushdown default **off**, and known to *regress* when enabled (#3457) → no late materialization | both | ✗ (Spark has page skip + lazy dict) | ✗✗ (Velox: full late materialization) | | 3 | Iceberg `dataFileConcurrencyLimit` default **1** — files read serially within a task | iceberg | ✗ | ✗ | | 4 | Iceberg path is a separate IO stack (iceberg-rust/opendal): bypasses cache/prefetch/s3a translation; per-file overhead (~30% CPU in worst cases, iceberg-rust #2172) | iceberg | ✗ | ✗ | | 5 | Per-batch schema-adaptation casts (INT96/ts-ms/unsigned/decimal/UUID; Iceberg field-ID re-projection) | both | ✗ (when triggered) | ✗ | | 6 | Dictionary encoding not preserved / no SIMD filter memoization; Utf8 not StringView | both | ~ (Spark has lazy dict decode) | ✗✗ | | 7 | Fallback cliffs silently hand the scan back to Spark + transition tax | both (worse for iceberg) | ✗ (adds C2R) | ✗ | | 8 | No cross-task footer/page-index reuse; per-task `RuntimeEnv` | both | ~ (parity: Spark also reads per task) | ~ (Velox has file-handle/data caches, but Gluten ships them default-off too) | | 9 | Dead tuning knobs: `comet.parquet.read.parallel.io.*`, `mergeRanges` are no-ops → operators "tune" nothing | both | trap | trap | **Historical (fixed — check the version before chasing these):** per-file ObjectStore/DNS storms and per-file S3 `HeadBucket` (fixed 0.15.0, PR #3802), OpenDAL `get_ranges` sequential-read regression on HDFS (fixed 0.15.0, PR #3965), DPP fallback that *dropped* the DPP filters and read all partitions (fixed 0.15.0, #3870/PR #3982), double memory pools (#3868/PR #3924, 0.15.0), `native_iceberg_compat`'s per-batch native→JVM→native FFI round-trip (path removed in **0.17.0**, #3431/#4020), page-index re-fetched from object storage per split (fixed on main 2026-06, PR #4707 — **not in any release yet**, ships post-0.17.0). **Architectural note that invalidates older analyses:** as of current main there is **no JVM-side Comet Parquet reader at all**. `common/.../comet/parquet/` is gone in upstream main; `native_comet`, `native_iceberg_compat`, and `spark.comet.scan.impl` no longer exist. Plain Parquet = DataFusion `DataSourceExec`/`ParquetSource` over arrow-rs + `object_store`. Iceberg = native `IcebergScanExec` over iceberg-rust + opendal (default on). Any explanation involving "per-page JNI transfer" or "JVM reads, native decodes" is about dead code. --- ## Part 1 — What the baseline and the competitor actually do ### 1.1 Plain Spark's vectorized reader (the bar to clear) Spark 3.5's Parquet scan is better than its reputation, and beating it requires matching four things: 1. **Footer read once.** The footer is read *with* row groups at reader-construction and reused; no second parse (`ParquetFileFormat.scala:209-216`, `SpecificParquetRecordReaderBase.java:104-106`). 2. **Row-group + page-level pruning by default.** `spark.sql.parquet.filterPushdown=true` (`SQLConf.scala:1029`) drives parquet-mr's `readNextFilteredRowGroup()` (`SpecificParquetRecordReaderBase.java:287`, reached via `checkEndOfRowGroup()` at `VectorizedParquetRecordReader.java:416-430`), which consults column/offset indexes and skips non-matching *pages*, with skip-aware decoders driven by the surviving row ranges (`ParquetReadState.java:102-128`). 3. **Lazy dictionary decode.** Dictionary-encoded columns keep dictionary IDs in the batch and decode at *access time* (`VectorizedColumnReader.java:203-238`, `OnHeapColumnVector.java:326-330`) — a filtered-away or projected-away dictionary column may never be decoded at all. 4. **Whole-stage codegen consumes ColumnVectors directly** — no row materialization between scan and fused filter/project (`Columnar.scala:103-198`). Spark's weaknesses (what Gluten exploits, and where Comet *could* win): decode is JVM code with no SIMD; IO is synchronous `readNextRowGroup` with **no** vectored/async IO in 3.5 (parquet-mr 1.13.1 predates vectored IO — relies purely on S3A/ABFS connector readahead); no row-level late materialization (the pushed filter only prunes, then FilterExec re-evaluates above the scan); on-heap by default. ### 1.2 Why Gluten/Velox scans are fast (mechanism catalogue) Ranked by contribution. All are on by default in Gluten **except the caches** (mechanism 7): the RAM/SSD `AsyncDataCache` is created only when `spark.gluten...cacheEnabled` is set (`VeloxBackend.cc:350`, default false) and the file-handle cache also defaults off (`VeloxConfig.h:165`) — so out of the box Gluten's edge is the IO architecture and decode path, not caching: 1. **Async split preloading on a dedicated IO thread pool.** While the driver thread processes split N, splits N+1..N+2 are opened, footer-read, and IO-scheduled on a separate `folly` executor (`velox/velox/exec/TableScan.cpp:452-517`; pool sized to task slots by default, created only when `IOThreads > 0`, `VeloxBackend.cc:232-239`; `SplitPreloadPerDriver=2`, `VeloxConfig.scala:241`). IO and compute overlap continuously. 2. **Coalesced, quantized, density-gated reads.** Nearby column chunks merge into single large IOs (Gluten defaults: 64 MB max coalesced, 512 KB distance, **256 MB load quantum** — `ConfigExtractor.cc:303-311`); large columns split into independently-prefetchable quanta; prefetch gated by measured access density ≥ 0.8 (`CachedBufferedInput.cpp:105-149,293-331`). Note the default no-cache path is `DirectBufferedInput`, which applies the same coalescing/quantization/prefetch scheme (`DirectBufferedInput.cpp:88-165`); `CachedBufferedInput` is the with-cache variant (`GlutenBufferedInputBuilder.h:37-51` selects between them). 3. **Late materialization.** Filter columns decode first; non-filter columns are `LazyVector`s decoded only for surviving rows — and only if something downstream actually reads them (`dwio/common/ColumnLoader.cpp:24-84`). 4. **Adaptive filter reordering.** Pushed filters re-sorted continuously by measured time-to-drop-value using RDTSC timings (`ScanSpec.cpp:62-117`, `SelectivityInfo.h:25-76`). 5. **Dictionary preservation + SIMD filter memoization.** Dictionary-encoded columns return `DictionaryVector` (indices + shared dictionary, no flatten — `StringColumnReader.cpp:47-59`); filters evaluate once per distinct dictionary code with an xsimd 8-way cached lookup (`ColumnVisitors.h:865-1024`). 6. **SIMD decode + zero-copy strings.** AVX2/BMI2 bit-unpacking (`BitPackDecoder.cpp:25-129`, guarded by `#if XSIMD_WITH_AVX2` at :23 — x86/AVX2 builds only, scalar fallback elsewhere), thread-local reused ZSTD contexts (`PageReader.cpp:179-183`), 16-byte inline `StringView`s pointing into decompressed buffers. 7. **RAM + SSD data cache available under the scan** (`AsyncDataCache`/`SsdCache`, wired at `VeloxBackend.cc:349-368` but **default-off** — gated on `kVeloxCacheEnabled`, `VeloxBackend.cc:350`) plus an optional file-handle cache (also default-off, `VeloxConfig.h:165`). On by default regardless of caching: the speculative 1 MB single-IO footer tail read (`ParquetReader.cpp:367-399`) and row-group prefetch-ahead (`ParquetReader.cpp:1344-1361`). 8. **Adaptive batch sizing + vector reuse** — output batch size adjusts to observed filter ratio; result vectors recycled across batches (`TableScan.cpp:534-549`, `SelectiveStructColumnReader.cpp:54-68`). 9. **Iceberg rides the same path.** `IcebergSplitReader` subclasses `FileSplitReader` — the same base as the plain Hive split reader — so it inherits *all* of the above, layering on delete handling (positional/equality delete readers, V3 deletion vectors, row lineage — `velox/connectors/hive/iceberg/IcebergSplitReader.h:32,128-211`). Gluten's Iceberg scan is therefore "fast Parquet scan + delete merge," not a separate slower stack. The JVM never touches data bytes in Gluten: Substrait `LocalFilesNode` ships file descriptors only — paths/offsets/lengths plus sizes, partition/metadata columns, and schema (`LocalFilesNode.java:35-108`). --- ## Part 2 — Plain-Parquet causes in current Comet (`native_datafusion`, the only path) ### P1. No IO/decode overlap by default — cold object-store scans are demand-driven **[highest impact on S3/ABFS/HDFS]** In the default config, the scan reads through DataFusion's `ParquetOpener` straight to the raw `object_store` client: no read-ahead layer, no byte cache, no overlap of fetch with decode. `maybe_wrap_with_data_cache` returns the raw store when the cache is disabled (`parquet_support.rs:559-581`), and the async prefetcher is only spawned when enabled (`planner.rs:1513-1526`). - `spark.comet.scan.dataCache.enabled` = **false** (`CometConf.scala:135-145`) - `spark.comet.scan.dataCache.prefetch.enabled` = **false**, and ignored unless the cache is on (`CometConf.scala:219-228`) Both Spark (via S3A/ABFS connector readahead) and especially Gluten (mechanisms 1–2 above) overlap IO with compute; default Comet does not. This branch's prefetcher (`prefetch.rs`: filter-aware row-group pruning, projected-chunk ranges, `buffer_unordered`) closes the gap *when enabled* — upstream, the same territory is issue #4695 / PR #4828. Additionally, within a task all files land in a **single file group** (`planner.rs:1506-1507`, `target_partitions = spark.task.cpus` ≈ 1, `jni_api.rs:558-563`), so N files are fetched and decoded strictly serially — parity with Spark, but no equivalent of Velox split preload. ### P2. Row-level filter pushdown is off by default — and turning it on is currently a regression **[highest impact on selective queries over wide tables]** `spark.comet.parquet.rowFilterPushdown.enabled` = **false** (`CometConf.scala:275-287`); only when true does Comet set DataFusion's `pushdown_filters`/`reorder_filters` (`jni_api.rs:586-594`). Format-level pruning (row-group stats, page index, bloom) **does** run regardless — the predicate reaches `source.predicate` via `try_pushdown_filters` (`parquet_exec.rs:164-187`) — so Comet matches Spark's pruning. What's missing is everything after pruning: - Spark lazily decodes dictionary columns and its fused codegen filter touches ColumnVectors directly. - Velox decodes filter columns first and materializes survivors only. - Comet fully decodes every projected column of every surviving page, then filters in a separate `CometFilter` operator. The kicker: upstream issue **#3457** (open) found that *enabling* the row-filter config makes TPC-H **worse** — arrow-rs's `RowFilter`/late-materialization machinery costs more than it saves in its current form — which is why it ships off. So this is not a flip-a-flag fix; it needs arrow-rs/DataFusion-level work (adaptive filter ordering — gap-assessment item #13 — is part of the same fix). ### P3. Per-batch schema-adaptation cast passes **[moderate; workload-dependent]** The `SparkPhysicalExprAdapterFactory` rewrites mismatched columns per batch: INT96→µs + tz re-tag, Timestamp(µs)→(ms) divide-by-1000 pass (`cast_column.rs:145-167,276-283`), unsigned/decimal promotion casts (`schema_adapter.rs:626,915-923`), dictionary `take`-materialization (`parquet_support.rs:175-195`), UUID FixedSizeBinary→String building a new array per value (`parquet_support.rs:237-256`). Conditional on type mismatch, but when triggered it's a full extra array pass per column per batch on top of decode. Upstream #3748 (native_datafusion ~2× memory of the old path, slower after SchemaAdapter change) is the live tracking issue; DataFusion #21158 (skip rewrite when schemas match) was a partial fix. ### P4. No dictionary preservation, no string views **[moderate CPU; biggest on low-cardinality strings]** Comet's scan output flattens dictionaries and produces `Utf8`, not `Utf8View` (existing gap-assessment items #14 and #6). Against *Spark* this is a real loss: Spark keeps dictionary IDs and decodes at access time, so `GROUP BY low_cardinality_string` over dictionary-encoded Parquet can be cheaper in vanilla Spark than in Comet, which materializes every string. Against Velox the gap is larger (dictionary passthrough + SIMD filter memoization + StringView). ### P5. Metadata handling: fine within a task, nothing across tasks **[minor vs Spark, real vs Gluten]** Footer + page index are read natively once per task and cached in the per-task `RuntimeEnv` (`parquet_exec.rs:150-162`; the per-split page-index re-fetch was fixed upstream by PR #4707, and #4717 added the footer size hint). But the `RuntimeEnv` is per-task (`jni_api.rs:543-598`), so two tasks on one executor re-fetch and re-parse the same footer. Spark has the same per-task behavior (parity); Velox caches file handles/footers process-wide. Note also `fetch_metadata` bypasses the `bytes_scanned` metric (`parquet_exec.rs:154-156`), so metadata IO is invisible when diagnosing. ### P6. Dead tuning knobs **[a trap, not a slowdown]** `spark.comet.parquet.read.parallel.io.enabled` (default true!), `...parallel.io.thread-pool.size`, `spark.comet.parquet.read.io.mergeRanges[.delta]`, and `COMET_IO_ADJUST_READRANGE_SKEW` are defined in `CometConf.scala:289-333` and **consumed nowhere** — leftovers from the deleted JVM reader. Anyone benchmarking "Comet with IO tuning" is tuning a no-op; the only real knobs are the (default-off) cache/prefetch settings and batch size (`COMET_BATCH_SIZE` = 8192, fine). ### P7. Fallback cliffs put Spark's reader back — with an added transition tax Metadata columns, `input_file_name()`, row-index columns, unsupported schemes, encryption configs, nested default values, etc. (`CometScanRule.scala:174-289`) silently revert the scan to `CometScanExec` (Spark reads, Arrow-FFI export) or full Spark. The FFI-fed `ScanExec` path pays a per-column copy-or-unpack on **every batch** (`operators/scan.rs:154-162`, `copy.rs:70-92`) — so a "Comet" plan whose scan fell back can genuinely be slower than never enabling Comet (the docs admit this: `datasources.md:26-28`). This — not the native reader — is a plausible explanation for many "Comet slower than Spark" field reports on plain Parquet, alongside the fixed historical bugs. --- ## Part 3 — Iceberg-specific causes (native `IcebergScanExec`, default on) Data flow is architecturally good — zero per-batch JNI crossings during the scan; iceberg-rust reads Parquet natively via opendal, batches stay native through the plan, and results cross to the JVM once per output batch via the standard Arrow C-FFI export (`prepare_output`, `jni_api.rs:650`, called from `executePlan`). The baseline here is also weaker than for plain Parquet: vanilla Spark reads Iceberg through **Iceberg's own Arrow-based vectorized reader** (default on, batch size 5000 — `TableProperties.java:260-264`, `SparkBatch.java:131-152`), which prunes at **row-group granularity only** (stats + dictionary + bloom, `ReadConf.java:90-113`) — it calls `readNextRowGroup()`, never `readNextFilteredRowGroup()` (`VectorizedParquetReader.java:161`), so it has **no column-index page skipping and no lazy dictionary decode**, and it falls to a row-based reader whenever any projected top-level column is non-primitive (`SparkBatch.java:181-183`). Comet's iceberg-rust path enables row selection (`with_row_selection_enabled(true)`), so where its predicate conversion succeeds it can actually prune *finer* than the vanilla Iceberg reader. The taxes are: ### I1. `spark.comet.scan.icebergNative.dataFileConcurrencyLimit` defaults to **1** Files in a task group are read strictly serially (`CometConf.scala:124-133` → `with_data_file_concurrency_limit`, `iceberg_scan.rs:204`). The docs admit the default exists "to maintain test behavior … without ORDER BY" and recommend 2–8 (`iceberg.md:55-58`); even Comet's own benchmark engine config leaves it at 1. On S3 this exposes full sequential GET latency per file. **This is probably the single cheapest Iceberg win available.** ### I2. Separate IO stack with none of the Parquet path's infrastructure The Iceberg path builds its own opendal `FileIO` — **per `execute()` call, per task** (`iceberg_scan.rs:169-207,345-364`) — and bypasses `maybe_wrap_with_data_cache` entirely: no block cache, no prefetcher, and the Hadoop `fs.s3a.*` → object_store translation doesn't apply (Iceberg reads use `s3./gcs./adls./client.` catalog properties instead, `iceberg_scan.rs:367`). Tuning that fixes plain-Parquet IO does nothing for Iceberg. Upstream, iceberg-rust's per-file overhead was measured at **~30% of executor CPU** in Comet-driven scans (iceberg-rust epic #2172: per-task operator creation, `stat()` calls, TLS handshakes, credential init); the metadata-prefetch, file-size-from-manifest, double-open, and range-coalescing fixes merged Feb–Mar 2026, but **operator/FileIO caching was deferred** (#2177 closed unmerged). ### I3. Per-batch re-projection whenever schemas aren't pointer-identical Every batch passes `adapt_batch_with_expressions`; the whole-batch zero-copy fast path requires `Arc::ptr_eq(batch.schema(), target_schema)` (`iceberg_scan.rs:595-621`). When that fails (schema evolution, field-ID renames, type promotion, metadata differences), every projected column is re-evaluated per batch (`iceberg_scan.rs:615-618`) — but columns that match by name/type reduce to plain `Column` expressions whose evaluation is a zero-copy Arc clone; only genuinely mismatched columns are **materialized** (cast/promoted), plus a shallow `RecordBatch` rebuild per batch. Still a *double* mapping (iceberg-rust already projected by field ID: `planner.rs:3844-3855`), and expression construction is cached per file schema while evaluation runs per batch — but the cost is proportional to the number of mismatched columns, not the full projection. ### I4. Merge-on-read: one extra HEAD per unique delete file per task Delete-file sizes aren't serialized (arrive as 0, `planner.rs:3744`), so the native side must `stat()` each unique delete file before reading it (`iceberg_scan.rs:265-343`; the in-code comment cites apache/iceberg#12554 — a `rewrite_table_path` stale-size bug, closed 2026-06-27, which is why manifest-carried sizes weren't trusted; the workaround remains in Comet). Plus the standard equality-delete anti-join cost inside iceberg-rust. This is a genuinely Comet-specific tax: vanilla Spark-Iceberg takes delete sizes straight from the manifest (`ContentFile.fileSizeInBytes()`/`DeleteFile.contentSizeInBytes()`, read via `IOUtil.readFully` in `BaseDeleteLoader.java:173-178` — no stat), applies positional deletes as a vectorized row-id mapping without copying rows (`ColumnarBatchUtil.buildRowIdMapping` + `ColumnVectorWithFilter`), and loads deletes on a worker pool with optional executor-side caching. Gluten's delete-bitmap approach likewise rides its coalesced IO path. ### I5. Narrower pushdown + a large fallback surface Only the identity-transform residual predicate with `=,≠,<,≤,>,≥,IN,NOT IN,IS (NOT) NULL,AND,OR,NOT` is pushed (`CometIcebergNativeScan.scala:568-642`); anything else is unfiltered at the reader (correct, just slower). The comparison with vanilla cuts both ways: vanilla Iceberg hands its *entire* residual to its reader but uses it only for row-group pruning (stats/dict/bloom — no page-level skip), while Comet pushes a narrower operator subset that iceberg-rust applies at row-group *and* page/row-selection granularity — so on supported predicates Comet prunes finer, on unsupported ones coarser. Hard fallbacks to **plain Spark Iceberg** (no acceleration at all): format v3, ORC/Avro data files, `truncate/bucket/year/month/day/hour` residuals, `IS NULL` on complex types, equality deletes on structs, binary/fixed or >28-precision-decimal partition columns, DPP-under-AQE with non-`InSubqueryExec` subqueries (`CometScanRule.scala:347-649`). Each is a cliff where the user *think s* Comet is running and it isn't. Community discussion #3199 (Glue-catalog Iceberg, "Spark faster than Comet across many configs") is consistent with exactly this + transition tax. ### I6. Driver-side planning walks all `FileScanTask`s twice via reflection Validation pass + serialization pass are separate full traversals of all tasks. The two main walks *do* cache their reflection method handles (`CometIcebergNativeScan.scala:767-774`, `CometScanRule.scala:827-833`), but per-field `getMethod` lookups persist inside `serializePartitionData` (`CometIcebergNativeScan.scala:349-450`) and `buildFieldIdMapping` (`IcebergReflection.scala:553-579`, run per task in the no-deletes branch). Scales with file count; for 10k+-file tables this is real driver latency before the first byte is read. **vs Gluten on Iceberg:** Gluten inherits its entire fast scan for Iceberg (delete bitmap on top — Part 1.2 #9). So the Iceberg gap ≈ the plain-Parquet gap **plus** I1–I6. This matches the field observation that Comet is "much slower than Gluten" and that the gap is worse on Iceberg than plain Parquet. --- ## Part 4 — Historical causes now fixed (version triage) If the report comes from Comet **≤ 0.14.x** or an un-pinned build from early 2026, these fixed issues likely dominate; upgrade before analyzing further: | Issue | Symptom | Fixed | |---|---|---| | PR #3802 (epic #3799) | New ObjectStore + reqwest client + DNS + S3 `HeadBucket` **per file** — up to 5,000 DNS q/s/pod, ~500× vanilla per the EKS blog post; Route53 throttling | 0.15.0 (global store cache keyed by URL+config hash; per-bucket region cache). Workaround: set `fs.s3a.endpoint.region` | | #3926 / PR #3965 | HDFS scan task 3 min → 5 min after OpenDAL bump: `get_ranges` degraded ~2× (opendal#7380) | 0.15.0 (2026-04) | | #3870 | DPP fallback **dropped the DPP filters** → scanned all partitions (q25 blowup) | 0.15.0 (PR #3982); native DPP in 0.16.0 | | #3868 / PR #3924 | Needed ≥32 GB off-heap where Spark didn't (two native memory pools per task) | 0.15.0 (2026-04) | | PR #4707 | Page index re-fetched **from object storage per split** (GBs wasted on TPC-DS q88) | merged 2026-06-23 — **unreleased** (post-0.17.0; only on main) | | PR #4717 | Parquet footer read took 3 metadata round-trips (no size hint) | merged 2026-06-24 — **unreleased** (post-0.17.0) | | #3431 / #4020 | `native_iceberg_compat` round-tripped every batch native→JVM→native with per-batch schema serialization | Path deleted in **0.17.0** (PRs #4019/#4363, 2026-05) | | #2878 | native_datafusion *planning* 10–30× slower per query | fixed 2026-02 | Residual credential caveat in current code: the object-store cache key includes static credentials, so hourly-rotating static `fs.s3a.access.key`/`session.token` deployments rebuild the client (new HTTP pool) on every rotation (`parquet_support.rs:544-550,706-731`); dynamic providers (IMDS/STS) are unaffected. Still-open upstream perf issues worth tracking: #4361 (TPC-DS q50 0.77× — undiagnosed), #3457 (row-filter pushdown regression), #3748 (SchemaAdapter memory/CPU), #4072 (per-batch JNI metrics protobuf round-trip), #4614 (Iceberg q72 13s→42s — two candidate causes per the issue, lost WSCG join fusion vs. slower native non-equi SMJ, only under experimental `sortMergeJoinWithJoinFilter`; not scan), iceberg-rust #2177 (operator caching — closed unmerged 2026-07-06, deferred) and #2220 (parallel file-level scan, open). --- ## Part 5 — Leverage-ordered remediation | # | Action | Fixes | Effort | Feasibility | |---|--------|-------|--------|-------------| | 1 | **Triage version/storage first**: if report predates 0.15/0.16 or lacks `fs.s3a.endpoint.region`, re-benchmark on 0.17+ (or main — the page-index/footer-hint fixes #4707/#4717 are merged but unreleased) before any code work | Part 4 class | XS | High | | 2 | **Raise Iceberg `dataFileConcurrencyLimit` default** (needs the ordering-nondeterminism in tests resolved, or default-N with per-query ordering guard) | I1 | S | High — config + test work only | | 3 | **Productionize data cache + prefetch and default them on** (this fork's `feat/scan-prefetch` + object-store cache; upstream #4695/PR #4828) — the only mechanism matching Velox Tier-1 (split preload + coalesced prefetch + RAM/SSD cache) | P1 | M (in flight) | High — code exists on this branch; needs hardening + defaults debate upstream | | 4 | **Wire the Iceberg path into the same byte cache / prefetcher** — either an opendal layer over the caching store or teach `IcebergScanExec` to use `caching_store_for`; also hoist `FileIO` construction out of `execute()` and cache per (config-hash) process-wide | I2 | M | Medium — two IO stacks to bridge; opendal `Layer` API makes it tractable | | 5 | **Fix row-filter pushdown so it can default on**: profile #3457, add measured filter reordering (gap item #13) and cheap-filter-first policy in arrow-rs `row_filter.rs`; until decode-time filtering wins, at minimum stop decoding non-filter columns for fully-pruned pages | P2 | M–L (upstream arrow-rs/DataFusion) | Medium — known-hard; upstream is receptive, #3457 already tracks it | | 6 | **Cheapen schema adaptation**: replace Iceberg's `Arc::ptr_eq` fast path with structural schema equality; skip identity casts (upstream DF #21158 pattern) on both paths; special-case UUID and timestamp-unit rewrites into decode where possible | P3, I3 | S–M | High | | 7 | **Consume iceberg-rust fixes + push the deferred ones**: bump to a rev with metadata prefetch/file-size/coalescing (#2173/#2175/#2181), serialize `file_size_in_bytes` for delete files to kill the per-file HEAD (iceberg#12554 — the stale-size bug that motivated distrust — closed 2026-06-27, so manifest sizes can now be plumbed through), revive operator caching (iceberg-rust #2177, closed unmerged) | I2, I4 | S (bump) + M (upstream) | High / Medium | | 8 | **Process-wide Parquet metadata cache**: share a `CacheManager`/file-metadata cache across tasks in the executor (the per-task `RuntimeEnv` boundary is the obstacle); also count `fetch_metadata` in `bytes_scanned` | P5 | M | Medium — memory-accounting design needed | | 9 | **Delete or implement the dead IO configs** (`parallel.io.*`, `mergeRanges`) so benchmarking isn't misled | P6 | XS | High | | 10 | **Dictionary preservation + Utf8View at scan output** (existing gap items #14/#6) — closes the lazy-dictionary loss vs Spark and part of the CPU gap vs Velox | P4 | L | Medium — staged: view types first, then encoding-transparent operators | | 11 | **Shrink fallback cliffs + make them loud**: metadata columns/row-index/`input_file_name` support on the Parquet side; Iceberg transform residuals (push unpushable residual as post-filter instead of falling back); surface fallback reasons in the UI by default | P7, I5 | M–L (per item) | Medium | | 12 | **Iceberg driver planning**: single-pass validate+serialize, cache `Method` handles per class, parallelize task serialization | I6 | S–M | High | Items 3+4 together are the answer to "much slower than Gluten" on IO-bound workloads; items 5+10 are the answer on CPU-bound workloads; item 2 is the quick Iceberg win; items 1+9 prevent misdiagnosis. --- ## Appendix — how the three engines read a Parquet file today | Dimension | Spark 3.5 | Comet (current main) | Gluten/Velox | |---|---|---|---| | Data bytes touched by JVM | yes (decode in JVM) | no (native decode, arrow-rs) | no (native decode, C++) | | Footer reads per split | 1 (JVM) | 1 (native, per-task cache) | 1, speculative 1 MB tail read (handle cache exists, default-off) | | Row-group/page pruning | yes/yes (default) | yes/yes (default) | yes/yes (default) | | Row-level late materialization | no | off (regresses when on, #3457) | yes (LazyVector) | | Adaptive filter ordering | no | no | yes (RDTSC-measured) | | Dictionary handling | lazy decode at access | flatten at scan | preserve + SIMD filter memoization | | IO/compute overlap | FS-connector readahead only | none by default (cache+prefetch experimental) | split preload + quantized coalesced prefetch, dedicated IO pool | | Data cache | no | experimental, off | RAM + SSD, on with cache configured | | Strings | UTF8String (copy) | Utf8 (copy; no views) | StringView (zero-copy) | | Iceberg | Iceberg's own Arrow vectorized reader (default on, batch 5000; row-group pruning only — no page skip, no lazy dict; nested types → row reader) | separate native stack (iceberg-rust/opendal) with page-level row selection, but conc=1, no cache | same fast path + delete handling | -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
