You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/10 13:15:59 UTC
[PR] feat: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
andygrove opened a new pull request, #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851
## Which issue does this PR close?
No issue. This lifts one of the restrictions listed in #5643: the fallback that #6531 added for #6138.
## Rationale for this change
Since #6531, a native Iceberg write falls back to iceberg-java when the output partition spec has a `float` or `double` field. iceberg-rust compared float partition values with `OrderedFloat`'s equality, which treats `-0.0` and `0.0` as one value. Both of the native writer's grouping paths then filed rows with either value under whichever arrived first, and a read that pruned on the other value lost rows (#6138).
apache/iceberg-rust#3327 gives `PrimitiveLiteral` hand-written `PartialEq` and `PartialOrd` that follow Java's `Float.compare` and `Double.compare`: `-0.0` sorts before `0.0`, and all NaNs are equal. That is how iceberg-java compares partition keys, so with the pin past it the fallback can go.
## What changes are included in this PR?
- The `iceberg` and `iceberg-storage-opendal` pin moves from `af1da4c` to `1f3bc34`, the merge commit of apache/iceberg-rust#3327. That is 17 commits. `Cargo.lock` picks up apache-avro 0.22 and nine new crate versions (ouroboros, strum 0.28, aliasable and their helpers).
- `CometIcebergNativeWrite` drops the float and double partition rule, and `IcebergReflection` drops its helper.
- Two other upstream changes in the range need code changes:
- apache/iceberg-rust#3145 makes `FileScanTaskDeleteFile`'s fields private behind a builder that validates on `build()`, and holds the deletion-vector offset and size as `u64`. The planner builds delete files through the builder and rejects a negative offset or size, the way it already rejects a negative record count. `IcebergScanExec::fill_delete_file_sizes` sizes a delete file by rebuilding it, as it already does for the task. The JVM serde already sends every field the builder requires for a deletion vector.
- apache/iceberg-rust#3323 fixes iceberg-rust's `day` for pre-epoch timestamps (apache/iceberg-rust#3315). Comet computes the time-transform partition values of timestamps itself, so nothing it writes changes. The canary test that pins iceberg-rust's values now expects the fixed `day`. iceberg-rust still differs from iceberg-java for a timestamp exactly 999999 microseconds into a unit (#6426), so Comet keeps computing these values.
- Docs: the user guide no longer lists the fallback. The contributor guide and the Iceberg write review skill keep #6138 as an example of iceberg-rust's value semantics, and now note the upstream fix.
The rest of the range doesn't need Comet changes:
- apache-avro 0.22 for manifest reading (apache/iceberg-rust#3354).
- Typed NULLs for nested columns added by schema evolution (apache/iceberg-rust#2635).
- Page-index pruning for `FIXED_LEN_BYTE_ARRAY` decimals (apache/iceberg-rust#3328).
- An Azure path fix (apache/iceberg-rust#3311).
- REST catalog sessions, which Comet doesn't use.
- Dependency and CI bumps.
The nested schema evolution guard from #6504 stays, because it waits on apache/iceberg-rust#3255, which is still open.
## How are these changes tested?
- `CometIcebergWriteActionSuite` "signed-zero float and double identity partitions match iceberg-java" now requires the native writer to run.
- Every row goes through one task, so the zeros meet in one writer. The fanout writer gets them interleaved, and the clustered writer gets one row per partition.
- It compares against iceberg-java three things: the partition directories, the committed partition values, and the rows a pruned read returns for each zero.
- Against a `libcomet` built from `main` with the fallback removed, the test fails: both writers put every row under `f=-0.0/d=0.0`. With this pin it passes.
- "float and double partition paths match iceberg-java" now checks the native writer's directories.
- `CometIcebergWriteDetectionSuite` replaces four fallback tests with one that checks that float, double and nested double identity partitions are `Compatible`.
- New Rust tests cover the planner's delete-file validation and the scan's delete-file rebuild.
- Locally, on the default Spark 4.1 profile (Iceberg 1.11), all of these pass:
- all 718 `datafusion-comet` lib tests;
- clippy, with the same flags as CI;
- the 14 Iceberg JVM suites that don't need MinIO: 419 passed, 0 failed, and 1 canceled (the unknown-transform test, which is canceled on this profile on `main` too).
A pin bump can change file bytes and manifests, and the other Spark profiles pin Iceberg 1.5.2, 1.8.1 and 1.10.0. So this PR is labelled `run-iceberg-tests` and `run-all-spark-profiles`.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#issuecomment-6110921047
@zhangfengcdt you added the fallback for float and double identity partitions in #6531, and this PR lifts it. If you have time, would you be able to review it? Thanks.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4260658500
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -2172,51 +2172,74 @@ class CometIcebergWriteActionSuite
}
}
- // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compares float partition
- // values with an equality that treats -0.0 and 0.0 as one value, so its writers filed both under
- // whichever arrived first, and a read that pruned on the other value lost rows. iceberg-java
- // keeps the two apart, so these writes fall back until iceberg-rust does too.
- test("signed-zero float and double identity partitions match iceberg-java") {
+ // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compared float partition
+ // values with an equality that treated -0.0 and 0.0 as one value, so its writers filed both
+ // under whichever arrived first, and a read that pruned on the other value lost rows. Since
+ // apache/iceberg-rust#3327 it keeps them apart and puts every NaN in one partition, as
+ // iceberg-java does. One shuffle partition puts every row in one task, so the values meet in one
+ // writer. The fanout writer gets them interleaved. The clustered writer gets each partition's
+ // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats could come back
+ // unclustered, which iceberg-java rejects too.
Review Comment:
Thanks, looks good to me!
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4237796386
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -1500,95 +1500,25 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- test("fall-back: identity partition on a float or double column") {
- // iceberg-rust groups float partition values with an equality that treats -0.0 and 0.0 as one
- // value, where iceberg-java keeps them apart (#6138).
+ test("Compatible when partitioned by identity on a float or double column") {
+ // iceberg-rust grouped float partition values with an equality that treated -0.0 and 0.0 as
+ // one value, so these writes fell back (#6138) until apache/iceberg-rust#3327 kept the two
+ // apart as iceberg-java does. A nested source field resolves the same way.
withDetectionCatalog { _ =>
- Seq("float" -> "FLOAT", "double" -> "DOUBLE").foreach { case (typeName, sqlType) =>
- val table = s"part_$typeName"
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.$table (id INT, v $sqlType)
- USING iceberg PARTITIONED BY (v)
- """)
- val writeExec = captureWriteExec(table, allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, CAST(1.5 AS $sqlType))")
- }
- assertUnsupportedContains(writeExec, table, "partition field v", typeName, "-0.0")
- }
- }
- }
-
- test("fall-back: identity partition on a nested double field") {
- // The source of a partition field can be nested inside a struct. `Schema.findField` resolves
- // a nested id too, so the rule must not fail open for it.
- withDetectionCatalog { _ =>
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.part_nested (id INT, s STRUCT<v: DOUBLE>)
- USING iceberg PARTITIONED BY (s.v)
- """)
- val writeExec = captureWriteExec("part_nested", allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_nested VALUES (1, named_struct('v', 1.5D))")
- }
- assertUnsupportedContains(writeExec, "part_nested", "partition field s.v", "double", "-0.0")
- }
- }
-
- test("fall-back: double identity partition beside a dropped partition field") {
- // A format-version-1 spec keeps a dropped partition field as a `void` transform, and that
- // field's source column can be dropped afterwards. The surviving double field must still be
- // found, whatever the dropped one does to the spec's partition type.
- withDetectionCatalog { dir =>
- createTable(
- dir,
- "part_dropped",
- partitionSpec = "PARTITIONED BY (region, amount)",
- properties = Some("'format-version'='1'"))
- // Loaded afresh for each change: the insert in between commits through another handle.
- def table: org.apache.iceberg.Table =
- loadIcebergTable(spark, catalog, ns, "part_dropped")
- .asInstanceOf[org.apache.iceberg.Table]
- table.updateSpec().removeField("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- assertUnsupportedContains("part_dropped", "partition field amount", "double", "-0.0")
-
- // Iceberg before 1.11 cannot plan a write once the `void` field's source column is gone.
- // On 1.11 iceberg-java plans it but cannot build a partition key for a spec that mixes that
- // field with a live one, so the write itself fails on either path; only the gate's decision
- // is checked.
- if (icebergVersionAtLeast(1, 11)) {
- table.updateSchema().deleteColumn("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- val writeExec = captureWriteExec("part_dropped", allowWriteFailure = true) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_dropped VALUES (2, 2.0)")
- }
- assertUnsupportedContains(
- writeExec,
- "part_dropped",
- "partition field amount",
- "double",
- "-0.0")
+ Seq(
+ "part_float" -> ("v FLOAT", "v", "CAST(1.5 AS FLOAT)"),
Review Comment:
[P2] Construct these entries as explicit nested tuples. The three new `"part_*" -> (...)` expressions trigger Scala 2.12’s adapted-argument warning because `->` accepts one argument and the compiler implicitly creates a three-tuple. The supported Spark 3.5 strict-warnings build promotes this warning to an error, so `test-compile` fails before tests can run. Using `("part_float", ("v FLOAT", "v", "CAST(1.5 AS FLOAT)"))`, and the equivalent form for the other entries, avoids the adaptation and restores compilation.
Evidence: Exact-head CI job https://github.com/apache/datafusion-comet/actions/runs/38055061563/job/114222370917 fails at lines 1509–1511 with `Adapting argument list by creating a 3-tuple` while running `./mvnw -B test-compile -Pspark-3.5 -Pstrict-warnings -DskipTests`. Independently compiling the three entries with Scala 2.12.18 and `-Xlint:adapted-args -Xfatal-warnings` exited 1. The explicit nested-tuple version exited 0.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4287946288
##########
native/core/src/execution/operators/iceberg_partition_value.rs:
##########
@@ -39,7 +39,7 @@ use iceberg::{Error, ErrorKind, Result};
/// `timestamp`, or `timestamptz` source, which go through Comet's `iceberg_years` /
/// `iceberg_months` / `iceberg_days` / `iceberg_hours` kernels instead: the ones the sort in front
/// of a clustered write runs, pinned against iceberg-java's `DateTimeUtil` over the whole domain.
-/// iceberg-rust's transforms differ from iceberg-java's in three ways:
+/// iceberg-rust's transforms differ from iceberg-java's in two ways:
Review Comment:
Yes, dropped in f6c69ce4a0. The bullet now lists only the 999999-microsecond difference (#6426), which is still open in the pin range.
##########
native/core/src/execution/planner.rs:
##########
@@ -8213,6 +8216,54 @@ mod tests {
assert_eq!(fields[1].name, "category");
}
+ /// iceberg-rust holds a deletion vector's blob coordinates as `u64`, so the planner rejects a
+ /// negative one from the proto rather than build the delete file with it.
+ #[test]
+ fn test_delete_file_pool_rejects_negative_coordinates() {
+ let deletion_vector = spark_operator::IcebergDeleteFile {
+ file_path_idx: 0,
+ content_type: "POSITION_DELETES".to_string(),
+ file_format: "PUFFIN".to_string(),
+ referenced_data_file: Some("file:///tmp/data.parquet".to_string()),
+ content_offset: Some(4),
+ content_size_in_bytes: Some(40),
+ record_count: Some(2),
+ ..Default::default()
+ };
+ let parse = |delete: spark_operator::IcebergDeleteFile| {
+ let proto_common = spark_operator::IcebergScanCommon {
+ delete_file_path_pool: vec!["file:///tmp/dv.puffin".to_string()],
+ delete_file_pool: vec![delete],
+ ..Default::default()
+ };
+ parse_file_scan_tasks_from_common(&proto_common, &[]).map(|_| ())
+ };
+
+ assert!(parse(deletion_vector.clone()).is_ok());
+ for (delete, expected) in [
+ (
+ spark_operator::IcebergDeleteFile {
+ content_offset: Some(-1),
+ ..deletion_vector.clone()
+ },
+ "negative content offset",
+ ),
+ (
+ spark_operator::IcebergDeleteFile {
+ content_size_in_bytes: Some(-1),
+ ..deletion_vector.clone()
+ },
+ "negative content size",
+ ),
+ ] {
Review Comment:
Added in f6c69ce4a0. The test is renamed to `test_delete_file_pool_rejects_invalid_deletion_vectors` and now has a negative record count and a missing record count. The missing one asserts on iceberg-rust's `is missing record_count` message, so a change in what the builder validates fails here.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4237914628
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -1500,95 +1500,25 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- test("fall-back: identity partition on a float or double column") {
- // iceberg-rust groups float partition values with an equality that treats -0.0 and 0.0 as one
- // value, where iceberg-java keeps them apart (#6138).
+ test("Compatible when partitioned by identity on a float or double column") {
+ // iceberg-rust grouped float partition values with an equality that treated -0.0 and 0.0 as
+ // one value, so these writes fell back (#6138) until apache/iceberg-rust#3327 kept the two
+ // apart as iceberg-java does. A nested source field resolves the same way.
withDetectionCatalog { _ =>
- Seq("float" -> "FLOAT", "double" -> "DOUBLE").foreach { case (typeName, sqlType) =>
- val table = s"part_$typeName"
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.$table (id INT, v $sqlType)
- USING iceberg PARTITIONED BY (v)
- """)
- val writeExec = captureWriteExec(table, allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, CAST(1.5 AS $sqlType))")
- }
- assertUnsupportedContains(writeExec, table, "partition field v", typeName, "-0.0")
- }
- }
- }
-
- test("fall-back: identity partition on a nested double field") {
- // The source of a partition field can be nested inside a struct. `Schema.findField` resolves
- // a nested id too, so the rule must not fail open for it.
- withDetectionCatalog { _ =>
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.part_nested (id INT, s STRUCT<v: DOUBLE>)
- USING iceberg PARTITIONED BY (s.v)
- """)
- val writeExec = captureWriteExec("part_nested", allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_nested VALUES (1, named_struct('v', 1.5D))")
- }
- assertUnsupportedContains(writeExec, "part_nested", "partition field s.v", "double", "-0.0")
- }
- }
-
- test("fall-back: double identity partition beside a dropped partition field") {
- // A format-version-1 spec keeps a dropped partition field as a `void` transform, and that
- // field's source column can be dropped afterwards. The surviving double field must still be
- // found, whatever the dropped one does to the spec's partition type.
- withDetectionCatalog { dir =>
- createTable(
- dir,
- "part_dropped",
- partitionSpec = "PARTITIONED BY (region, amount)",
- properties = Some("'format-version'='1'"))
- // Loaded afresh for each change: the insert in between commits through another handle.
- def table: org.apache.iceberg.Table =
- loadIcebergTable(spark, catalog, ns, "part_dropped")
- .asInstanceOf[org.apache.iceberg.Table]
- table.updateSpec().removeField("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- assertUnsupportedContains("part_dropped", "partition field amount", "double", "-0.0")
-
- // Iceberg before 1.11 cannot plan a write once the `void` field's source column is gone.
- // On 1.11 iceberg-java plans it but cannot build a partition key for a spec that mixes that
- // field with a live one, so the write itself fails on either path; only the gate's decision
- // is checked.
- if (icebergVersionAtLeast(1, 11)) {
- table.updateSchema().deleteColumn("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- val writeExec = captureWriteExec("part_dropped", allowWriteFailure = true) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_dropped VALUES (2, 2.0)")
- }
- assertUnsupportedContains(
- writeExec,
- "part_dropped",
- "partition field amount",
- "double",
- "-0.0")
+ Seq(
+ "part_float" -> ("v FLOAT", "v", "CAST(1.5 AS FLOAT)"),
Review Comment:
Thanks. Fixed in 1036ae5d6f, and 3ffae5ad24 then switched this test to the suite's `createTable` table, so the tuples are gone. `Strict Scala warnings (Spark 3.5, JDK 17)` passes on 3ffae5ad24.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4296383523
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -342,19 +342,41 @@ impl IcebergScanExec {
let deletes = task
.deletes()
.iter()
- .map(|delete| {
- let mut delete = delete.clone();
- if let Some(&size) = size_map.get(&delete.file_path) {
- delete.file_size_in_bytes = size;
- }
- delete
+ .map(|delete| match size_map.get(delete.file_path()) {
+ Some(&size) => Self::sized_delete_file(delete, size),
+ None => Ok(delete.clone()),
})
- .collect::<Vec<_>>();
+ .collect::<Result<Vec<_>, Error>>()?;
*task = Self::rebuild_task_with_deletes(task, deletes)?;
}
Ok(())
}
+ /// Copies a delete file with its size set to `size`.
+ ///
+ /// iceberg-rust 1f3bc34 made `FileScanTaskDeleteFile`'s fields private too, so, as for the
+ /// task in [`Self::rebuild_task_with_deletes`], the copy goes through the builder, forwarding
+ /// every other field through its public accessors. The builder repeats the delete file's
+ /// validation on `build()`.
Review Comment:
Should this doc describe the API as it is rather than the revision that changed it? Once the pin moves past `1f3bc34`, the hash only sends a reader off to look up a commit. The [`rebuild_task_with_deletes` doc](https://github.com/apache/datafusion-comet/blob/a66d0e5ed77b680534e1e38d16a236b6009399cb/native/core/src/execution/operators/iceberg_scan.rs#L382-L384) right below already reads that way:
```suggestion
/// `FileScanTaskDeleteFile`'s fields are private with no mutator, so, as for the task in
/// [`Self::rebuild_task_with_deletes`], the copy goes through the builder, forwarding every
/// other field through its public accessors. The builder repeats the delete file's validation
/// on `build()`.
```
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4260845967
##########
native/core/src/execution/operators/iceberg_partition_value.rs:
##########
@@ -39,7 +39,7 @@ use iceberg::{Error, ErrorKind, Result};
/// `timestamp`, or `timestamptz` source, which go through Comet's `iceberg_years` /
/// `iceberg_months` / `iceberg_days` / `iceberg_hours` kernels instead: the ones the sort in front
/// of a clustered write runs, pinned against iceberg-java's `DateTimeUtil` over the whole domain.
-/// iceberg-rust's transforms differ from iceberg-java's in three ways:
+/// iceberg-rust's transforms differ from iceberg-java's in two ways:
Review Comment:
This drops the `day` difference now that apache/iceberg-rust#3323 is in the pin range. Should the "Partition values" bullet in the contributor guide drop it too? It still says iceberg-rust's `day` "puts some timestamps from the last second before a pre-1970 midnight in the next day" ([`iceberg-writes.md` lines 247-251](https://github.com/apache/datafusion-comet/blob/4eae2874fe160ad53800c089d6d647913a982a72/docs/source/contributor-guide/iceberg-writes.md?plain=1#L247-L251)).
##########
native/core/src/execution/planner.rs:
##########
@@ -8213,6 +8216,54 @@ mod tests {
assert_eq!(fields[1].name, "category");
}
+ /// iceberg-rust holds a deletion vector's blob coordinates as `u64`, so the planner rejects a
+ /// negative one from the proto rather than build the delete file with it.
+ #[test]
+ fn test_delete_file_pool_rejects_negative_coordinates() {
+ let deletion_vector = spark_operator::IcebergDeleteFile {
+ file_path_idx: 0,
+ content_type: "POSITION_DELETES".to_string(),
+ file_format: "PUFFIN".to_string(),
+ referenced_data_file: Some("file:///tmp/data.parquet".to_string()),
+ content_offset: Some(4),
+ content_size_in_bytes: Some(40),
+ record_count: Some(2),
+ ..Default::default()
+ };
+ let parse = |delete: spark_operator::IcebergDeleteFile| {
+ let proto_common = spark_operator::IcebergScanCommon {
+ delete_file_path_pool: vec!["file:///tmp/dv.puffin".to_string()],
+ delete_file_pool: vec![delete],
+ ..Default::default()
+ };
+ parse_file_scan_tasks_from_common(&proto_common, &[]).map(|_| ())
+ };
+
+ assert!(parse(deletion_vector.clone()).is_ok());
+ for (delete, expected) in [
+ (
+ spark_operator::IcebergDeleteFile {
+ content_offset: Some(-1),
+ ..deletion_vector.clone()
+ },
+ "negative content offset",
+ ),
+ (
+ spark_operator::IcebergDeleteFile {
+ content_size_in_bytes: Some(-1),
+ ..deletion_vector.clone()
+ },
+ "negative content size",
+ ),
+ ] {
Review Comment:
The refactor moves the record count check into `non_negative` too, and the new `.build()` call now rejects a deletion vector that lacks a field, but neither case is in this test. Could we add a negative record count and a missing record count here? The second one also checks that the builder's error reaches the caller through the `Invalid Iceberg delete file` mapping, so a change in what iceberg-rust's builder validates fails here instead of at read time. If you add these, `test_delete_file_pool_rejects_invalid_deletion_vectors` might fit the test better than `..._negative_coordinates`.
```suggestion
(
spark_operator::IcebergDeleteFile {
content_size_in_bytes: Some(-1),
..deletion_vector.clone()
},
"negative content size",
),
(
spark_operator::IcebergDeleteFile {
record_count: Some(-1),
..deletion_vector.clone()
},
"negative record count",
),
(
spark_operator::IcebergDeleteFile {
record_count: None,
..deletion_vector.clone()
},
"is missing record_count",
),
] {
```
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4260583848
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -2172,51 +2172,74 @@ class CometIcebergWriteActionSuite
}
}
- // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compares float partition
- // values with an equality that treats -0.0 and 0.0 as one value, so its writers filed both under
- // whichever arrived first, and a read that pruned on the other value lost rows. iceberg-java
- // keeps the two apart, so these writes fall back until iceberg-rust does too.
- test("signed-zero float and double identity partitions match iceberg-java") {
+ // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compared float partition
+ // values with an equality that treated -0.0 and 0.0 as one value, so its writers filed both
+ // under whichever arrived first, and a read that pruned on the other value lost rows. Since
+ // apache/iceberg-rust#3327 it keeps them apart and puts every NaN in one partition, as
+ // iceberg-java does. One shuffle partition puts every row in one task, so the values meet in one
+ // writer. The fanout writer gets them interleaved. The clustered writer gets each partition's
+ // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats could come back
+ // unclustered, which iceberg-java rejects too.
Review Comment:
Added in 4eae2874fe as "clustered writer rejects interleaved signed zeros like the JVM writer". It hands the clustered writer `-0.0`, `0.0` and `-0.0` in one task, and both writers have to fail with the same `IllegalStateException`, at the same depth in the cause chain and with the same message, which names the `f=-0.0/d=0.0` partition.
I used `failedWrites` and `clusteredWriterError` rather than `assertFailsLikeSpark`, as the clustered-writer test for #5698 does. `assertFailsLikeSpark` compares Spark's write with the split plan, but it doesn't check that the native writer ran, and it only compares exception classes. `failedWrites` fails unless the native write went through `CometIcebergWriteExec`. The rows also skip Spark's sort, so the order the writers see doesn't depend on how the sort orders the two tied zeros. Against a libcomet built from main, the native write succeeds and the test fails.
--
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: bump iceberg-rust and write float and double identity partitions natively [datafusion-comet]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4244539655
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -2172,51 +2172,74 @@ class CometIcebergWriteActionSuite
}
}
- // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compares float partition
- // values with an equality that treats -0.0 and 0.0 as one value, so its writers filed both under
- // whichever arrived first, and a read that pruned on the other value lost rows. iceberg-java
- // keeps the two apart, so these writes fall back until iceberg-rust does too.
- test("signed-zero float and double identity partitions match iceberg-java") {
+ // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compared float partition
+ // values with an equality that treated -0.0 and 0.0 as one value, so its writers filed both
+ // under whichever arrived first, and a read that pruned on the other value lost rows. Since
+ // apache/iceberg-rust#3327 it keeps them apart and puts every NaN in one partition, as
+ // iceberg-java does. One shuffle partition puts every row in one task, so the values meet in one
+ // writer. The fanout writer gets them interleaved. The clustered writer gets each partition's
+ // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats could come back
+ // unclustered, which iceberg-java rejects too.
Review Comment:
Could we pin this with assertFailsLikeSpark so both writers are shown to fail with the same exception here? Before this PR, the native writer merged the zeros silently, so it might be worth a test that it now fails the same way.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -1500,92 +1500,13 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- test("fall-back: identity partition on a float or double column") {
- // iceberg-rust groups float partition values with an equality that treats -0.0 and 0.0 as one
- // value, where iceberg-java keeps them apart (#6138).
- withDetectionCatalog { _ =>
- Seq("float" -> "FLOAT", "double" -> "DOUBLE").foreach { case (typeName, sqlType) =>
- val table = s"part_$typeName"
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.$table (id INT, v $sqlType)
- USING iceberg PARTITIONED BY (v)
- """)
- val writeExec = captureWriteExec(table, allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, CAST(1.5 AS $sqlType))")
- }
- assertUnsupportedContains(writeExec, table, "partition field v", typeName, "-0.0")
- }
- }
- }
-
- test("fall-back: identity partition on a nested double field") {
- // The source of a partition field can be nested inside a struct. `Schema.findField` resolves
- // a nested id too, so the rule must not fail open for it.
- withDetectionCatalog { _ =>
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.part_nested (id INT, s STRUCT<v: DOUBLE>)
- USING iceberg PARTITIONED BY (s.v)
- """)
- val writeExec = captureWriteExec("part_nested", allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_nested VALUES (1, named_struct('v', 1.5D))")
- }
- assertUnsupportedContains(writeExec, "part_nested", "partition field s.v", "double", "-0.0")
- }
- }
-
- test("fall-back: double identity partition beside a dropped partition field") {
- // A format-version-1 spec keeps a dropped partition field as a `void` transform, and that
- // field's source column can be dropped afterwards. The surviving double field must still be
- // found, whatever the dropped one does to the spec's partition type.
- withDetectionCatalog { dir =>
- createTable(
- dir,
- "part_dropped",
- partitionSpec = "PARTITIONED BY (region, amount)",
- properties = Some("'format-version'='1'"))
- // Loaded afresh for each change: the insert in between commits through another handle.
- def table: org.apache.iceberg.Table =
- loadIcebergTable(spark, catalog, ns, "part_dropped")
- .asInstanceOf[org.apache.iceberg.Table]
- table.updateSpec().removeField("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- assertUnsupportedContains("part_dropped", "partition field amount", "double", "-0.0")
-
- // Iceberg before 1.11 cannot plan a write once the `void` field's source column is gone.
- // On 1.11 iceberg-java plans it but cannot build a partition key for a spec that mixes that
- // field with a live one, so the write itself fails on either path; only the gate's decision
- // is checked.
- if (icebergVersionAtLeast(1, 11)) {
- table.updateSchema().deleteColumn("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- val writeExec = captureWriteExec("part_dropped", allowWriteFailure = true) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_dropped VALUES (2, 2.0)")
- }
- assertUnsupportedContains(
- writeExec,
- "part_dropped",
- "partition field amount",
- "double",
- "-0.0")
- }
- }
- }
-
- test("Compatible when a dropped double partition field remains as void") {
- // The `void` field only ever holds null, so there are no signed zeros to keep apart.
+ test("Compatible when partitioned by identity on a double column") {
Review Comment:
Thanks for lifting this, and the signed-zero test is really thorough.
--
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