diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/aggregate.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/aggregate.md index 7aac7d3ff0..fc133dc04d 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/aggregate.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/aggregate.md @@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a The following differences from Spark are always present and do not require any additional configuration: - This function is not deterministic. Results may not match Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/array.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/array.md index b207cbcbc4..29db8ad4cc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/array.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/array.md @@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau The following cases are not supported by Comet: - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/datetime.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/datetime.md index daae231855..28c05636bc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/datetime.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/datetime.md @@ -136,4 +136,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/map.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/map.md index cd26218a31..349be08ae8 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/map.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/map.md @@ -38,4 +38,5 @@ By default, Comet accelerates `MapFromEntries` using JVM codegen dispatch, which - `BinaryType` is not supported as a map key in `map_from_entries` - `BinaryType` is not supported as a map value in `map_from_entries` + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/math.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/math.md index 8b2aefc327..5f456c9e3f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/math.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/math.md @@ -38,4 +38,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - The `seed` argument must be a literal value + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/misc.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/misc.md index 185282dcdf..6ab37a60b4 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/misc.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/misc.md @@ -50,4 +50,5 @@ The following cases are not supported by Comet: The following incompatibilities cause `SortOrder` to fall back to Spark by default. Set `spark.comet.expression.SortOrder.allowIncompatible=true` to enable Comet acceleration despite these differences. - When `spark.comet.exec.strictFloatingPoint=true`, sorting on floating-point types is not 100% compatible with Spark + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/string.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/string.md index f5b06c6175..cc731b85f1 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/string.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/string.md @@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences. - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/struct.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/struct.md index 72db34d875..af5ed3f503 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/struct.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/struct.md @@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de The following cases are not supported by Comet: - Complex types (arrays, maps, structs) in the schema are not supported + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/aggregate.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/aggregate.md index 7aac7d3ff0..fc133dc04d 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/aggregate.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/aggregate.md @@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a The following differences from Spark are always present and do not require any additional configuration: - This function is not deterministic. Results may not match Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/array.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/array.md index b207cbcbc4..29db8ad4cc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/array.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/array.md @@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau The following cases are not supported by Comet: - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/datetime.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/datetime.md index daae231855..28c05636bc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/datetime.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/datetime.md @@ -136,4 +136,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/map.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/map.md index cd26218a31..349be08ae8 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/map.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/map.md @@ -38,4 +38,5 @@ By default, Comet accelerates `MapFromEntries` using JVM codegen dispatch, which - `BinaryType` is not supported as a map key in `map_from_entries` - `BinaryType` is not supported as a map value in `map_from_entries` + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/math.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/math.md index 8b2aefc327..5f456c9e3f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/math.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/math.md @@ -38,4 +38,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - The `seed` argument must be a literal value + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/misc.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/misc.md index 67541f5551..d786b0aa0f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/misc.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/misc.md @@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau The following cases are not supported by Comet: - Falls back to Spark when the input type cannot be cast to string. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/string.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/string.md index f5b06c6175..cc731b85f1 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/string.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/string.md @@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences. - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/struct.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/struct.md index 72db34d875..af5ed3f503 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/struct.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.5/struct.md @@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de The following cases are not supported by Comet: - Complex types (arrays, maps, structs) in the schema are not supported + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/aggregate.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/aggregate.md index 7aac7d3ff0..fc133dc04d 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/aggregate.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/aggregate.md @@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a The following differences from Spark are always present and do not require any additional configuration: - This function is not deterministic. Results may not match Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/array.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/array.md index b207cbcbc4..29db8ad4cc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/array.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/array.md @@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau The following cases are not supported by Comet: - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/datetime.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/datetime.md index daae231855..28c05636bc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/datetime.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/datetime.md @@ -136,4 +136,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/map.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/map.md index 3d872a3a41..1bf4790b49 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/map.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/map.md @@ -48,4 +48,5 @@ The following incompatibilities cause `MapSort` to fall back to Spark by default The following cases are not supported by Comet: - MapSort is unsupported for non-scalar key types (struct, array, map, etc.). + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/math.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/math.md index 8b2aefc327..5f456c9e3f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/math.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/math.md @@ -38,4 +38,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - The `seed` argument must be a literal value + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/misc.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/misc.md index 67541f5551..d786b0aa0f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/misc.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/misc.md @@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau The following cases are not supported by Comet: - Falls back to Spark when the input type cannot be cast to string. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/string.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/string.md index f5b06c6175..cc731b85f1 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/string.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/string.md @@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences. - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/struct.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/struct.md index 72db34d875..af5ed3f503 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/struct.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.0/struct.md @@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de The following cases are not supported by Comet: - Complex types (arrays, maps, structs) in the schema are not supported + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/aggregate.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/aggregate.md index 7aac7d3ff0..fc133dc04d 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/aggregate.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/aggregate.md @@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a The following differences from Spark are always present and do not require any additional configuration: - This function is not deterministic. Results may not match Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/array.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/array.md index b207cbcbc4..29db8ad4cc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/array.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/array.md @@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau The following cases are not supported by Comet: - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/datetime.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/datetime.md index daae231855..28c05636bc 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/datetime.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/datetime.md @@ -136,4 +136,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/map.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/map.md index 3d872a3a41..1bf4790b49 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/map.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/map.md @@ -48,4 +48,5 @@ The following incompatibilities cause `MapSort` to fall back to Spark by default The following cases are not supported by Comet: - MapSort is unsupported for non-scalar key types (struct, array, map, etc.). + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/math.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/math.md index 8b2aefc327..5f456c9e3f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/math.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/math.md @@ -38,4 +38,5 @@ The following cases are not supported by Comet: The following cases are not supported by Comet: - The `seed` argument must be a literal value + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/misc.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/misc.md index 67541f5551..d786b0aa0f 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/misc.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/misc.md @@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau The following cases are not supported by Comet: - Falls back to Spark when the input type cannot be cast to string. + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/string.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/string.md index f5b06c6175..cc731b85f1 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/string.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/string.md @@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences. - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel + diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/struct.md b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/struct.md index 72db34d875..af5ed3f503 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/struct.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-4.1/struct.md @@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de The following cases are not supported by Comet: - Complex types (arrays, maps, structs) in the schema are not supported + diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md index 442133facc..3bd1ec27ee 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -92,6 +92,7 @@ Here are the direct links for downloading the Comet $COMET_VERSION jar file. - [Comet plugin for Spark 3.5 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.5_2.13/$COMET_VERSION/comet-spark-spark3.5_2.13-$COMET_VERSION.jar) - [Comet plugin for Spark 4.0 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark4.0_2.13/$COMET_VERSION/comet-spark-spark4.0_2.13-$COMET_VERSION.jar) - [Comet plugin for Spark 4.1 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark4.1_2.13/$COMET_VERSION/comet-spark-spark4.1_2.13-$COMET_VERSION.jar) + ## Building from source diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index 0dbfff457b..d2549b7fa2 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -41,6 +41,7 @@ use futures::{Stream, StreamExt, TryStreamExt}; use iceberg::arrow::ScanMetrics; use iceberg::io::{FileIO, FileIOBuilder, StorageFactory}; use iceberg::Runtime as IcebergRuntime; +use iceberg::{Error, ErrorKind}; use iceberg_storage_opendal::CustomAwsCredentialLoader; use iceberg_storage_opendal::OpenDalStorageFactory; @@ -57,6 +58,14 @@ use iceberg::scan::FileScanTask; /// catalog's `s3.*` property bag. const ICEBERG_PROVIDER_CLASS_PROPERTY: &str = "s3.comet.credential.provider.class"; +/// A valid Parquet file ends with at least an 8-byte footer (4-byte metadata length + "PAR1"). +/// A delete file that stats below this cannot be read, so we reject it in the fill step. opendal +/// returns size 0 from a successful HEAD whose response carries no Content-Length header (some +/// S3-compatible endpoints/proxies, or a path-style mismatch), and for a genuinely empty object; +/// neither errors at the stat layer, so without this floor the 0 would flow into the Parquet +/// reader and surface as an opaque "file size of 0 is less than footer". +const MIN_PARQUET_FILE_SIZE: u64 = 8; + /// Iceberg table scan operator that uses iceberg-rust to read Iceberg tables. /// /// Executes pre-planned FileScanTasks for efficient parallel scanning. @@ -176,10 +185,20 @@ impl IcebergScanExec { let batch_size = context.session_config().batch_size(); let metrics = IcebergScanMetrics::new(&self.metrics); - let num_tasks = tasks.len(); - metrics.num_splits.add(num_tasks); - - let task_stream = futures::stream::iter(tasks.into_iter().map(Ok)).boxed(); + metrics.num_splits.add(tasks.len()); + + // Fill delete-file sizes as the first step of the task stream so the stats run on the + // iceberg runtime alongside the reads, not on the calling executor thread (see + // fill_delete_file_sizes). + let fill_io = file_io.clone(); + let concurrency_limit = self.data_file_concurrency_limit; + let task_stream = futures::stream::once(async move { + let mut tasks = tasks; + Self::fill_delete_file_sizes(&mut tasks, &fill_io, concurrency_limit).await?; + Ok::<_, Error>(futures::stream::iter(tasks.into_iter().map(Ok::<_, Error>))) + }) + .try_flatten() + .boxed(); // iceberg-rust's ArrowReader spawns IO/CPU work onto an iceberg::Runtime. execute() runs // on the JVM-called thread outside any tokio context, so Runtime::current() would panic; @@ -248,6 +267,86 @@ impl IcebergScanExec { } } + /// Stats each unique delete file to fill its `file_size_in_bytes` (0 on arrival, since it is + /// not serialized). + /// + /// iceberg-rust seeks the Parquet footer from this size, so it must be correct. A stat failure + /// is fatal: reading with missing deletes would silently leak deleted rows. + async fn fill_delete_file_sizes( + tasks: &mut [FileScanTask], + file_io: &FileIO, + concurrency_limit: usize, + ) -> Result<(), Error> { + use datafusion::common::{HashMap, HashSet}; + use futures::TryStreamExt; + + // Dedup: the JVM pools delete-file lists, not individual files, so the same delete file + // recurs across the tasks that share it. Owning the paths also avoids holding a borrow of + // `tasks` into the mutable write-back below. + let mut needed: HashSet = HashSet::new(); + for task in tasks.iter() { + for delete in &task.deletes { + // Delete-file sizes are never serialized, so they always arrive as 0. If we ever + // trust manifest sizes (pending the unreleased apache/iceberg#12554 fix), skip + // already-sized files here instead of asserting. + debug_assert_eq!(delete.file_size_in_bytes, 0); + needed.insert(delete.file_path.clone()); + } + } + if needed.is_empty() { + return Ok(()); + } + + // Bound the in-flight stats to match the downstream read concurrency (iceberg-rust uses + // try_buffer_unordered at the same limit for delete-file loads). An unbounded fan-out + // would burst N HEAD requests at once for no gain, since the reads are throttled anyway. + // Guaranteed > 0 by COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT; buffer_unordered(0) would + // never poll. + debug_assert!(concurrency_limit > 0); + let sizes: Vec<(String, u64)> = futures::stream::iter(needed.into_iter().map(|path| { + let file_io = file_io.clone(); + async move { + let size = file_io + .new_input(&path)? + .metadata() + .await + .map_err(|e| { + Error::new( + ErrorKind::Unexpected, + format!("Failed to stat delete file '{path}'"), + ) + .with_source(e) + })? + .size; + if size < MIN_PARQUET_FILE_SIZE { + return Err(Error::new( + ErrorKind::Unexpected, + format!( + "Delete file '{path}' statted to {size} bytes, below the \ + {MIN_PARQUET_FILE_SIZE}-byte Parquet minimum. The object is empty or \ + truncated, or the stat returned no Content-Length (check the object \ + store endpoint, TLS, and path-style config)." + ), + )); + } + Ok::<(String, u64), Error>((path, size)) + } + })) + .buffer_unordered(concurrency_limit) + .try_collect() + .await?; + + let size_map: HashMap = sizes.into_iter().collect(); + for task in tasks.iter_mut() { + for delete in task.deletes.iter_mut() { + if let Some(&size) = size_map.get(&delete.file_path) { + delete.file_size_in_bytes = size; + } + } + } + Ok(()) + } + fn load_file_io( catalog_properties: &HashMap, metadata_location: &str, @@ -525,3 +624,113 @@ fn adapt_batch_with_expressions( RecordBatch::try_new(Arc::clone(target_schema), columns).map_err(|e| e.into()) } + +#[cfg(test)] +mod tests { + use std::io::Write; + use std::sync::Arc; + + use iceberg::io::{FileIO, FileIOBuilder}; + use iceberg::scan::{FileScanTask, FileScanTaskDeleteFile}; + use iceberg::spec::{DataContentType, DataFileFormat, Schema}; + use iceberg_storage_opendal::OpenDalStorageFactory; + + use super::IcebergScanExec; + + fn fs_file_io() -> FileIO { + FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Fs)).build() + } + + fn task_with_deletes(deletes: Vec) -> FileScanTask { + FileScanTask { + file_size_in_bytes: 0, + start: 0, + length: 0, + record_count: None, + data_file_path: "data.parquet".to_string(), + data_file_format: DataFileFormat::Parquet, + schema: Arc::new(Schema::builder().build().unwrap()), + project_field_ids: vec![], + predicate: None, + deletes, + partition: None, + partition_spec: None, + name_mapping: None, + case_sensitive: false, + } + } + + fn delete_file(path: &str) -> FileScanTaskDeleteFile { + FileScanTaskDeleteFile { + file_path: path.to_string(), + file_type: DataContentType::PositionDeletes, + file_size_in_bytes: 0, + partition_spec_id: 0, + equality_ids: None, + } + } + + // A delete file we cannot stat must fail the scan, not be silently read with a missing/0 size. + #[tokio::test] + async fn fill_delete_file_sizes_errors_on_unreadable_delete_file() { + let dir = tempfile::tempdir().unwrap(); + let missing = dir.path().join("missing-delete.parquet"); + let mut tasks = vec![task_with_deletes(vec![delete_file( + missing.to_str().unwrap(), + )])]; + + let result = IcebergScanExec::fill_delete_file_sizes(&mut tasks, &fs_file_io(), 4).await; + + assert!( + result.is_err(), + "expected an error when a delete file cannot be statted" + ); + assert_eq!(tasks[0].deletes[0].file_size_in_bytes, 0); + } + + // The real on-disk size is filled in from the FileIO, replacing the 0 placeholder. + #[tokio::test] + async fn fill_delete_file_sizes_populates_real_size() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("delete.parquet"); + let bytes = b"these bytes stand in for a delete file"; + let mut f = std::fs::File::create(&path).unwrap(); + f.write_all(bytes).unwrap(); + f.flush().unwrap(); + + let mut tasks = vec![task_with_deletes(vec![delete_file(path.to_str().unwrap())])]; + IcebergScanExec::fill_delete_file_sizes(&mut tasks, &fs_file_io(), 4) + .await + .unwrap(); + + assert_eq!(tasks[0].deletes[0].file_size_in_bytes, bytes.len() as u64); + } + + // A present-but-undersized delete file (0-byte object, truncated write, or a HEAD with no + // Content-Length that opendal reports as size 0) must fail loudly rather than passing a + // sub-footer size into the Parquet reader. + #[tokio::test] + async fn fill_delete_file_sizes_errors_on_subfooter_delete_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("empty-delete.parquet"); + std::fs::File::create(&path).unwrap(); // 0 bytes on disk + + let mut tasks = vec![task_with_deletes(vec![delete_file(path.to_str().unwrap())])]; + let result = IcebergScanExec::fill_delete_file_sizes(&mut tasks, &fs_file_io(), 4).await; + + assert!( + result.is_err(), + "expected an error when a delete file is below the Parquet footer minimum" + ); + assert_eq!(tasks[0].deletes[0].file_size_in_bytes, 0); + } + + // No deletes means no stats and no error. + #[tokio::test] + async fn fill_delete_file_sizes_noop_without_deletes() { + let mut tasks = vec![task_with_deletes(vec![])]; + IcebergScanExec::fill_delete_file_sizes(&mut tasks, &fs_file_io(), 4) + .await + .unwrap(); + } +} diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index c07a92d700..eb46f36844 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3393,7 +3393,8 @@ fn parse_file_scan_tasks_from_common( Ok(iceberg::scan::FileScanTaskDeleteFile { file_path: del.file_path.clone(), file_type, - file_size_in_bytes: del.file_size_in_bytes, + // Not serialized; filled in by IcebergScanExec::fill_delete_file_sizes. + file_size_in_bytes: 0, partition_spec_id: del.partition_spec_id, equality_ids: if del.equality_ids.is_empty() { None diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 15265f1d86..c475ae5b0d 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -271,9 +271,6 @@ message IcebergDeleteFile { // Equality field IDs (empty for positional deletes) repeated int32 equality_ids = 4; - - // // Total file size from the manifest entry, used to skip stat/HEAD calls - uint64 file_size_in_bytes = 5; } message Projection { diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 8c88c19f1f..b8afc159bb 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -221,85 +221,69 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit /** * Extracts delete files from an Iceberg FileScanTask as a list (for deduplication). + * + * Delete-file size is not serialized; the native scan stats each file for it (see + * IcebergScanExec::fill_delete_file_sizes). The manifest size cannot be trusted as a substitute + * (apache/iceberg#12554). */ private def extractDeleteFilesList( task: Any, contentFileClass: Class[_], - fileScanTaskClass: Class[_], - fileIO: Option[Any]): Seq[OperatorOuterClass.IcebergDeleteFile] = { + fileScanTaskClass: Class[_]): Seq[OperatorOuterClass.IcebergDeleteFile] = { try { val deleteFileClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE) val deletes = IcebergReflection.getDeleteFilesFromTask(task, fileScanTaskClass) - deletes.asScala.flatMap { deleteFile => - try { - IcebergReflection - .extractFileLocation(contentFileClass, deleteFile) - .map { deletePath => - val deleteBuilder = - OperatorOuterClass.IcebergDeleteFile.newBuilder() - deleteBuilder.setFilePath(deletePath) - - val contentType = - try { - val contentMethod = deleteFileClass.getMethod("content") - val content = contentMethod.invoke(deleteFile) - content.toString match { - case IcebergReflection.ContentTypes.POSITION_DELETES => - IcebergReflection.ContentTypes.POSITION_DELETES - case IcebergReflection.ContentTypes.EQUALITY_DELETES => - IcebergReflection.ContentTypes.EQUALITY_DELETES - case other => other - } - } catch { - case _: Exception => - IcebergReflection.ContentTypes.POSITION_DELETES - } - deleteBuilder.setContentType(contentType) - - val specId = - try { - val specIdMethod = deleteFileClass.getMethod("specId") - specIdMethod.invoke(deleteFile).asInstanceOf[Int] - } catch { - case _: Exception => 0 - } - deleteBuilder.setPartitionSpecId(specId) - - // Workaround for https://github.com/apache/iceberg/issues/12554 - // RewriteTablePath rewrites path references inside position delete files, - // making the copied file possibly differ in size, but does not update - // file_size_in_bytes in the manifest. The manifest value cannot be trusted; - // always use FileIO to get the actual size. - val inputFile = fileIO.get.getClass - .getMethod("newInputFile", classOf[String]) - .invoke(fileIO.get, deletePath) - val actualDeleteFileSizeInBytes = - inputFile.getClass - .getMethod("getLength") - .invoke(inputFile) - .asInstanceOf[Long] - deleteBuilder.setFileSizeInBytes(actualDeleteFileSizeInBytes) - - try { - val equalityIdsMethod = - deleteFileClass.getMethod("equalityFieldIds") - val equalityIds = equalityIdsMethod - .invoke(deleteFile) - .asInstanceOf[java.util.List[Integer]] - equalityIds.forEach(id => deleteBuilder.addEqualityIds(id)) - } catch { - case _: Exception => - } - - deleteBuilder.build() + deletes.asScala.map { deleteFile => + // The path is the one essential field. A delete file we cannot locate cannot be applied, + // and silently skipping it would leak deleted rows, so treat a missing path as fatal. + val deletePath = IcebergReflection + .extractFileLocation(contentFileClass, deleteFile) + .getOrElse( + throw new RuntimeException("Failed to extract delete file path from FileScanTask")) + + val deleteBuilder = OperatorOuterClass.IcebergDeleteFile.newBuilder() + deleteBuilder.setFilePath(deletePath) + + val contentType = + try { + val contentMethod = deleteFileClass.getMethod("content") + val content = contentMethod.invoke(deleteFile) + content.toString match { + case IcebergReflection.ContentTypes.POSITION_DELETES => + IcebergReflection.ContentTypes.POSITION_DELETES + case IcebergReflection.ContentTypes.EQUALITY_DELETES => + IcebergReflection.ContentTypes.EQUALITY_DELETES + case other => other } + } catch { + case _: Exception => + IcebergReflection.ContentTypes.POSITION_DELETES + } + deleteBuilder.setContentType(contentType) + + val specId = + try { + val specIdMethod = deleteFileClass.getMethod("specId") + specIdMethod.invoke(deleteFile).asInstanceOf[Int] + } catch { + case _: Exception => 0 + } + deleteBuilder.setPartitionSpecId(specId) + + try { + val equalityIdsMethod = + deleteFileClass.getMethod("equalityFieldIds") + val equalityIds = equalityIdsMethod + .invoke(deleteFile) + .asInstanceOf[java.util.List[Integer]] + equalityIds.forEach(id => deleteBuilder.addEqualityIds(id)) } catch { - case e: Exception => - logWarning(s"Failed to serialize delete file: ${e.getMessage}") - None + case _: Exception => } + + deleteBuilder.build() }.toSeq } catch { case e: Exception => @@ -747,8 +731,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit mutable.HashMap[Seq[OperatorOuterClass.IcebergDeleteFile], Int]() val residualToPoolIndex = mutable.HashMap[Option[Expr], Int]() - val fileIO = IcebergReflection.getFileIO(metadata.table) - val perPartitionBuilders = mutable.ArrayBuffer[OperatorOuterClass.IcebergScan]() var totalTasks = 0 @@ -906,7 +888,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit taskBuilder.setProjectFieldIdsIdx(projectFieldIdsIdx) val deleteFilesList = - extractDeleteFilesList(task, contentFileClass, fileScanTaskClass, fileIO) + extractDeleteFilesList(task, contentFileClass, fileScanTaskClass) if (deleteFilesList.nonEmpty) { val deleteFilesIdx = deleteFilesToPoolIndex.getOrElseUpdate( deleteFilesList, { diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 9964fa854e..af612f64ca 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -20,6 +20,7 @@ package org.apache.comet import java.io.File +import java.net.URI import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -36,7 +37,7 @@ import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExcha import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{StringType, TimestampType} -import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark42Plus} +import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} import org.apache.comet.iceberg.RESTCatalogHelper import org.apache.comet.testing.{FuzzDataGenerator, SchemaGenOptions} @@ -695,6 +696,127 @@ class CometIcebergNativeSuite } } + // MOR positional deletes accumulated across multiple snapshots, then compacted with + // rewrite_data_files (the equivalent of Trino's OPTIMIZE). Exercises the native delete-file + // size fill over a multi-snapshot, post-compaction delete layout. + test("MOR positional deletes after rewrite_data_files compaction") { + assume(icebergAvailable, "Iceberg not available in classpath") + // CALL is only in Spark's own SQL grammar on 4.0+; on 3.x it needs Iceberg's + // session-extension parser, which the Comet test session does not register. + assume(isSpark40Plus, "CALL procedure syntax requires Spark 4.0+") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.compacted_delete_test ( + id INT, + value DOUBLE + ) USING iceberg + TBLPROPERTIES ( + 'write.delete.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read', + 'format-version' = '2' + ) + """) + + // Build a multi-snapshot history: several inserts into separate data files, + // interleaved with deletes that produce positional delete files. + (0 until 5).foreach { batch => + val lo = batch * 100 + val hi = lo + 100 + spark.sql(s""" + INSERT INTO test_cat.db.compacted_delete_test + SELECT id, CAST(id AS DOUBLE) * 1.5 AS value + FROM range($lo, $hi) + """) + spark.sql( + s"DELETE FROM test_cat.db.compacted_delete_test WHERE id % 7 = 0 AND id >= $lo AND id < $hi") + } + + // Compaction equivalent to Trino's OPTIMIZE; rewrites data files and may rewrite or + // re-reference position delete files. + spark.sql("CALL test_cat.system.rewrite_data_files(table => 'db.compacted_delete_test')") + + // More deletes after compaction so the live snapshot mixes pre- and post-compaction deletes. + spark.sql("DELETE FROM test_cat.db.compacted_delete_test WHERE id % 13 = 0") + + checkIcebergNativeScan("SELECT * FROM test_cat.db.compacted_delete_test ORDER BY id") + checkIcebergNativeScan("SELECT COUNT(*) FROM test_cat.db.compacted_delete_test") + checkIcebergNativeScan("SELECT SUM(value) FROM test_cat.db.compacted_delete_test") + + spark.sql("DROP TABLE test_cat.db.compacted_delete_test") + } + } + } + + // When a delete file cannot be statted natively, the scan must fail loudly rather than read the + // data file with deletes silently dropped. Here we delete the positional delete file from disk + // after it is committed, so the native fill_delete_file_sizes stat fails. + test("delete file stat failure surfaces as an exception") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.missing_delete_test ( + id INT, + value DOUBLE + ) USING iceberg + TBLPROPERTIES ( + 'write.delete.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read', + 'format-version' = '2' + ) + """) + + spark.sql(""" + INSERT INTO test_cat.db.missing_delete_test + SELECT id, CAST(id AS DOUBLE) FROM range(20) + """) + spark.sql("DELETE FROM test_cat.db.missing_delete_test WHERE id IN (3, 7, 11)") + + // Authoritative delete file paths from Iceberg's metadata table (Comet falls back to + // Spark for metadata tables, so this read does not depend on the native scan). + val deletePaths = spark + .sql("SELECT file_path FROM test_cat.db.missing_delete_test.delete_files") + .collect() + .map(_.getString(0)) + assert(deletePaths.nonEmpty, "expected at least one positional delete file") + + deletePaths.foreach { p => + // Iceberg may report the path with or without a file: scheme. + val local = if (p.startsWith("file:")) new File(new URI(p)) else new File(p) + assert(local.delete(), s"failed to remove delete file from disk: $p") + } + + // If deletes were silently dropped, COUNT(*) would return a wrong count with no error and + // this intercept would fail. The fill must fail the scan instead. + val e = intercept[Exception] { + spark.sql("SELECT COUNT(*) FROM test_cat.db.missing_delete_test").collect() + } + assert( + e.getMessage != null && e.getMessage.contains("stat delete file"), + s"expected a delete-file stat failure, got: ${e.getMessage}") + + spark.sql("DROP TABLE test_cat.db.missing_delete_test") + } + } + } + test("verify no duplicate rows across multiple partitions") { assume(icebergAvailable, "Iceberg not available in classpath")