You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "mbutrovich (via GitHub)" <gi...@apache.org> on 2025/10/06 02:37:24 UTC
[PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
mbutrovich opened a new pull request, #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
I was inspired by [Russell's recent talk](https://www.youtube.com/watch?v=kJaD0WuQ1Bg) and wanted to visit the abstraction layer at which Comet integrates with Iceberg. We have the `iceberg_compat` codepath for Iceberg integration, but this requires code changes in Iceberg to integrate with partition reader factories. Instead, this prototype works at the `FileScanTask` layer after planning. This prototype starts us toward a path of fully-native Iceberg scans to match our Parquet logic with `native_datafusion` scans without any changes in Iceberg code.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
- New `CometIcebergNativeScanExec` node on the Scala side.
- Use reflection to extract scan properties, mostly `FileScanTask`s and serialize to native code.
- New `IcebergScanExec` on native side that uses `FileScanTask`s to perform reads in `iceberg-rust`.
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
New `CometIcebergNativeSuite`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543838330
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -196,6 +196,19 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
val nativeOp = operator2Proto(scan).get
CometNativeScan.createExec(nativeOp, scan)
+ // Fully native Iceberg scan for V2
+ // Config checks (COMET_ICEBERG_NATIVE_ENABLED, COMET_EXEC_ENABLED) are done in CometScanRule
+ case scan: CometBatchScanExec
Review Comment:
Any possibility that iceberg_compat falls into this case?
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -807,6 +820,12 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
CometNativeScan.convert(scan, builder, childOp: _*)
+ // Fully native Iceberg scan for V2
+ case scan: CometBatchScanExec
Review Comment:
Same here
##########
spark/pom.xml:
##########
@@ -173,8 +173,52 @@ under the License.
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
</dependency>
+ <!-- Iceberg dependencies for testing native Iceberg scan -->
+ <!-- Note: The specific Iceberg artifact is defined in profiles below based on Spark version -->
</dependencies>
+ <profiles>
+ <!-- Iceberg dependencies vary by Spark version (Iceberg 1.8.1 for Spark 3.x, 1.10.0 for Spark 4.0) -->
+ <profile>
+ <id>spark-3.4</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-3.5</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-4.0</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.10.0</version>
Review Comment:
Are these are right combinations?
For ex 1.9 is not there
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3439036311
Finally got all 3000+ Iceberg Java tests to run, here is my progress today:
<img width="681" height="126" alt="Screenshot 2025-10-23 at 10 47 19 AM" src="https://github.com/user-attachments/assets/21ecff43-0603-402f-8795-4d42814d5eec" />
Fall back when FileScanTasks are for ORC or Avro...
<img width="682" height="123" alt="Screenshot 2025-10-23 at 12 16 12 PM" src="https://github.com/user-attachments/assets/55acaf28-dd35-4b6e-b214-3146978e11c4" />
Fall back when Iceberg table spec is V3 (some of these tests actually worked natively, but in theory we shouldn't support it yet)...
<img width="693" height="116" alt="Screenshot 2025-10-23 at 1 42 25 PM" src="https://github.com/user-attachments/assets/3458bd01-c182-4a86-b64f-9d2e05304c18" />
Fix explain on `CometIcebergNativeScanExec` to show pushed down filters...
<img width="669" height="121" alt="Screenshot 2025-10-23 at 3 25 40 PM" src="https://github.com/user-attachments/assets/667289b6-5913-4a62-87d8-62cb591be8ac" />
It's getting there!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3432211177
Chipping away at Iceberg tests, running via:
```
ENABLE_COMET=true ./gradlew -DsparkVersions=3.5 -DscalaVersion=2.13 -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-3.5_2.13:test \
-Pquick=true -x javadoc
```
Yesterday:
<img width="693" height="167" alt="Screenshot 2025-10-21 at 5 32 43 PM" src="https://github.com/user-attachments/assets/ff772924-f7c9-4109-873f-1b2d3f41d799" />
Today:
<img width="692" height="129" alt="Screenshot 2025-10-22 at 8 49 10 AM" src="https://github.com/user-attachments/assets/d266a3e4-cf51-4cb0-9bb4-ae9f4c0ad0f9" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3434277459
Today's progress:
<img width="681" height="113" alt="Screenshot 2025-10-22 at 12 17 19 PM" src="https://github.com/user-attachments/assets/eb74af26-ac89-4058-876d-7cf326c95a90" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3455857882
Sure I added fallbacks for 2 scenarios that iceberg-rust doesn't support yet (pushdown filters with complex types, and a few unsupported hive partitioning data types), but still...
<img width="673" height="117" alt="Screenshot 2025-10-28 at 6 52 20 AM" src="https://github.com/user-attachments/assets/1931728d-b7e2-4df4-8f84-2ce4ee4a9c30" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2534516919
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -0,0 +1,477 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Native Iceberg table scan operator using iceberg-rust
+
+use std::any::Any;
+use std::collections::{HashMap, VecDeque};
+use std::fmt;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use arrow::array::RecordBatch;
+use arrow::datatypes::SchemaRef;
+use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
+use datafusion::physical_expr::EquivalenceProperties;
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet,
+};
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
+};
+use futures::future::BoxFuture;
+use futures::{ready, FutureExt, Stream, StreamExt, TryStreamExt};
+use iceberg::io::FileIO;
+
+use crate::execution::operators::ExecutionError;
+use crate::parquet::parquet_support::SparkParquetOptions;
+use crate::parquet::schema_adapter::SparkSchemaAdapterFactory;
+use datafusion::datasource::schema_adapter::SchemaAdapterFactory;
+use datafusion_comet_spark_expr::EvalMode;
+use datafusion_datasource::file_stream::FileStreamMetrics;
+
+/// Iceberg table scan operator that uses iceberg-rust to read Iceberg tables.
+///
+/// Executes pre-planned FileScanTasks for efficient parallel scanning.
+#[derive(Debug)]
+pub struct IcebergScanExec {
+ /// Iceberg table metadata location for FileIO initialization
+ metadata_location: String,
+ /// Output schema after projection
+ output_schema: SchemaRef,
+ /// Cached execution plan properties
+ plan_properties: PlanProperties,
+ /// Catalog-specific configuration for FileIO
+ catalog_properties: HashMap<String, String>,
+ /// Pre-planned file scan tasks, grouped by partition
+ file_task_groups: Vec<Vec<iceberg::scan::FileScanTask>>,
+ /// Metrics
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl IcebergScanExec {
+ pub fn new(
+ metadata_location: String,
+ schema: SchemaRef,
+ catalog_properties: HashMap<String, String>,
+ file_task_groups: Vec<Vec<iceberg::scan::FileScanTask>>,
+ ) -> Result<Self, ExecutionError> {
+ let output_schema = schema;
+ let num_partitions = file_task_groups.len();
+ let plan_properties = Self::compute_properties(Arc::clone(&output_schema), num_partitions);
+
+ let metrics = ExecutionPlanMetricsSet::new();
+
+ Ok(Self {
+ metadata_location,
+ output_schema,
+ plan_properties,
+ catalog_properties,
+ file_task_groups,
+ metrics,
+ })
+ }
+
+ fn compute_properties(schema: SchemaRef, num_partitions: usize) -> PlanProperties {
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(num_partitions),
+ EmissionType::Incremental,
+ Boundedness::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergScanExec {
+ fn name(&self) -> &str {
+ "IcebergScanExec"
+ }
+
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.output_schema)
+ }
+
+ fn properties(&self) -> &PlanProperties {
+ &self.plan_properties
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ if partition < self.file_task_groups.len() {
+ let tasks = &self.file_task_groups[partition];
+ self.execute_with_tasks(tasks.clone(), partition, context)
+ } else {
+ Err(DataFusionError::Execution(format!(
+ "IcebergScanExec: Partition index {} out of range (only {} task groups available)",
+ partition,
+ self.file_task_groups.len()
+ )))
+ }
+ }
+
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+}
+
+impl IcebergScanExec {
+ /// Handles MOR (Merge-On-Read) tables by automatically applying positional and equality
+ /// deletes via iceberg-rust's ArrowReader.
+ fn execute_with_tasks(
+ &self,
+ tasks: Vec<iceberg::scan::FileScanTask>,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ let output_schema = Arc::clone(&self.output_schema);
+ let file_io = Self::load_file_io(&self.catalog_properties, &self.metadata_location)?;
+ let batch_size = context.session_config().batch_size();
+
+ let metrics = IcebergScanMetrics::new(&self.metrics, partition);
+
+ // Create parallel file stream that overlaps opening next file with reading current file
+ let file_stream = IcebergFileStream::new(
+ tasks,
+ file_io,
+ batch_size,
+ Arc::clone(&output_schema),
+ metrics,
+ )?;
+
+ // Note: BatchSplitStream adds overhead. Since we're already setting batch_size in
+ // iceberg-rust's ArrowReaderBuilder, it should produce correctly sized batches.
+ // Only use BatchSplitStream as a safety net if needed.
+ // For now, return the file_stream directly to reduce stream nesting overhead.
+
+ Ok(Box::pin(file_stream))
+ }
+
+ fn load_file_io(
+ catalog_properties: &HashMap<String, String>,
+ metadata_location: &str,
+ ) -> Result<FileIO, DataFusionError> {
+ let mut file_io_builder = FileIO::from_path(metadata_location)
+ .map_err(|e| DataFusionError::Execution(format!("Failed to create FileIO: {}", e)))?;
+
+ for (key, value) in catalog_properties {
+ file_io_builder = file_io_builder.with_prop(key, value);
+ }
+
+ file_io_builder
+ .build()
+ .map_err(|e| DataFusionError::Execution(format!("Failed to build FileIO: {}", e)))
+ }
+}
+
+/// Metrics for IcebergScanExec
+struct IcebergScanMetrics {
+ /// Baseline metrics (output rows, elapsed compute time)
+ baseline: BaselineMetrics,
+ /// File stream metrics (time opening, time scanning, etc.)
+ file_stream: FileStreamMetrics,
+ /// Count of file splits (FileScanTasks) processed
+ num_splits: Count,
+}
+
+impl IcebergScanMetrics {
+ fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self {
+ baseline: BaselineMetrics::new(metrics, partition),
+ file_stream: FileStreamMetrics::new(metrics, partition),
+ num_splits: MetricBuilder::new(metrics).counter("num_splits", partition),
+ }
+ }
+}
+
+/// State machine for IcebergFileStream
+enum FileStreamState {
+ /// Idle state - need to start opening next file
+ Idle,
+ /// Opening a file
+ Opening {
+ future: BoxFuture<'static, DFResult<SendableRecordBatchStream>>,
+ },
+ /// Reading from current file while potentially opening next file
+ Reading {
+ current: SendableRecordBatchStream,
+ next: Option<BoxFuture<'static, DFResult<SendableRecordBatchStream>>>,
+ },
+ /// Error state
+ Error,
+}
+
+/// Stream that reads Iceberg files with parallel opening optimization.
+/// Opens the next file while reading the current file to overlap IO with compute.
+///
+/// Inspired by DataFusion's [`FileStream`] pattern for overlapping file opening with reading.
+///
+/// [`FileStream`]: https://github.com/apache/datafusion/blob/main/datafusion/datasource/src/file_stream.rs
+struct IcebergFileStream {
+ schema: SchemaRef,
+ file_io: FileIO,
+ batch_size: usize,
+ tasks: VecDeque<iceberg::scan::FileScanTask>,
+ state: FileStreamState,
+ metrics: IcebergScanMetrics,
+}
+
+impl IcebergFileStream {
+ fn new(
+ tasks: Vec<iceberg::scan::FileScanTask>,
+ file_io: FileIO,
+ batch_size: usize,
+ schema: SchemaRef,
+ metrics: IcebergScanMetrics,
+ ) -> DFResult<Self> {
+ Ok(Self {
+ schema,
+ file_io,
+ batch_size,
+ tasks: tasks.into_iter().collect(),
+ state: FileStreamState::Idle,
+ metrics,
+ })
+ }
+
+ fn start_next_file(
+ &mut self,
+ ) -> Option<BoxFuture<'static, DFResult<SendableRecordBatchStream>>> {
+ let task = self.tasks.pop_front()?;
+
+ self.metrics.num_splits.add(1);
+
+ let file_io = self.file_io.clone();
+ let batch_size = self.batch_size;
+ let schema = Arc::clone(&self.schema);
+
+ Some(Box::pin(async move {
+ let task_stream = futures::stream::iter(vec![Ok(task)]).boxed();
+
+ let reader = iceberg::arrow::ArrowReaderBuilder::new(file_io)
+ .with_batch_size(batch_size)
+ .with_row_selection_enabled(true)
+ .build();
+
+ let stream = reader.read(task_stream).map_err(|e| {
+ DataFusionError::Execution(format!("Failed to read Iceberg task: {}", e))
+ })?;
+
+ let target_schema = Arc::clone(&schema);
+
+ // Schema adaptation handles differences in Arrow field names and metadata
+ // between the file schema and expected output schema
+ let mapped_stream = stream
+ .map_err(|e| DataFusionError::Execution(format!("Iceberg scan error: {}", e)))
+ .and_then(move |batch| {
+ let spark_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
Review Comment:
Do you have tests (perhaps via Iceberg suite) that test for non-UTC timezone timestamps?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2544048337
##########
native/Cargo.toml:
##########
@@ -34,7 +34,7 @@ edition = "2021"
rust-version = "1.86"
[workspace.dependencies]
-arrow = { version = "56.0.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "56.2.0", features = ["prettyprint", "ffi", "chrono-tz"] }
Review Comment:
Yeah I mean we should have updated this when parquet was updated. Anyway...
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3524123600
Added the 1.10.0.diff from #2709 and after a day of hacking:
Spark tests:
<img width="685" height="190" alt="Screenshot 2025-11-12 at 5 09 28 PM" src="https://github.com/user-attachments/assets/f5642e95-682e-4c94-b211-c2f6e000f8bf" />
Spark extensions tests:
<img width="679" height="189" alt="Screenshot 2025-11-12 at 5 09 37 PM" src="https://github.com/user-attachments/assets/131cb8ca-2600-4b73-a0f3-ef1366102b8b" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2534535280
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import scala.util.Random
+
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
+import org.apache.spark.sql.types._
+
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions}
+
+class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
+
+ test("select *") {
+ val sql = s"SELECT * FROM $icebergTableName"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("select * with limit") {
+ val sql = s"SELECT * FROM $icebergTableName LIMIT 500"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by single column") {
+ val df = spark.table(icebergTableName)
+ for (col <- df.columns) {
+ val sql = s"SELECT $col FROM $icebergTableName ORDER BY $col"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("order by multiple columns") {
+ val df = spark.table(icebergTableName)
+ val allCols = df.columns.mkString(",")
+ val sql = s"SELECT $allCols FROM $icebergTableName ORDER BY $allCols"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by random columns") {
+ val df = spark.table(icebergTableName)
+
+ for (_ <- 1 to 10) {
+ // We only do order by permutations of primitive types to exercise native shuffle's
+ // RangePartitioning which only supports those types.
+ // This used to be df.columns.slice(0, 14) but we're removing two binary columns in
+ // CometFuzzIcebergBase.beforeAll
+ val shuffledPrimitiveCols = Random.shuffle(df.columns.slice(0, 12).toList)
+ val randomSize = Random.nextInt(shuffledPrimitiveCols.length) + 1
+ val randomColsSubset = shuffledPrimitiveCols.take(randomSize).toArray.mkString(",")
+ val sql = s"SELECT $randomColsSubset FROM $icebergTableName ORDER BY $randomColsSubset"
+ checkSparkAnswerAndOperator(sql)
+ }
+ }
+
+ test("distribute by single column (complex types)") {
+ val df = spark.table(icebergTableName)
+ val columns = df.schema.fields.filter(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- columns) {
+ // DISTRIBUTE BY is equivalent to df.repartition($col) and uses
+ val sql = s"SELECT $col FROM $icebergTableName DISTRIBUTE BY $col"
+ val resultDf = spark.sql(sql)
+ resultDf.collect()
+ // check for Comet shuffle
+ val plan =
+ resultDf.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan
+ val cometShuffleExchanges = collectCometShuffleExchanges(plan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffleExchanges.length == 1)
+ }
+ }
+
+ test("shuffle supports all types") {
+ val df = spark.table(icebergTableName)
+ val df2 = df.repartition(8, df.col("c0")).sort("c1")
+ df2.collect()
+ val cometShuffles = collectCometShuffleExchanges(df2.queryExecution.executedPlan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffles.length == 2)
+ }
+
+ test("join") {
+ val df = spark.table(icebergTableName)
+ df.createOrReplaceTempView("t1")
+ df.createOrReplaceTempView("t2")
+ // Filter out complex types - iceberg-rust can't create predicates for struct/array/map equality
+ val primitiveColumns = df.schema.fields.filterNot(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- primitiveColumns) {
+ // cannot run fully native due to HashAggregate
+ val sql = s"SELECT count(*) FROM t1 JOIN t2 ON t1.$col = t2.$col"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(2 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("decode") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // BinaryType, since then this test would not run any queries and silently pass.
+ var testedBinary = false
+ for (field <- df.schema.fields if field.dataType == BinaryType) {
+ testedBinary = true
+ // Intentionally use odd capitalization of 'utf-8' to test normalization.
+ val sql = s"SELECT decode(${field.name}, 'utF-8') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedBinary)
+ }
+
+ test("regexp_replace") {
+ withSQLConf(CometConf.COMET_REGEXP_ALLOW_INCOMPATIBLE.key -> "true") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // StringType, since then this test would not run any queries and silently pass.
+ var testedString = false
+ for (field <- df.schema.fields if field.dataType == StringType) {
+ testedString = true
+ val sql = s"SELECT regexp_replace(${field.name}, 'a', 'b') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedString)
+ }
+ }
+
+ test("Iceberg temporal types written as INT96") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.INT96)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MICROS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MICROS)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MILLIS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MILLIS)
+ }
+
+ private def testIcebergTemporalTypes(
+ outputTimestampType: ParquetOutputTimestampType.Value,
+ generateArray: Boolean = true,
+ generateStruct: Boolean = true): Unit = {
+
+ val schema = FuzzDataGenerator.generateSchema(
+ SchemaGenOptions(
+ generateArray = generateArray,
+ generateStruct = generateStruct,
+ primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot { dataType =>
+ // Disable decimals - iceberg-rust doesn't support FIXED_LEN_BYTE_ARRAY in page index yet
+ dataType.isInstanceOf[DecimalType]
+ }))
+
+ val options =
+ DataGenOptions(generateNegativeZero = false)
+
+ withTempPath { filename =>
+ val random = new Random(42)
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString,
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> defaultTimezone) {
+ ParquetGenerator.makeParquetFile(random, spark, filename.toString, schema, 100, options)
+ }
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
+ Seq(true, false).foreach { inferTimestampNtzEnabled =>
+ Seq(true, false).foreach { int96TimestampConversion =>
+ Seq(true, false).foreach { int96AsTimestamp =>
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz,
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key -> int96AsTimestamp.toString,
+ SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION.key -> int96TimestampConversion.toString,
+ SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key -> inferTimestampNtzEnabled.toString) {
+
+ val df = spark.table(icebergTableName)
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
+ def hasTemporalType(t: DataType): Boolean = t match {
+ case DataTypes.DateType | DataTypes.TimestampType |
+ DataTypes.TimestampNTZType =>
+ true
+ case t: StructType => t.exists(f => hasTemporalType(f.dataType))
+ case t: ArrayType => hasTemporalType(t.elementType)
Review Comment:
Should this also test for `MapType`?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543864046
##########
.github/workflows/iceberg_spark_test.yml:
##########
@@ -156,3 +156,120 @@ jobs:
ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ matrix.spark-version.short }} -DscalaVersion=${{ matrix.scala-version }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-runtime-${{ matrix.spark-version.short }}_${{ matrix.scala-version }}:integrationTest \
-Pquick=true -x javadoc
+
+ iceberg-spark-rust:
+ if: contains(github.event.pull_request.title, '[iceberg]')
+ strategy:
+ matrix:
+ os: [ubuntu-24.04]
+ java-version: [11, 17]
+ iceberg-version: [{short: '1.8', full: '1.8.1'}, {short: '1.9', full: '1.9.1'}, {short: '1.10', full: '1.10.0'}]
+ spark-version: [{short: '3.5', full: '3.5.7'}]
Review Comment:
We don't currently test it, so it seems like a big can of worms to open. I can add a workflow and see how it goes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "jordepic (via GitHub)" <gi...@apache.org>.
jordepic commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-4158430803
Yep - I was taking a look. It's not a SQL operator which is a problem, we need write support, etc etc etc.
I've been playing around a bit with starrocks table maintenance since in practice it's just so much faster than the Spark ones. Unfortunately, the API is a bit limited. I may experiment with just using a spark job that does JNI to native code in C++ or rust in the meantime, but thank you for the ticket! I can follow up there :)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3482439350
CI runs these already for PRs tagged `[iceberg]` so people can see that it's green there, but for posterity here is Iceberg's Spark tests...
<img width="685" height="195" alt="Screenshot 2025-11-03 at 9 09 52 AM" src="https://github.com/user-attachments/assets/814ba49f-95e6-4387-a33e-a413a53bfa22" />
and Spark extensions tests...
<img width="677" height="184" alt="Screenshot 2025-11-03 at 9 09 43 AM" src="https://github.com/user-attachments/assets/b797319c-3b61-479e-9a38-d09058782df3" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543702690
##########
.github/workflows/iceberg_spark_test.yml:
##########
@@ -156,3 +156,120 @@ jobs:
ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ matrix.spark-version.short }} -DscalaVersion=${{ matrix.scala-version }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-runtime-${{ matrix.spark-version.short }}_${{ matrix.scala-version }}:integrationTest \
-Pquick=true -x javadoc
+
+ iceberg-spark-rust:
+ if: contains(github.event.pull_request.title, '[iceberg]')
+ strategy:
+ matrix:
+ os: [ubuntu-24.04]
+ java-version: [11, 17]
+ iceberg-version: [{short: '1.8', full: '1.8.1'}, {short: '1.9', full: '1.9.1'}, {short: '1.10', full: '1.10.0'}]
+ spark-version: [{short: '3.5', full: '3.5.7'}]
Review Comment:
I feel we need a test for Spark 3.4
maybe we can skip for Iceberg 1.8 instead
##########
native/Cargo.toml:
##########
@@ -34,7 +34,7 @@ edition = "2021"
rust-version = "1.86"
[workspace.dependencies]
-arrow = { version = "56.0.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "56.2.0", features = ["prettyprint", "ffi", "chrono-tz"] }
Review Comment:
Is this a bug? I thought arrow ver == parquet ver always
##########
native/core/Cargo.toml:
##########
@@ -60,6 +60,7 @@ tempfile = "3.8.0"
itertools = "0.14.0"
paste = "1.0.14"
datafusion = { workspace = true, features = ["parquet_encryption"] }
+datafusion-datasource = { version = "50.3.0" }
Review Comment:
Is pinning the version necessary? may be missed for upgrading in the future....
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543874537
##########
spark/pom.xml:
##########
@@ -173,8 +173,52 @@ under the License.
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
</dependency>
+ <!-- Iceberg dependencies for testing native Iceberg scan -->
+ <!-- Note: The specific Iceberg artifact is defined in profiles below based on Spark version -->
</dependencies>
+ <profiles>
+ <!-- Iceberg dependencies vary by Spark version (Iceberg 1.8.1 for Spark 3.x, 1.10.0 for Spark 4.0) -->
+ <profile>
+ <id>spark-3.4</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-3.5</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-4.0</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.10.0</version>
Review Comment:
We could run 1.8.1. with 3.4, 1.9.1 with 3.5, and 1.10.0 with 4.0. It's not testing the full matrix, but you'd touch every version.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3746773873
> I'm having some issues when using a RestCatalog -
>
> ```
> 26[/01/13](https://jupyterhub-us/01/13) 15:00:10 ERROR IcebergReflection: Iceberg reflection failure: Failed to get table metadata: class org.apache.comet.iceberg.IcebergReflection$ cannot access a member of class org.apache.iceberg.rest.RESTTableOperations with modifiers "public"
> ```
>
> Using comet .12, iceberg 1.10.0, spark 3.5.6 with scala 2.13. Did you only test this for the hadoop catalog? Or did you try other types as well?
@hsiang-c tested this a REST catalog and in theory we have a test that exercises this as well after https://github.com/apache/datafusion-comet/pull/2895. I'm wondering if the jars aren't all getting loaded when used with Jupyter notebooks? I'm not as familiar with this scenario. Would you mind opening a new issue and we can track discussion there?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "hsiang-c (via GitHub)" <gi...@apache.org>.
hsiang-c commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3746924899
Yes, we need the upcoming release for REST Catalog support or [build](https://github.com/apache/datafusion-comet/tree/main/dev/release#build-the-jars) the JAR by yourself.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2547444993
##########
.github/workflows/iceberg_spark_test.yml:
##########
@@ -156,3 +156,120 @@ jobs:
ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ matrix.spark-version.short }} -DscalaVersion=${{ matrix.scala-version }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-runtime-${{ matrix.spark-version.short }}_${{ matrix.scala-version }}:integrationTest \
-Pquick=true -x javadoc
+
+ iceberg-spark-rust:
+ if: contains(github.event.pull_request.title, '[iceberg]')
+ strategy:
+ matrix:
+ os: [ubuntu-24.04]
+ java-version: [11, 17]
+ iceberg-version: [{short: '1.8', full: '1.8.1'}, {short: '1.9', full: '1.9.1'}, {short: '1.10', full: '1.10.0'}]
+ spark-version: [{short: '3.4', full: '3.4.3'}, {short: '3.5', full: '3.5.7'}]
Review Comment:
Here's what we currently test with this PR:
| | 3.4 | 3.5 | 4.0 |
|-------|-------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------|
| 1.5.2 | CometIcebergNativeSuite CometFuzzIcebergSuite IcebergReadFromS3Suite (not run in CI due to MinIO container) | | |
| 1.8.1 | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests CometIcebergNativeSuite CometFuzzIcebergSuite IcebergReadFromS3Suite (not run in CI due to MinIO container) | |
| 1.9.1 | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests | |
| 1.10 | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests | Iceberg Spark Tests Iceberg Spark Extensions Tests Iceberg Spark Runtime Tests | CometIcebergNativeSuite CometFuzzIcebergSuite IcebergReadFromS3Suite (not run in CI due to MinIO container) |
I leaned on newer versions for the Iceberg tests because as best as I could tell, never versions are a superset of the older versions. For the Comet-native tests we are running 1.5.2.
We should have a discussion of what we want to run long term, because right now tagging a PR `[iceberg]` makes CI take hours and causes so many parallel Iceberg suites that we start getting network timeouts (likely due to throttling).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2547379474
##########
.github/workflows/iceberg_spark_test.yml:
##########
@@ -156,3 +156,120 @@ jobs:
ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ matrix.spark-version.short }} -DscalaVersion=${{ matrix.scala-version }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-runtime-${{ matrix.spark-version.short }}_${{ matrix.scala-version }}:integrationTest \
-Pquick=true -x javadoc
+
+ iceberg-spark-rust:
+ if: contains(github.event.pull_request.title, '[iceberg]')
+ strategy:
+ matrix:
+ os: [ubuntu-24.04]
+ java-version: [11, 17]
+ iceberg-version: [{short: '1.8', full: '1.8.1'}, {short: '1.9', full: '1.9.1'}, {short: '1.10', full: '1.10.0'}]
+ spark-version: [{short: '3.4', full: '3.4.3'}, {short: '3.5', full: '3.5.7'}]
Review Comment:
The profile now says to use Iceberg 1.5 with Spark 3.4, but we do not have 1.5 here. Not sure if it causes problems...
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3451653977
I had to turn off `countDeletes` in the `TestSparkReaderDeletes` suite because iceberg-rust (rightly) merges the equality deletes with table filters to evaluate them together in Arrow-rs's Parquet reader. This makes filtered rows and deleted rows indistinguishable, so the counts won't match. We still get correctness checks after skipped the counts though, so I'm confident in the tests. They still assert that:
- Deletes are correctly applied (rows filtered)
- The _deleted metadata column works
- Equality and positional deletes function properly
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3451633537
This morning's progress:
<img width="671" height="140" alt="Screenshot 2025-10-27 at 10 39 41 AM" src="https://github.com/user-attachments/assets/432eb498-9ff9-4ac8-9d1e-68c3c7739443" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543852662
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -196,6 +196,19 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
val nativeOp = operator2Proto(scan).get
CometNativeScan.createExec(nativeOp, scan)
+ // Fully native Iceberg scan for V2
+ // Config checks (COMET_ICEBERG_NATIVE_ENABLED, COMET_EXEC_ENABLED) are done in CometScanRule
+ case scan: CometBatchScanExec
Review Comment:
Yep, just fixed this after noticing a new CI failure after enabling both Iceberg code paths.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -807,6 +820,12 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
CometNativeScan.convert(scan, builder, childOp: _*)
+ // Fully native Iceberg scan for V2
+ case scan: CometBatchScanExec
Review Comment:
Yep, just fixed this after noticing a new CI failure after enabling both Iceberg code paths.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543870285
##########
native/Cargo.toml:
##########
@@ -34,7 +34,7 @@ edition = "2021"
rust-version = "1.86"
[workspace.dependencies]
-arrow = { version = "56.0.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "56.2.0", features = ["prettyprint", "ffi", "chrono-tz"] }
Review Comment:
We're already on Parquet 56.2.0:
https://github.com/apache/datafusion-comet/blob/80bef433ff6e4a60bbc3e30450638d7db45ec17f/native/Cargo.toml#L40
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2543861929
##########
spark/pom.xml:
##########
@@ -173,8 +173,52 @@ under the License.
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
</dependency>
+ <!-- Iceberg dependencies for testing native Iceberg scan -->
+ <!-- Note: The specific Iceberg artifact is defined in profiles below based on Spark version -->
</dependencies>
+ <profiles>
+ <!-- Iceberg dependencies vary by Spark version (Iceberg 1.8.1 for Spark 3.x, 1.10.0 for Spark 4.0) -->
+ <profile>
+ <id>spark-3.4</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-3.5</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <profile>
+ <id>spark-4.0</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>1.10.0</version>
Review Comment:
This is just the Iceberg version to use for `CometFuzzIcebergSuite`, `IcebergReadFromS3Suite`, and `CometIcebergNativeSuite`. We could discuss how many versions to test, but I don't expect major behavior differences from 1.9 if we're covering 1.8.1 and 1.10.0. We should discuss how large we want the Iceberg test matrix to get.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "jordepic (via GitHub)" <gi...@apache.org>.
jordepic commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3746559978
Hey Matt, thanks for the commit! This is super impactful :)
I'm having some issues when using a RestCatalog -
```
26[/01/13](https://jupyterhub-us/01/13) 15:00:10 ERROR IcebergReflection: Iceberg reflection failure: Failed to get table metadata: class org.apache.comet.iceberg.IcebergReflection$ cannot access a member of class org.apache.iceberg.rest.RESTTableOperations with modifiers "public"
```
Using comet .12, iceberg 1.10.0, spark 3.5.6 with scala 2.13. Did you only test this for the hadoop catalog? Or did you try other types as well?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "jordepic (via GitHub)" <gi...@apache.org>.
jordepic commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-4158223641
Hey @mbutrovich ! Following up here one more time.
One area where I see the potential for a ton of impact in comet is performing iceberg table maintenance procedures. Existing spark-based readers have to convert columnar data back to row format to combined multiple data files, which in practice looks to use tons of resources.
I'm curious whether you have any plans to take a look at this aspect of things. If not, I may do so myself!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-4158410998
> Hey @mbutrovich ! Following up here one more time.
>
> One area where I see the potential for a ton of impact in comet is performing iceberg table maintenance procedures. Existing spark-based readers have to convert columnar data back to row format to combined multiple data files, which in practice looks to use tons of resources.
>
> I'm curious whether you have any plans to take a look at this aspect of things. If not, I may do so myself!
Thanks for following up! There are discussion issues here where we can chat more:
https://github.com/apache/datafusion-comet/issues/3371
The tl;dr is there's a lot of upstream work to be done in iceberg-rust's write path before we can do table maintenance in Comet. I think it's a great fit for Comet some day, but I have higher priority stuff to continue to add to the read path first.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich merged PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3369756033
## [Codecov](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0.89686%` with `442 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 10.15%. Comparing base ([`f09f8af`](https://app.codecov.io/gh/apache/datafusion-comet/commit/f09f8af64c6599255e116a376f4f008f2fd63b43?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`4afec43`](https://app.codecov.io/gh/apache/datafusion-comet/commit/4afec43b0db958fcda15fc28e2fa91918fb1451d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 574 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../scala/org/apache/comet/serde/QueryPlanSerde.scala](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fcomet%2Fserde%2FQueryPlanSerde.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9jb21ldC9zZXJkZS9RdWVyeVBsYW5TZXJkZS5zY2FsYQ==) | 0.00% | [243 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/spark/sql/comet/CometIcebergNativeScanExec.scala](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fcomet%2FCometIcebergNativeScanExec.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9zcGFyay9zcWwvY29tZXQvQ29tZXRJY2ViZXJnTmF0aXZlU2NhbkV4ZWMuc2NhbGE=) | 0.00% | [159 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...n/scala/org/apache/comet/rules/CometScanRule.scala](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fcomet%2Frules%2FCometScanRule.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9jb21ldC9ydWxlcy9Db21ldFNjYW5SdWxlLnNjYWxh) | 0.00% | [34 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...n/scala/org/apache/spark/sql/comet/operators.scala](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fcomet%2Foperators.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9zcGFyay9zcWwvY29tZXQvb3BlcmF0b3JzLnNjYWxh) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...n/scala/org/apache/comet/rules/CometExecRule.scala](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fcomet%2Frules%2FCometExecRule.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9jb21ldC9ydWxlcy9Db21ldEV4ZWNSdWxlLnNjYWxh) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #2528 +/- ##
=============================================
- Coverage 56.12% 10.15% -45.98%
+ Complexity 976 232 -744
=============================================
Files 119 147 +28
Lines 11743 13990 +2247
Branches 2251 2399 +148
=============================================
- Hits 6591 1420 -5171
- Misses 4012 12393 +8381
+ Partials 1140 177 -963
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/datafusion-comet/pull/2528?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Iceberg scan based serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#issuecomment-3372276529
It is promising!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2534533647
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import scala.util.Random
+
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
+import org.apache.spark.sql.types._
+
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions}
+
+class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
+
+ test("select *") {
+ val sql = s"SELECT * FROM $icebergTableName"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("select * with limit") {
+ val sql = s"SELECT * FROM $icebergTableName LIMIT 500"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by single column") {
+ val df = spark.table(icebergTableName)
+ for (col <- df.columns) {
+ val sql = s"SELECT $col FROM $icebergTableName ORDER BY $col"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("order by multiple columns") {
+ val df = spark.table(icebergTableName)
+ val allCols = df.columns.mkString(",")
+ val sql = s"SELECT $allCols FROM $icebergTableName ORDER BY $allCols"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by random columns") {
+ val df = spark.table(icebergTableName)
+
+ for (_ <- 1 to 10) {
+ // We only do order by permutations of primitive types to exercise native shuffle's
+ // RangePartitioning which only supports those types.
+ // This used to be df.columns.slice(0, 14) but we're removing two binary columns in
+ // CometFuzzIcebergBase.beforeAll
+ val shuffledPrimitiveCols = Random.shuffle(df.columns.slice(0, 12).toList)
+ val randomSize = Random.nextInt(shuffledPrimitiveCols.length) + 1
+ val randomColsSubset = shuffledPrimitiveCols.take(randomSize).toArray.mkString(",")
+ val sql = s"SELECT $randomColsSubset FROM $icebergTableName ORDER BY $randomColsSubset"
+ checkSparkAnswerAndOperator(sql)
+ }
+ }
+
+ test("distribute by single column (complex types)") {
+ val df = spark.table(icebergTableName)
+ val columns = df.schema.fields.filter(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- columns) {
+ // DISTRIBUTE BY is equivalent to df.repartition($col) and uses
+ val sql = s"SELECT $col FROM $icebergTableName DISTRIBUTE BY $col"
+ val resultDf = spark.sql(sql)
+ resultDf.collect()
+ // check for Comet shuffle
+ val plan =
+ resultDf.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan
+ val cometShuffleExchanges = collectCometShuffleExchanges(plan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffleExchanges.length == 1)
+ }
+ }
+
+ test("shuffle supports all types") {
+ val df = spark.table(icebergTableName)
+ val df2 = df.repartition(8, df.col("c0")).sort("c1")
+ df2.collect()
+ val cometShuffles = collectCometShuffleExchanges(df2.queryExecution.executedPlan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffles.length == 2)
+ }
+
+ test("join") {
+ val df = spark.table(icebergTableName)
+ df.createOrReplaceTempView("t1")
+ df.createOrReplaceTempView("t2")
+ // Filter out complex types - iceberg-rust can't create predicates for struct/array/map equality
+ val primitiveColumns = df.schema.fields.filterNot(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- primitiveColumns) {
+ // cannot run fully native due to HashAggregate
+ val sql = s"SELECT count(*) FROM t1 JOIN t2 ON t1.$col = t2.$col"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(2 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("decode") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // BinaryType, since then this test would not run any queries and silently pass.
+ var testedBinary = false
+ for (field <- df.schema.fields if field.dataType == BinaryType) {
+ testedBinary = true
+ // Intentionally use odd capitalization of 'utf-8' to test normalization.
+ val sql = s"SELECT decode(${field.name}, 'utF-8') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedBinary)
+ }
+
+ test("regexp_replace") {
+ withSQLConf(CometConf.COMET_REGEXP_ALLOW_INCOMPATIBLE.key -> "true") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // StringType, since then this test would not run any queries and silently pass.
+ var testedString = false
+ for (field <- df.schema.fields if field.dataType == StringType) {
+ testedString = true
+ val sql = s"SELECT regexp_replace(${field.name}, 'a', 'b') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedString)
+ }
+ }
+
+ test("Iceberg temporal types written as INT96") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.INT96)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MICROS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MICROS)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MILLIS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MILLIS)
+ }
+
+ private def testIcebergTemporalTypes(
+ outputTimestampType: ParquetOutputTimestampType.Value,
+ generateArray: Boolean = true,
+ generateStruct: Boolean = true): Unit = {
+
+ val schema = FuzzDataGenerator.generateSchema(
+ SchemaGenOptions(
+ generateArray = generateArray,
+ generateStruct = generateStruct,
+ primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot { dataType =>
+ // Disable decimals - iceberg-rust doesn't support FIXED_LEN_BYTE_ARRAY in page index yet
+ dataType.isInstanceOf[DecimalType]
+ }))
+
+ val options =
+ DataGenOptions(generateNegativeZero = false)
+
+ withTempPath { filename =>
+ val random = new Random(42)
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString,
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> defaultTimezone) {
+ ParquetGenerator.makeParquetFile(random, spark, filename.toString, schema, 100, options)
+ }
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
+ Seq(true, false).foreach { inferTimestampNtzEnabled =>
+ Seq(true, false).foreach { int96TimestampConversion =>
+ Seq(true, false).foreach { int96AsTimestamp =>
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz,
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key -> int96AsTimestamp.toString,
+ SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION.key -> int96TimestampConversion.toString,
+ SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key -> inferTimestampNtzEnabled.toString) {
+
+ val df = spark.table(icebergTableName)
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
Review Comment:
This answers my earlier question about timezone testing
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: [iceberg] Native scan by serializing FileScanTasks to iceberg-rust [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2528:
URL: https://github.com/apache/datafusion-comet/pull/2528#discussion_r2534890396
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import scala.util.Random
+
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
+import org.apache.spark.sql.types._
+
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions}
+
+class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
+
+ test("select *") {
+ val sql = s"SELECT * FROM $icebergTableName"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("select * with limit") {
+ val sql = s"SELECT * FROM $icebergTableName LIMIT 500"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by single column") {
+ val df = spark.table(icebergTableName)
+ for (col <- df.columns) {
+ val sql = s"SELECT $col FROM $icebergTableName ORDER BY $col"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("order by multiple columns") {
+ val df = spark.table(icebergTableName)
+ val allCols = df.columns.mkString(",")
+ val sql = s"SELECT $allCols FROM $icebergTableName ORDER BY $allCols"
+ // cannot run fully natively due to range partitioning and sort
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(1 == collectIcebergNativeScans(cometPlan).length)
+ }
+
+ test("order by random columns") {
+ val df = spark.table(icebergTableName)
+
+ for (_ <- 1 to 10) {
+ // We only do order by permutations of primitive types to exercise native shuffle's
+ // RangePartitioning which only supports those types.
+ // This used to be df.columns.slice(0, 14) but we're removing two binary columns in
+ // CometFuzzIcebergBase.beforeAll
+ val shuffledPrimitiveCols = Random.shuffle(df.columns.slice(0, 12).toList)
+ val randomSize = Random.nextInt(shuffledPrimitiveCols.length) + 1
+ val randomColsSubset = shuffledPrimitiveCols.take(randomSize).toArray.mkString(",")
+ val sql = s"SELECT $randomColsSubset FROM $icebergTableName ORDER BY $randomColsSubset"
+ checkSparkAnswerAndOperator(sql)
+ }
+ }
+
+ test("distribute by single column (complex types)") {
+ val df = spark.table(icebergTableName)
+ val columns = df.schema.fields.filter(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- columns) {
+ // DISTRIBUTE BY is equivalent to df.repartition($col) and uses
+ val sql = s"SELECT $col FROM $icebergTableName DISTRIBUTE BY $col"
+ val resultDf = spark.sql(sql)
+ resultDf.collect()
+ // check for Comet shuffle
+ val plan =
+ resultDf.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan
+ val cometShuffleExchanges = collectCometShuffleExchanges(plan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffleExchanges.length == 1)
+ }
+ }
+
+ test("shuffle supports all types") {
+ val df = spark.table(icebergTableName)
+ val df2 = df.repartition(8, df.col("c0")).sort("c1")
+ df2.collect()
+ val cometShuffles = collectCometShuffleExchanges(df2.queryExecution.executedPlan)
+ // Iceberg native scan supports complex types
+ assert(cometShuffles.length == 2)
+ }
+
+ test("join") {
+ val df = spark.table(icebergTableName)
+ df.createOrReplaceTempView("t1")
+ df.createOrReplaceTempView("t2")
+ // Filter out complex types - iceberg-rust can't create predicates for struct/array/map equality
+ val primitiveColumns = df.schema.fields.filterNot(f => isComplexType(f.dataType)).map(_.name)
+ for (col <- primitiveColumns) {
+ // cannot run fully native due to HashAggregate
+ val sql = s"SELECT count(*) FROM t1 JOIN t2 ON t1.$col = t2.$col"
+ val (_, cometPlan) = checkSparkAnswer(sql)
+ assert(2 == collectIcebergNativeScans(cometPlan).length)
+ }
+ }
+
+ test("decode") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // BinaryType, since then this test would not run any queries and silently pass.
+ var testedBinary = false
+ for (field <- df.schema.fields if field.dataType == BinaryType) {
+ testedBinary = true
+ // Intentionally use odd capitalization of 'utf-8' to test normalization.
+ val sql = s"SELECT decode(${field.name}, 'utF-8') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedBinary)
+ }
+
+ test("regexp_replace") {
+ withSQLConf(CometConf.COMET_REGEXP_ALLOW_INCOMPATIBLE.key -> "true") {
+ val df = spark.table(icebergTableName)
+ // We want to make sure that the schema generator wasn't modified to accidentally omit
+ // StringType, since then this test would not run any queries and silently pass.
+ var testedString = false
+ for (field <- df.schema.fields if field.dataType == StringType) {
+ testedString = true
+ val sql = s"SELECT regexp_replace(${field.name}, 'a', 'b') FROM $icebergTableName"
+ checkSparkAnswerAndOperator(sql)
+ }
+ assert(testedString)
+ }
+ }
+
+ test("Iceberg temporal types written as INT96") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.INT96)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MICROS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MICROS)
+ }
+
+ test("Iceberg temporal types written as TIMESTAMP_MILLIS") {
+ testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MILLIS)
+ }
+
+ private def testIcebergTemporalTypes(
+ outputTimestampType: ParquetOutputTimestampType.Value,
+ generateArray: Boolean = true,
+ generateStruct: Boolean = true): Unit = {
+
+ val schema = FuzzDataGenerator.generateSchema(
+ SchemaGenOptions(
+ generateArray = generateArray,
+ generateStruct = generateStruct,
+ primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot { dataType =>
+ // Disable decimals - iceberg-rust doesn't support FIXED_LEN_BYTE_ARRAY in page index yet
+ dataType.isInstanceOf[DecimalType]
+ }))
+
+ val options =
+ DataGenOptions(generateNegativeZero = false)
+
+ withTempPath { filename =>
+ val random = new Random(42)
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString,
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> defaultTimezone) {
+ ParquetGenerator.makeParquetFile(random, spark, filename.toString, schema, 100, options)
+ }
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
+ Seq(true, false).foreach { inferTimestampNtzEnabled =>
+ Seq(true, false).foreach { int96TimestampConversion =>
+ Seq(true, false).foreach { int96AsTimestamp =>
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz,
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key -> int96AsTimestamp.toString,
+ SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION.key -> int96TimestampConversion.toString,
+ SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key -> inferTimestampNtzEnabled.toString) {
+
+ val df = spark.table(icebergTableName)
+
+ Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz =>
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
+ def hasTemporalType(t: DataType): Boolean = t match {
+ case DataTypes.DateType | DataTypes.TimestampType |
+ DataTypes.TimestampNTZType =>
+ true
+ case t: StructType => t.exists(f => hasTemporalType(f.dataType))
+ case t: ArrayType => hasTemporalType(t.elementType)
Review Comment:
Not sure, the existing fuzz test suite doesn't, and I just duplicated that logic. It doesn't mean we shouldn't test it though.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org