parthchandra commented on code in PR #2681: URL: https://github.com/apache/datafusion-comet/pull/2681#discussion_r2500023366
########## docs/source/contributor-guide/parquet_scans.md: ########## @@ -131,7 +131,505 @@ The S3 support of `native_datafusion` has the following limitations: 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](https://github.com/apache/datafusion-comet/issues/1829) for more details. +## Architecture Diagrams + +This section provides detailed architecture diagrams for each of the three Parquet scan implementations. + +### `native_comet` Scan Architecture + +The `native_comet` scan is a hybrid approach where the JVM reads Parquet files and passes data to native execution. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_COMET SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_comet" │ +└───────────┬───────────────────┘ + │ + │ wraps + ↓ +┌───────────────────────────────┐ +│ FileSourceScanExec │ Original Spark scan (from Catalyst) +│ (from Spark) │ +└───────────┬───────────────────┘ + │ + │ creates RDD using + ↓ +┌───────────────────────────────┐ +│ CometParquetFileFormat │ Custom Parquet file format handler +│ (CometParquetFileFormat.scala)│ +│ │ +│ Key methods: │ +│ - buildReaderWithPartition │ +│ Values() │ +│ - supportBatch() │ +└───────────┬───────────────────┘ + │ + │ creates via factory + ↓ +┌───────────────────────────────┐ +│ CometParquetPartitionReader │ Factory for creating partition readers +│ Factory (with prefetch) │ Supports asynchronous prefetch optimization +│ (.../CometParquetPartition │ +│ ReaderFactory.scala) │ +└───────────┬───────────────────┘ + │ + │ creates + ↓ +┌───────────────────────────────┐ +│ BatchReader │ Main Parquet reader +│ (BatchReader.java:90) │ Reads Parquet files via Hadoop APIs +│ │ +│ Key components: │ +│ ├─ ParquetFileReader │ Opens Parquet file, reads metadata +│ ├─ PageReadStore │ Reads column pages from file +│ ├─ ColumnDescriptors │ Parquet column metadata +│ └─ Capacity (batch size) │ Configurable batch size +│ │ +│ Process per row group: │ +│ 1. Read file footer/metadata │ +│ 2. For each column: │ +│ - Get ColumnDescriptor │ +│ - Read pages via │ +│ PageReadStore │ +│ - Create CometVector │ +│ from native data │ +│ 3. Return ColumnarBatch │ +└───────────┬───────────────────┘ + │ + │ Uses JNI to access native decoders + │ (not for page reading, only for + │ specialized operations if needed) + │ + ↓ +┌───────────────────────────────┐ +│ ColumnarBatch │ +│ with CometVector[] │ Arrow-compatible columnar vectors +│ │ +│ Each CometVector wraps: │ +│ - Arrow ColumnVector │ +│ - Native memory buffers │ +└───────────┬───────────────────┘ + │ + │ Passed to native execution via JNI + │ + ↓ ┌───────────────────────────┐ +┌───────────────────────────────┐ │ ScanExec │ +│ CometExecIterator │──────→│ (operators/scan.rs:54) │ +│ (JNI boundary) │ JNI │ │ +└───────────────────────────────┘ │ Reads batches from JVM │ + │ via CometBatchIterator │ + │ │ + │ Key operations: │ + │ ├─ next_batch() │ Review Comment: There is one more JNI call here. `ScanExec.get_next` makes a call back to `CometBatchIterator` which exports the batch back to native (so an FFI call) ########## docs/source/contributor-guide/parquet_scans.md: ########## @@ -131,7 +131,505 @@ The S3 support of `native_datafusion` has the following limitations: 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](https://github.com/apache/datafusion-comet/issues/1829) for more details. +## Architecture Diagrams + +This section provides detailed architecture diagrams for each of the three Parquet scan implementations. + +### `native_comet` Scan Architecture + +The `native_comet` scan is a hybrid approach where the JVM reads Parquet files and passes data to native execution. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_COMET SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_comet" │ +└───────────┬───────────────────┘ + │ + │ wraps + ↓ +┌───────────────────────────────┐ +│ FileSourceScanExec │ Original Spark scan (from Catalyst) +│ (from Spark) │ +└───────────┬───────────────────┘ + │ + │ creates RDD using + ↓ +┌───────────────────────────────┐ +│ CometParquetFileFormat │ Custom Parquet file format handler +│ (CometParquetFileFormat.scala)│ +│ │ +│ Key methods: │ +│ - buildReaderWithPartition │ +│ Values() │ +│ - supportBatch() │ +└───────────┬───────────────────┘ + │ + │ creates via factory + ↓ +┌───────────────────────────────┐ +│ CometParquetPartitionReader │ Factory for creating partition readers +│ Factory (with prefetch) │ Supports asynchronous prefetch optimization +│ (.../CometParquetPartition │ +│ ReaderFactory.scala) │ +└───────────┬───────────────────┘ + │ + │ creates + ↓ +┌───────────────────────────────┐ +│ BatchReader │ Main Parquet reader +│ (BatchReader.java:90) │ Reads Parquet files via Hadoop APIs +│ │ +│ Key components: │ +│ ├─ ParquetFileReader │ Opens Parquet file, reads metadata +│ ├─ PageReadStore │ Reads column pages from file +│ ├─ ColumnDescriptors │ Parquet column metadata +│ └─ Capacity (batch size) │ Configurable batch size +│ │ +│ Process per row group: │ +│ 1. Read file footer/metadata │ +│ 2. For each column: │ +│ - Get ColumnDescriptor │ +│ - Read pages via │ +│ PageReadStore │ +│ - Create CometVector │ +│ from native data │ +│ 3. Return ColumnarBatch │ +└───────────┬───────────────────┘ + │ + │ Uses JNI to access native decoders + │ (not for page reading, only for + │ specialized operations if needed) + │ + ↓ +┌───────────────────────────────┐ +│ ColumnarBatch │ +│ with CometVector[] │ Arrow-compatible columnar vectors +│ │ +│ Each CometVector wraps: │ +│ - Arrow ColumnVector │ +│ - Native memory buffers │ +└───────────┬───────────────────┘ + │ + │ Passed to native execution via JNI + │ + ↓ ┌───────────────────────────┐ +┌───────────────────────────────┐ │ ScanExec │ +│ CometExecIterator │──────→│ (operators/scan.rs:54) │ +│ (JNI boundary) │ JNI │ │ +└───────────────────────────────┘ │ Reads batches from JVM │ + │ via CometBatchIterator │ + │ │ + │ Key operations: │ + │ ├─ next_batch() │ + │ ├─ FFI Arrow conversion │ + │ └─ Selection vectors │ + │ (for deletes) │ + └───────────┬───────────────┘ + │ + │ RecordBatch + ↓ + ┌───────────────────────────┐ + │ Native Execution Pipeline │ + │ (DataFusion operators) │ + │ │ + │ - FilterExec │ + │ - ProjectExec │ + │ - AggregateExec │ + │ - etc. │ + └───────────────────────────┘ + +═══════════════════════════════════════════════════════════════════════════════ +STORAGE ACCESS (native_comet) +═══════════════════════════════════════════════════════════════════════════════ + +┌────────────────────────────────┐ +│ Hadoop FileSystem API │ Used by BatchReader +│ │ +│ - LocalFileSystem │ For local files +│ - HDFS │ For HDFS +│ - S3A (Hadoop-AWS module) │ For S3 (using AWS Java SDK) +│ └─ Uses AWS Java SDK │ +│ │ +│ Configuration: │ +│ - fs.s3a.access.key │ +│ - fs.s3a.secret.key │ +│ - fs.s3a.endpoint │ +│ - Standard Hadoop S3A configs │ +└────────────────────────────────┘ + +Key Characteristics: +- ✅ Maximum Spark compatibility (uses Spark's Parquet reading) +- ✅ All Hadoop FileSystem implementations supported (HDFS, S3A, etc.) +- ✅ Standard Hadoop S3A configuration +- ✅ Can use prefetch for improved performance +- ❌ No complex type support (structs, arrays, maps) +- ❌ Cannot read UINT_8/UINT_16 types by default +- ⚠️ Data crosses JVM→Native boundary via FFI +``` + +### `native_datafusion` Scan Architecture + +The `native_datafusion` scan is a fully native implementation using DataFusion's ParquetExec. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_DATAFUSION SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Initial scan created by CometScanRule +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_datafusion"│ +│ │ +│ Note: This is temporary! │ Only exists between CometScanRule +│ CometExecRule converts this │ and CometExecRule optimizer passes +│ to CometNativeScanExec │ +└───────────┬───────────────────┘ + │ + │ CometExecRule transformation + │ (CometExecRule.scala:160-162) + ↓ +┌───────────────────────────────┐ +│ CometNativeScanExec │ Fully native scan operator +│ (CometNativeScanExec.scala:46)│ +│ │ Extends CometLeafExec +│ Fields: │ (native execution root) +│ ├─ nativeOp: Operator │ Protobuf operator definition +│ ├─ relation: HadoopFsRelation │ File relation metadata +│ ├─ output: Seq[Attribute] │ Output schema +│ ├─ requiredSchema: StructType │ Projected schema +│ ├─ dataFilters: Seq[Expr] │ Pushed-down filters +│ └─ serializedPlanOpt │ Serialized native plan +└───────────┬───────────────────┘ + │ + │ Serialization via QueryPlanSerde + │ operator2Proto(scan) + ↓ +┌───────────────────────────────┐ +│ Protobuf Operator │ Serialized scan operator +│ (Operator protobuf message) │ +│ │ Contains: +│ Contains: │ - File paths +│ ├─ file_paths: Vec<String> │ - Schema +│ ├─ schema: Schema │ - Filters (as Expr protobuf) +│ ├─ filters: Vec<Expr> │ - Projection +│ └─ projection: Vec<usize> │ +└───────────┬───────────────────┘ + │ + │ Deserialization in native code + │ + ↓ ┌───────────────────────────┐ + │ PhysicalPlanner │ + │ (planner.rs) │ + │ │ + │ Deserializes protobuf │ + │ Creates physical plan │ + └───────────┬───────────────┘ + │ + │ calls + ↓ + ┌───────────────────────────┐ + │ init_datasource_exec() │ + │ (parquet_exec.rs) │ + │ │ + │ Creates DataFusion │ + │ DataSourceExec with: │ + │ │ + │ ├─ ParquetSource │ + │ ├─ SparkSchemaAdapter │ + │ │ (for Spark/Parquet │ + │ │ type compatibility) │ + │ ├─ Projection │ + │ ├─ Filters │ + │ └─ Partition fields │ + └───────────┬───────────────┘ + │ + ↓ + ┌───────────────────────────┐ + │ DataFusion │ + │ DataSourceExec │ + │ │ + │ ExecutionPlan trait │ + └───────────┬───────────────┘ + │ + │ execute() + ↓ + ┌───────────────────────────┐ + │ ParquetSource │ + │ │ + │ Opens and reads Parquet │ + │ files using arrow-rs │ + │ ParquetRecordBatchReader │ + └───────────┬───────────────┘ + │ + ↓ + ┌────────────────────────────┐ + │ Arrow Parquet Reader │ + │ (from arrow-rs crate) │ + │ │ + │ Operations: │ + │ ├─ Open file │ + │ ├─ Read metadata/foote r │ + │ ├─ Decode pages │ + │ ├─ Build RecordBatches │ + │ └─ Apply filters/projection│ + └───────────┬────────────────┘ + │ + │ RecordBatch stream + ↓ + ┌───────────────────────────┐ + │ Native Execution Pipeline │ + │ (all DataFusion) │ + │ │ + │ - FilterExec │ + │ - ProjectExec │ + │ - AggregateExec │ + │ - etc. │ + └───────────┬───────────────┘ + │ + │ Results via FFI + ↓ +┌───────────────────────────────┐ ┌───────────────────────────┐ +│ Spark execution continues │←──────│ Arrow C Data Interface │ +│ (receives ColumnarBatch) │ FFI │ (Arrow FFI) │ +└───────────────────────────────┘ └───────────────────────────┘ + +═══════════════════════════════════════════════════════════════════════════════ +STORAGE ACCESS (native_datafusion) +═══════════════════════════════════════════════════════════════════════════════ + + ┌───────────────────────────┐ + │ object_store crate │ + │ (objectstore/mod.rs) │ + │ │ + │ Implementations: │ + │ ├─ LocalFileSystem │ + │ ├─ S3 (via aws-sdk-rust) │ + │ │ └─ AWS Rust SDK │ + │ ├─ GCS │ + │ └─ Azure │ + │ │ + │ Configuration translation:│ + │ Hadoop S3A configs → │ + │ object_store configs │ + │ │ + │ - fs.s3a.access.key │ + │ - fs.s3a.secret.key │ + │ - fs.s3a.endpoint │ + │ - fs.s3a.endpoint.region │ + └───────────────────────────┘ + +Key Characteristics: +- ✅ Fully native execution (no JVM→Native data transfer for scan) +- ✅ Complex type support (structs, arrays, maps) +- ✅ Leverages DataFusion community improvements +- ✅ Better performance for some workloads +- ❌ More restrictions than native_comet (no bucketed scans) +- ❌ Cannot read UINT_8/UINT_16 types by default +- ❌ No support for ignoreMissingFiles/ignoreCorruptFiles +- ⚠️ Requires COMET_EXEC_ENABLED=true +- ⚠️ Uses object_store (different from Hadoop FileSystem) +``` + +### `native_iceberg_compat` Scan Architecture + +The `native_iceberg_compat` scan is designed for Iceberg integration, similar to `native_datafusion` but with an Iceberg-specific API layer. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_ICEBERG_COMPAT SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ Auto-selected for Iceberg tables +│ scanImpl = │ by CometScanRule.selectScan() +│ "native_iceberg_compat" │ (CometScanRule.scala:357) +│ │ +│ Key difference: │ +│ - Prefetch is DISABLED │ Iceberg manages its own data access +│ (line 434: !usingDataFusion │ patterns +│ Reader) │ +└───────────┬───────────────────┘ + │ + │ Uses Iceberg-specific reader + ↓ +┌───────────────────────────────┐ +│ IcebergCometBatchReader │ Public API for Iceberg integration +│ (IcebergCometBatchReader.java │ +│ :29-43) │ +│ │ +│ Extends: BatchReader │ Inherits batch reading functionality +│ │ +│ Key method: │ +│ init(AbstractColumnReader[]) │ Iceberg provides column readers +│ │ +│ Purpose: │ +│ - Allows Iceberg to control │ +│ column reader initialization│ +│ - Iceberg passes its own │ +│ AbstractColumnReader[] │ +│ - Comet provides batch reading│ +│ interface │ +└───────────┬───────────────────┘ + │ + │ Iceberg creates AbstractColumnReader[] + │ for each Parquet column + │ + ↓ +┌───────────────────────────────┐ +│ AbstractColumnReader[] │ Iceberg-managed column readers Review Comment: For Iceberg, we will use `IcebergCometNativeBatchReader` which does not pass in column readers. Iceberg integration is not done (and may not be done this way) anyway, so there must be a way that native_iceberg_compat works without Iceberg. ########## docs/source/contributor-guide/parquet_scans.md: ########## @@ -131,7 +131,505 @@ The S3 support of `native_datafusion` has the following limitations: 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](https://github.com/apache/datafusion-comet/issues/1829) for more details. +## Architecture Diagrams + +This section provides detailed architecture diagrams for each of the three Parquet scan implementations. + +### `native_comet` Scan Architecture + +The `native_comet` scan is a hybrid approach where the JVM reads Parquet files and passes data to native execution. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_COMET SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_comet" │ +└───────────┬───────────────────┘ + │ + │ wraps + ↓ +┌───────────────────────────────┐ +│ FileSourceScanExec │ Original Spark scan (from Catalyst) +│ (from Spark) │ +└───────────┬───────────────────┘ + │ + │ creates RDD using + ↓ +┌───────────────────────────────┐ +│ CometParquetFileFormat │ Custom Parquet file format handler +│ (CometParquetFileFormat.scala)│ +│ │ +│ Key methods: │ +│ - buildReaderWithPartition │ +│ Values() │ +│ - supportBatch() │ +└───────────┬───────────────────┘ + │ + │ creates via factory + ↓ +┌───────────────────────────────┐ +│ CometParquetPartitionReader │ Factory for creating partition readers +│ Factory (with prefetch) │ Supports asynchronous prefetch optimization +│ (.../CometParquetPartition │ +│ ReaderFactory.scala) │ +└───────────┬───────────────────┘ + │ + │ creates + ↓ +┌───────────────────────────────┐ +│ BatchReader │ Main Parquet reader +│ (BatchReader.java:90) │ Reads Parquet files via Hadoop APIs +│ │ +│ Key components: │ +│ ├─ ParquetFileReader │ Opens Parquet file, reads metadata +│ ├─ PageReadStore │ Reads column pages from file +│ ├─ ColumnDescriptors │ Parquet column metadata +│ └─ Capacity (batch size) │ Configurable batch size +│ │ +│ Process per row group: │ +│ 1. Read file footer/metadata │ +│ 2. For each column: │ +│ - Get ColumnDescriptor │ +│ - Read pages via │ +│ PageReadStore │ +│ - Create CometVector │ +│ from native data │ +│ 3. Return ColumnarBatch │ +└───────────┬───────────────────┘ + │ + │ Uses JNI to access native decoders + │ (not for page reading, only for + │ specialized operations if needed) Review Comment: Between steps 2-3 is where there is a set of JNI calls to _decode_ individual columns. ########## docs/source/contributor-guide/parquet_scans.md: ########## @@ -131,7 +131,505 @@ The S3 support of `native_datafusion` has the following limitations: 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](https://github.com/apache/datafusion-comet/issues/1829) for more details. +## Architecture Diagrams + +This section provides detailed architecture diagrams for each of the three Parquet scan implementations. + +### `native_comet` Scan Architecture + +The `native_comet` scan is a hybrid approach where the JVM reads Parquet files and passes data to native execution. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_COMET SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_comet" │ +└───────────┬───────────────────┘ + │ + │ wraps + ↓ +┌───────────────────────────────┐ +│ FileSourceScanExec │ Original Spark scan (from Catalyst) +│ (from Spark) │ +└───────────┬───────────────────┘ + │ + │ creates RDD using + ↓ +┌───────────────────────────────┐ +│ CometParquetFileFormat │ Custom Parquet file format handler +│ (CometParquetFileFormat.scala)│ +│ │ +│ Key methods: │ +│ - buildReaderWithPartition │ +│ Values() │ +│ - supportBatch() │ +└───────────┬───────────────────┘ + │ + │ creates via factory + ↓ +┌───────────────────────────────┐ +│ CometParquetPartitionReader │ Factory for creating partition readers +│ Factory (with prefetch) │ Supports asynchronous prefetch optimization +│ (.../CometParquetPartition │ +│ ReaderFactory.scala) │ +└───────────┬───────────────────┘ + │ + │ creates + ↓ +┌───────────────────────────────┐ +│ BatchReader │ Main Parquet reader +│ (BatchReader.java:90) │ Reads Parquet files via Hadoop APIs +│ │ +│ Key components: │ +│ ├─ ParquetFileReader │ Opens Parquet file, reads metadata +│ ├─ PageReadStore │ Reads column pages from file +│ ├─ ColumnDescriptors │ Parquet column metadata +│ └─ Capacity (batch size) │ Configurable batch size +│ │ +│ Process per row group: │ +│ 1. Read file footer/metadata │ +│ 2. For each column: │ +│ - Get ColumnDescriptor │ +│ - Read pages via │ +│ PageReadStore │ +│ - Create CometVector │ +│ from native data │ +│ 3. Return ColumnarBatch │ +└───────────┬───────────────────┘ + │ + │ Uses JNI to access native decoders + │ (not for page reading, only for + │ specialized operations if needed) + │ + ↓ +┌───────────────────────────────┐ +│ ColumnarBatch │ +│ with CometVector[] │ Arrow-compatible columnar vectors +│ │ +│ Each CometVector wraps: │ +│ - Arrow ColumnVector │ +│ - Native memory buffers │ +└───────────┬───────────────────┘ + │ + │ Passed to native execution via JNI + │ + ↓ ┌───────────────────────────┐ +┌───────────────────────────────┐ │ ScanExec │ +│ CometExecIterator │──────→│ (operators/scan.rs:54) │ +│ (JNI boundary) │ JNI │ │ +└───────────────────────────────┘ │ Reads batches from JVM │ + │ via CometBatchIterator │ + │ │ + │ Key operations: │ + │ ├─ next_batch() │ + │ ├─ FFI Arrow conversion │ + │ └─ Selection vectors │ + │ (for deletes) │ + └───────────┬───────────────┘ + │ + │ RecordBatch + ↓ + ┌───────────────────────────┐ + │ Native Execution Pipeline │ + │ (DataFusion operators) │ + │ │ + │ - FilterExec │ + │ - ProjectExec │ + │ - AggregateExec │ + │ - etc. │ + └───────────────────────────┘ + +═══════════════════════════════════════════════════════════════════════════════ +STORAGE ACCESS (native_comet) +═══════════════════════════════════════════════════════════════════════════════ + +┌────────────────────────────────┐ +│ Hadoop FileSystem API │ Used by BatchReader +│ │ +│ - LocalFileSystem │ For local files +│ - HDFS │ For HDFS +│ - S3A (Hadoop-AWS module) │ For S3 (using AWS Java SDK) +│ └─ Uses AWS Java SDK │ +│ │ +│ Configuration: │ +│ - fs.s3a.access.key │ +│ - fs.s3a.secret.key │ +│ - fs.s3a.endpoint │ +│ - Standard Hadoop S3A configs │ +└────────────────────────────────┘ + +Key Characteristics: +- ✅ Maximum Spark compatibility (uses Spark's Parquet reading) +- ✅ All Hadoop FileSystem implementations supported (HDFS, S3A, etc.) +- ✅ Standard Hadoop S3A configuration +- ✅ Can use prefetch for improved performance +- ❌ No complex type support (structs, arrays, maps) +- ❌ Cannot read UINT_8/UINT_16 types by default +- ⚠️ Data crosses JVM→Native boundary via FFI +``` + +### `native_datafusion` Scan Architecture + +The `native_datafusion` scan is a fully native implementation using DataFusion's ParquetExec. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_DATAFUSION SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Initial scan created by CometScanRule +│ (CometScanExec.scala:61) │ +│ │ +│ scanImpl = "native_datafusion"│ +│ │ +│ Note: This is temporary! │ Only exists between CometScanRule +│ CometExecRule converts this │ and CometExecRule optimizer passes +│ to CometNativeScanExec │ +└───────────┬───────────────────┘ + │ + │ CometExecRule transformation + │ (CometExecRule.scala:160-162) + ↓ +┌───────────────────────────────┐ +│ CometNativeScanExec │ Fully native scan operator +│ (CometNativeScanExec.scala:46)│ +│ │ Extends CometLeafExec +│ Fields: │ (native execution root) +│ ├─ nativeOp: Operator │ Protobuf operator definition +│ ├─ relation: HadoopFsRelation │ File relation metadata +│ ├─ output: Seq[Attribute] │ Output schema +│ ├─ requiredSchema: StructType │ Projected schema +│ ├─ dataFilters: Seq[Expr] │ Pushed-down filters +│ └─ serializedPlanOpt │ Serialized native plan +└───────────┬───────────────────┘ + │ + │ Serialization via QueryPlanSerde + │ operator2Proto(scan) + ↓ +┌───────────────────────────────┐ +│ Protobuf Operator │ Serialized scan operator +│ (Operator protobuf message) │ +│ │ Contains: +│ Contains: │ - File paths +│ ├─ file_paths: Vec<String> │ - Schema +│ ├─ schema: Schema │ - Filters (as Expr protobuf) +│ ├─ filters: Vec<Expr> │ - Projection +│ └─ projection: Vec<usize> │ +└───────────┬───────────────────┘ + │ + │ Deserialization in native code + │ + ↓ ┌───────────────────────────┐ + │ PhysicalPlanner │ + │ (planner.rs) │ + │ │ + │ Deserializes protobuf │ + │ Creates physical plan │ + └───────────┬───────────────┘ + │ + │ calls + ↓ + ┌───────────────────────────┐ + │ init_datasource_exec() │ + │ (parquet_exec.rs) │ + │ │ + │ Creates DataFusion │ + │ DataSourceExec with: │ + │ │ + │ ├─ ParquetSource │ + │ ├─ SparkSchemaAdapter │ + │ │ (for Spark/Parquet │ + │ │ type compatibility) │ + │ ├─ Projection │ + │ ├─ Filters │ + │ └─ Partition fields │ + └───────────┬───────────────┘ + │ + ↓ + ┌───────────────────────────┐ + │ DataFusion │ + │ DataSourceExec │ + │ │ + │ ExecutionPlan trait │ + └───────────┬───────────────┘ + │ + │ execute() + ↓ + ┌───────────────────────────┐ + │ ParquetSource │ + │ │ + │ Opens and reads Parquet │ + │ files using arrow-rs │ + │ ParquetRecordBatchReader │ + └───────────┬───────────────┘ + │ + ↓ + ┌────────────────────────────┐ + │ Arrow Parquet Reader │ + │ (from arrow-rs crate) │ + │ │ + │ Operations: │ + │ ├─ Open file │ + │ ├─ Read metadata/foote r │ + │ ├─ Decode pages │ + │ ├─ Build RecordBatches │ + │ └─ Apply filters/projection│ + └───────────┬────────────────┘ + │ + │ RecordBatch stream + ↓ + ┌───────────────────────────┐ + │ Native Execution Pipeline │ + │ (all DataFusion) │ + │ │ + │ - FilterExec │ + │ - ProjectExec │ + │ - AggregateExec │ + │ - etc. │ + └───────────┬───────────────┘ + │ + │ Results via FFI + ↓ +┌───────────────────────────────┐ ┌───────────────────────────┐ +│ Spark execution continues │←──────│ Arrow C Data Interface │ +│ (receives ColumnarBatch) │ FFI │ (Arrow FFI) │ +└───────────────────────────────┘ └───────────────────────────┘ + +═══════════════════════════════════════════════════════════════════════════════ +STORAGE ACCESS (native_datafusion) +═══════════════════════════════════════════════════════════════════════════════ + + ┌───────────────────────────┐ + │ object_store crate │ + │ (objectstore/mod.rs) │ + │ │ + │ Implementations: │ + │ ├─ LocalFileSystem │ + │ ├─ S3 (via aws-sdk-rust) │ + │ │ └─ AWS Rust SDK │ + │ ├─ GCS │ + │ └─ Azure │ + │ │ + │ Configuration translation:│ + │ Hadoop S3A configs → │ + │ object_store configs │ + │ │ + │ - fs.s3a.access.key │ + │ - fs.s3a.secret.key │ + │ - fs.s3a.endpoint │ + │ - fs.s3a.endpoint.region │ + └───────────────────────────┘ + +Key Characteristics: +- ✅ Fully native execution (no JVM→Native data transfer for scan) +- ✅ Complex type support (structs, arrays, maps) +- ✅ Leverages DataFusion community improvements +- ✅ Better performance for some workloads +- ❌ More restrictions than native_comet (no bucketed scans) +- ❌ Cannot read UINT_8/UINT_16 types by default +- ❌ No support for ignoreMissingFiles/ignoreCorruptFiles +- ⚠️ Requires COMET_EXEC_ENABLED=true +- ⚠️ Uses object_store (different from Hadoop FileSystem) +``` + +### `native_iceberg_compat` Scan Architecture + +The `native_iceberg_compat` scan is designed for Iceberg integration, similar to `native_datafusion` but with an Iceberg-specific API layer. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NATIVE_ICEBERG_COMPAT SCAN FLOW │ +└─────────────────────────────────────────────────────────────────────────────┘ + +JVM (Scala/Java) Native (Rust) +════════════════════════════════════════════════════════════════════════════════ + +┌───────────────────────────────┐ +│ CometScanExec │ Physical operator for DataSource V1 +│ (CometScanExec.scala:61) │ +│ │ Auto-selected for Iceberg tables +│ scanImpl = │ by CometScanRule.selectScan() +│ "native_iceberg_compat" │ (CometScanRule.scala:357) +│ │ +│ Key difference: │ +│ - Prefetch is DISABLED │ Iceberg manages its own data access +│ (line 434: !usingDataFusion │ patterns +│ Reader) │ +└───────────┬───────────────────┘ + │ + │ Uses Iceberg-specific reader + ↓ +┌───────────────────────────────┐ +│ IcebergCometBatchReader │ Public API for Iceberg integration +│ (IcebergCometBatchReader.java │ +│ :29-43) │ +│ │ +│ Extends: BatchReader │ Inherits batch reading functionality +│ │ +│ Key method: │ +│ init(AbstractColumnReader[]) │ Iceberg provides column readers +│ │ +│ Purpose: │ Review Comment: This is not correct. (This is how the current integration of native_comet and Iceberg works). `native_iceberg_compat` uses the same `init_datasource_exec` call to create a DataFusion `DataSourceExec` and wraps the natve batch and native columns in corresponding classes on the JVM side (NativeBatchReader and NativeColumnReader) -- 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]
