Accelerating Apache Iceberg Parquet Scans using Comet#

Native Reader#

Comet’s native Iceberg reader relies on reflection to extract FileScanTasks from Iceberg, which are then serialized to Comet’s native execution engine (see PR #2528).

The example below uses Spark’s package downloader to retrieve Comet 1.1.0-SNAPSHOT and Iceberg 1.8.1, but Comet has been tested with Iceberg 1.5, 1.7, 1.8, 1.9, 1.10, and 1.11. The native Iceberg reader is enabled by default. To disable it, set spark.comet.scan.icebergNative.enabled=false.

The example uses the Spark 3.5 / Scala 2.12 build of Comet; substitute the Comet artifact matching your Spark and Scala versions (Comet also ships Spark 3.5 / Scala 2.13 and Spark 4.0/4.1 / Scala 2.13 jars; see the installation guide for the full list).

$SPARK_HOME/bin/spark-shell \
    --packages org.apache.datafusion:comet-spark-spark3.5_2.12:1.1.0-SNAPSHOT,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.apache.iceberg:iceberg-core:1.8.1 \
    --repositories https://repo1.maven.org/maven2/ \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
    --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.spark_catalog.type=hadoop \
    --conf spark.sql.catalog.spark_catalog.warehouse=/tmp/warehouse \
    --conf spark.plugins=org.apache.spark.CometPlugin \
    --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
    --conf spark.comet.explain.fallback.enabled=true \
    --conf spark.memory.offHeap.enabled=true \
    --conf spark.memory.offHeap.size=2g

Catalog configuration is standard Iceberg-on-Spark and independent of Comet. The native reader has been tested with Hadoop, Hive, and REST catalogs. The example above uses a Hadoop catalog. For the full catalog configuration reference, see Iceberg’s Spark catalog configuration.

Tuning#

Comet’s native Iceberg reader supports fetching multiple files in parallel to hide I/O latency with the config spark.comet.scan.icebergNative.dataFileConcurrencyLimit. This value defaults to 1 to maintain test behavior on Iceberg Java tests without ORDER BY clauses, but we suggest increasing it to values between 2 and 8 based on your workload.

Supported features#

The native Iceberg reader supports the following features:

Table specifications:

  • Iceberg table spec v1, v2, and v3

Encryption:

  • Encrypted v3 tables using 128-bit or 256-bit AES-GCM data keys (requires Iceberg 1.11 or newer). Iceberg-Java unwraps the key envelope on the driver during planning and stores the plaintext data key in each file’s key_metadata, which the native reader uses directly, so no KMS integration is needed on the native side. Iceberg-Java also permits 192-bit data keys; those tables fall back to Spark (no AES-192-GCM in the underlying crypto).

Schema and data types:

  • All primitive types including UUID

  • Complex types: arrays, maps, and structs

  • Schema evolution (adding and dropping columns)

Time travel and branching:

  • VERSION AS OF queries to read historical snapshots

  • Branch reads for accessing named branches

Delete handling (Merge-On-Read tables):

  • Positional deletes

  • Equality deletes

  • Mixed delete types

  • Deletion vectors (v3)

Filter pushdown:

  • Equality and comparison predicates (=, !=, >, >=, <, <=)

  • Logical operators (AND, OR)

  • NULL checks (IS NULL, IS NOT NULL) on primitive columns

  • IN and NOT IN list operations

  • BETWEEN operations

NULL checks on struct, array, and map columns still use native scans and return correct results, but are not pushed into iceberg-rust, which binds accessors only for primitive fields. These residuals provide no native row-group pruning, nor do conjunctions containing them; safe partial pruning is tracked in #5883.

Partitioning:

  • Standard partitioning with partition pruning

  • Date partitioning with days() transform

  • Bucket partitioning

  • Truncate transform

  • Hour transform

Storage:

  • Local filesystem

  • S3-compatible storage (AWS S3, MinIO)

  • Google Cloud Storage (gs) and Alibaba Cloud OSS (oss)

HDFS-backed tables are not supported by the native Iceberg reader and fall back to Spark.

REST Catalog#

Comet’s native Iceberg reader also supports REST catalogs. The following example shows how to configure Spark to use a REST catalog with Comet’s native Iceberg scan:

$SPARK_HOME/bin/spark-shell \
    --packages org.apache.datafusion:comet-spark-spark3.5_2.12:1.1.0-SNAPSHOT,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.apache.iceberg:iceberg-core:1.8.1 \
    --repositories https://repo1.maven.org/maven2/ \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
    --conf spark.sql.catalog.rest_cat=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.rest_cat.catalog-impl=org.apache.iceberg.rest.RESTCatalog \
    --conf spark.sql.catalog.rest_cat.uri=http://localhost:8181 \
    --conf spark.sql.catalog.rest_cat.warehouse=/tmp/warehouse \
    --conf spark.plugins=org.apache.spark.CometPlugin \
    --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
    --conf spark.comet.explain.fallback.enabled=true \
    --conf spark.memory.offHeap.enabled=true \
    --conf spark.memory.offHeap.size=2g

Note that REST catalogs require explicit namespace creation before creating tables:

scala> spark.sql("CREATE NAMESPACE rest_cat.db")
scala> spark.sql("CREATE TABLE rest_cat.db.test_table (id INT, name STRING) USING iceberg")
scala> spark.sql("INSERT INTO rest_cat.db.test_table VALUES (1, 'Alice'), (2, 'Bob')")
scala> spark.sql("SELECT * FROM rest_cat.db.test_table").show()

Object store configuration (S3)#

The native reader has its own Rust object store client and does not go through Iceberg’s JVM FileIO, neither S3FileIO nor the older Hadoop S3A filesystem. It configures that client from the catalog’s s3.* properties (the same keys S3FileIO reads), from spark.hadoop.fs.s3a.* settings, or, for a scheme opted into spark.hadoop.fs.comet.s3Compliant.schemes, from vendor-style fs.<scheme>.<authority>.* keys (see S3-Compliant Filesystem Schemes). That third source is translated into the same fs.s3a.* shape as the second before it reaches the reader. S3 configuration therefore reaches the native reader through one of these three channels.

For a custom S3-compatible endpoint, configure the catalog with the endpoint, path-style access, region, and credentials (Hive shown):

    --conf spark.sql.catalog.s3_cat=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.s3_cat.type=hive \
    --conf spark.sql.catalog.s3_cat.uri=thrift://metastore:9083 \
    --conf spark.sql.catalog.s3_cat.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
    --conf spark.sql.catalog.s3_cat.s3.endpoint=https://s3.example.com:9000 \
    --conf spark.sql.catalog.s3_cat.s3.path-style-access=true \
    --conf spark.sql.catalog.s3_cat.client.region=us-east-1 \
    --conf spark.sql.catalog.s3_cat.s3.access-key-id=... \
    --conf spark.sql.catalog.s3_cat.s3.secret-access-key=...

These s3.* storage properties are not specific to the Hive catalog shown here. When s3.access-key-id / s3.secret-access-key are omitted, credentials come from the standard AWS chain (environment variables, instance profiles, and so on). client.region is auto-detected for AWS but should be set for non-AWS endpoints. If your REST catalog vends temporary credentials, the native reader does not consume them automatically, and wiring that requires the credential provider bridge. See Iceberg’s S3 FileIO docs for the full property list, and S3 Credential Providers for vended or per-request credentials.

Current limitations#

The following scenarios will fall back to the JVM Iceberg reader:

  • Iceberg table spec v4 or newer

  • v3 tables with columns that declare an initial default value

  • v3 column types the native reader cannot read (variant, geometry, geography, unknown)

  • Encrypted tables with 192-bit data keys (no AES-192-GCM in the underlying crypto)

  • Delete files in a format other than Parquet or Puffin (Avro or ORC positional/equality deletes)

  • Iceberg writes (reads are accelerated, writes use Spark)

  • Tables backed by Avro or ORC data files (only Parquet is accelerated)

  • Tables partitioned on BINARY or DECIMAL (with precision >28) columns

  • Scans with residual filters using truncate, bucket, year, month, day, or hour transform functions (partition pruning still works, but row-level filtering of these transforms falls back)

Iceberg UDFs#

Iceberg ships several ScalaUDFs that surface in user queries and maintenance actions:

  • IcebergSpark.registerBucketUDF and registerTruncateUDF register bucket(N, col) and truncate(W, col) for use in SELECT / JOIN / WHERE predicates that align with hidden partitioning.

  • RewriteDataFiles with sort-strategy=zorder builds a tree of per-type ordered-bytes UDFs (INT_ORDERED_BYTES, LONG_ORDERED_BYTES, …, INTERLEAVE_BYTES) over the sort key columns during compaction.

Scala UDF and Java UDF Support is enabled by default (spark.comet.exec.scalaUDF.codegen.enabled=true), so these UDFs run through native execution and the project, exchange, and sort operators around them stay on the Comet path end-to-end. Setting spark.comet.exec.scalaUDF.codegen.enabled=false causes the enclosing operator to fall back to Spark, which forces a columnar-to-row roundtrip and demotes the surrounding shuffle from CometExchange to CometColumnarExchange.

Iceberg system functions#

Iceberg’s system functions bucket, truncate, years, months, days, and hours (the SQL form of its partition transforms, for example SELECT system.bucket(16, id) FROM t) run natively. Spark binds them as static invocations of Iceberg’s per-type implementations under org.apache.iceberg.spark.functions, and Comet recognizes those classes wherever the expression appears: in a projection, a filter, a sort key, or the hash partitioning of a shuffle.

The native kernels reproduce Iceberg’s Java semantics exactly rather than approximately:

  • bucket hashes the spec’s byte encoding of each value (8-byte little-endian for integers, dates, and timestamps; UTF-8 for strings; raw bytes for binary; the minimal big-endian two’s complement of the unscaled value for decimals) with 32-bit Murmur3 and masks the sign bit before taking the modulus.

  • truncate uses Java’s wrapping integer arithmetic and counts code points (not bytes) for strings. Decimal inputs are the one case that stays with Spark, see below.

  • years, months, days, and hours are evaluated in UTC regardless of the session timezone and go negative before the epoch; days returns a date, the other three an int. They cover the whole DATE and TIMESTAMP domain, as Iceberg’s DateTimeUtil does.

truncate on a decimal column falls back to Spark. Truncating a negative decimal grows its magnitude, so the result can need one more digit than the column’s precision allows: truncate(10, v) on a decimal(18,4) value of -99999999999999.9999 is -100000000000000.0000, which has 19 digits. Iceberg’s TruncateDecimal hands that oversized value back unchanged and Spark turns it into null only when the row is materialized. An Arrow Decimal128(precision, scale) array has no encoding for that intermediate, so a native kernel would have to null it during evaluation, which changes what an enclosing predicate or hash sees. Every other truncate input type, and bucket on decimals, runs natively.

This matters most for writes. A partitioned table with the default write.distribution-mode (hash) is planned with a shuffle and a local sort keyed on the partition transforms, and with these functions native the whole sub-plan feeding the native Iceberg writer stays in Comet. A numBuckets or width argument that is not a positive integer literal makes the expression fall back to Spark.

Task input metrics#

The native Iceberg reader populates Spark’s task-level inputMetrics.bytesRead (visible in the Spark UI Stages tab) using the bytes_read counter from iceberg-rust’s ScanMetrics. This counter includes bytes read from both data files and delete files.

Iceberg Java does not explicitly report bytesRead to Spark’s task input metrics. On the iceberg Java path, any bytesRead value comes from Hadoop’s filesystem-level I/O counters, not from Iceberg itself. Because Comet’s native reader and the Hadoop filesystem use different counting mechanisms, the exact byte counts will differ between the two paths.