Comet Parquet Scan Implementations#

Comet currently has three distinct implementations of the Parquet scan operator. The configuration property spark.comet.scan.impl is used to select an implementation. The default setting is spark.comet.scan.impl=auto, and Comet will choose the most appropriate implementation based on the Parquet schema and other Comet configuration settings. Most users should not need to change this setting. However, it is possible to force Comet to try and use a particular implementation for all scan operations by setting this configuration property to one of the following implementations.

Implementation

Description

native_comet

Deprecated. This implementation provides strong compatibility with Spark but does not support complex types. This is the original scan implementation in Comet and will be removed in a future release.

native_iceberg_compat

This implementation delegates to DataFusion’s DataSourceExec but uses a hybrid approach of JVM and native code. This scan is designed to be integrated with Iceberg in the future.

native_datafusion

This experimental implementation delegates to DataFusion’s DataSourceExec for full native execution. There are known compatibility issues when using this scan.

The native_datafusion and native_iceberg_compat scans provide the following benefits over the native_comet implementation:

  • Leverages the DataFusion community’s ongoing improvements to DataSourceExec

  • Provides support for reading complex types (structs, arrays, and maps)

  • Removes the use of reusable mutable-buffers in Comet, which is complex to maintain

  • Improves performance

The native_datafusion and native_iceberg_compat scans share the following limitations:

  • When reading Parquet files written by systems other than Spark that contain columns with the logical type UINT_8 (unsigned 8-bit integers), Comet may produce different results than Spark. Spark maps UINT_8 to ShortType, but Comet’s Arrow-based readers respect the unsigned type and read the data as unsigned rather than signed. Since Comet cannot distinguish ShortType columns that came from UINT_8 versus signed INT16, by default Comet falls back to Spark when scanning Parquet files containing ShortType columns. This behavior can be disabled by setting spark.comet.scan.unsignedSmallIntSafetyCheck=false. Note that ByteType columns are always safe because they can only come from signed INT8, where truncation preserves the signed value.

  • No support for default values that are nested types (e.g., maps, arrays, structs). Literal default values are supported.

  • No support for datetime rebasing detection or the spark.comet.exceptionOnDatetimeRebase configuration. When reading Parquet files containing dates or timestamps written before Spark 3.0 (which used a hybrid Julian/Gregorian calendar), the native_comet implementation can detect these legacy values and either throw an exception or read them without rebasing. The DataFusion-based implementations do not have this detection capability and will read all dates/timestamps as if they were written using the Proleptic Gregorian calendar. This may produce incorrect results for dates before October 15, 1582.

  • No support for Spark’s Datasource V2 API. When spark.sql.sources.useV1SourceList does not include parquet, Spark uses the V2 API for Parquet scans. The DataFusion-based implementations only support the V1 API, so Comet will fall back to native_comet when V2 is enabled.

The native_datafusion scan has some additional limitations:

  • No support for row indexes

  • PARQUET_FIELD_ID_READ_ENABLED is not respected #1758

  • There are failures in the Spark SQL test suite #1545

  • Setting Spark configs ignoreMissingFiles or ignoreCorruptFiles to true is not compatible with Spark

S3 Support#

There are some differences in S3 support between the scan implementations.

native_comet (Deprecated)#

Note: The native_comet scan implementation is deprecated and will be removed in a future release.

The native_comet Parquet scan implementation reads data from S3 using the Hadoop-AWS module, which is identical to the approach commonly used with vanilla Spark. AWS credential configuration and other Hadoop S3A configurations works the same way as in vanilla Spark.

native_datafusion and native_iceberg_compat#

The native_datafusion and native_iceberg_compat Parquet scan implementations completely offload data loading to native code. They use the object_store crate to read data from S3 and support configuring S3 access using standard Hadoop S3A configurations by translating them to the object_store crate’s format.

This implementation maintains compatibility with existing Hadoop S3A configurations, so existing code will continue to work as long as the configurations are supported and can be translated without loss of functionality.

Additional S3 Configuration Options#

Beyond credential providers, the native_datafusion implementation supports additional S3 configuration options:

Option

Description

fs.s3a.endpoint

The endpoint of the S3 service

fs.s3a.endpoint.region

The AWS region for the S3 service. If not specified, the region will be auto-detected.

fs.s3a.path.style.access

Whether to use path style access for the S3 service (true/false, defaults to virtual hosted style)

fs.s3a.requester.pays.enabled

Whether to enable requester pays for S3 requests (true/false)

All configuration options support bucket-specific overrides using the pattern fs.s3a.bucket.{bucket-name}.{option}.

Examples#

The following examples demonstrate how to configure S3 access with the native_datafusion Parquet scan implementation using different authentication methods.

Example 1: Simple Credentials

This example shows how to access a private S3 bucket using an access key and secret key. The fs.s3a.aws.credentials.provider configuration can be omitted since org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider is included in Hadoop S3A’s default credential provider chain.

$SPARK_HOME/bin/spark-shell \
...
--conf spark.comet.scan.impl=native_datafusion \
--conf spark.hadoop.fs.s3a.access.key=my-access-key \
--conf spark.hadoop.fs.s3a.secret.key=my-secret-key
...

Example 2: Assume Role with Web Identity Token

This example demonstrates using an assumed role credential to access a private S3 bucket, where the base credential for assuming the role is provided by a web identity token credentials provider.

$SPARK_HOME/bin/spark-shell \
...
--conf spark.comet.scan.impl=native_datafusion \
--conf spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider \
--conf spark.hadoop.fs.s3a.assumed.role.arn=arn:aws:iam::123456789012:role/my-role \
--conf spark.hadoop.fs.s3a.assumed.role.session.name=my-session \
--conf spark.hadoop.fs.s3a.assumed.role.credentials.provider=com.amazonaws.auth.WebIdentityTokenCredentialsProvider
...

Limitations#

The S3 support of native_datafusion has the following limitations:

  1. Partial Hadoop S3A configuration support: Not all Hadoop S3A configurations are currently supported. Only the configurations listed in the tables above are translated and applied to the underlying object_store crate.

  2. Custom credential providers: Custom implementations of AWS credential providers are not supported. The implementation only supports the standard credential providers listed in the table above. We are planning to add support for custom credential providers through a JNI-based adapter that will allow calling Java credential providers from native code. See issue #1829 for more details.