You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/02 19:30:54 UTC
(datafusion-comet) 01/02: fix: fall back for Iceberg writes with float or double identity partitions (#6531)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit f57b232ad051d5d296512823bdef21822bce0ced
Author: Feng Zhang <fe...@apache.org>
AuthorDate: Fri Oct 2 18:20:24 2026 +0000
fix: fall back for Iceberg writes with float or double identity partitions (#6531)
* fix: fall back for Iceberg writes with float or double identity partitions
iceberg-rust's writers group partition values with an equality that
treats -0.0 and 0.0 as one value, so the native writer filed both under
one partition and a pruned read lost rows. Decline these writes until
apache/iceberg-rust#3325 is resolved.
Closes #6138
* fix: resolve partition field types by source id and address review
Resolve each partition field's type from its source id instead of by
position, cover a dropped partition field beside a live double field,
link apache/iceberg-rust#3325, and update the docs wording.
* test: cover an identity partition on a nested double field
Schema.findField resolves nested source ids too, so the rule declines a
float or double inside a struct.
---
docs/source/user-guide/latest/iceberg-writes.md | 14 +++-
.../apache/comet/iceberg/IcebergReflection.scala | 40 ++++++++++
.../serde/operator/CometIcebergNativeWrite.scala | 24 ++++++
.../comet/CometIcebergWriteActionSuite.scala | 48 +++++++++++-
.../comet/CometIcebergWriteDetectionSuite.scala | 89 ++++++++++++++++++++++
5 files changed, 209 insertions(+), 6 deletions(-)
diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md
index 049c868109..d1b14c1e71 100644
--- a/docs/source/user-guide/latest/iceberg-writes.md
+++ b/docs/source/user-guide/latest/iceberg-writes.md
@@ -176,7 +176,7 @@ A write is eligible only when ALL of the following hold:
| `write.target-file-size-bytes` | any value (the two writers can choose different roll points; see accepted divergences) [...]
| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs` (`gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below) [...]
| resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` [...]
-| partition spec | any [...]
+| partition spec | any, except an identity partition on a `float` or `double` column (see below) [...]
| column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) [...]
Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` —
@@ -205,6 +205,13 @@ could resolve a different storage identity or endpoint than the JVM writer would
combination falls back; a `GCSFileIO` carries its `gcs.*` settings in `FileIO.properties()`,
which are forwarded.
+An identity partition on a `float` or `double` column falls back. iceberg-rust compares float
+partition values with an equality that treats `-0.0` and `0.0` as one value, so the native writer
+would put rows with either value in the same partition, where iceberg-java writes two. A read
+that prunes on the other value's partition would then miss rows. The fall-back stays until
+iceberg-rust distinguishes the two values
+([apache/iceberg-rust#3325](https://github.com/apache/iceberg-rust/issues/3325)).
+
Other `write.*` properties are intentionally not gated because they cannot make the native
writer produce different data files: distribution and ordering settings shape the Spark plan
identically on both paths, WAP / branch / snapshot properties act on the JVM committer,
@@ -298,8 +305,9 @@ a data file but not what any reader computes from it:
(iceberg-java names files `<partition>-<task>-<operation>-<count>`; iceberg-rust uses a
process-local counter).
- Partition directory names match iceberg-java 1.8+'s `PartitionSpec.partitionToPath` for every
- partition type. `float` and `double` values are rendered with Java's `Float.toString` /
- `Double.toString` rules (`f=1.0`, `f=1.0E10`), as iceberg-java renders them. On Iceberg 1.5.x,
+ partition type the native writer accepts. Identity partitions on `float` and `double` columns
+ fall back (see [Native Parquet write eligibility](#native-parquet-write-eligibility)), so
+ iceberg-java names those directories itself. On Iceberg 1.5.x,
which the Spark 3.4 profile pins, iceberg-java itself spelled `timestamp` and `timestamptz`
directories with `LocalDateTime.toString()` / `OffsetDateTime.toString()`
(`ts=1969-12-31T23:59:58.500Z`) and left the partition field name unescaped; Comet uses the
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index a96dc952db..37a13c7674 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -764,6 +764,46 @@ object IcebergReflection extends Logging {
.toSeq
}
+ /**
+ * The partition fields of `spec` whose source column is a `float` or `double`, as (partition
+ * field name, Iceberg type name). Only the identity transform applies to those types, so such a
+ * field holds the column's own values. A `void` field is skipped: it only ever holds null, and
+ * its source column may no longer exist. Each field is resolved through its own `sourceId`
+ * rather than by position in `partitionType()`. Throws on reflection failure, or when a field
+ * that is not `void` has no source column, so the caller can fail closed.
+ */
+ def floatingPointPartitionFields(spec: Any): Seq[(String, String)] = {
+ import scala.jdk.CollectionConverters._
+ val schema = getMethod(spec.getClass, "schema").invoke(spec)
+ val findField = getMethod(schema.getClass, "findField", classOf[Int])
+ getMethod(spec.getClass, "fields")
+ .invoke(spec)
+ .asInstanceOf[java.util.List[_]]
+ .asScala
+ .flatMap { partitionField =>
+ val transform =
+ getMethod(partitionField.getClass, "transform").invoke(partitionField).toString
+ if (transform == "void") {
+ None
+ } else {
+ val name =
+ getMethod(partitionField.getClass, "name").invoke(partitionField).asInstanceOf[String]
+ val sourceId =
+ getMethod(partitionField.getClass, "sourceId")
+ .invoke(partitionField)
+ .asInstanceOf[Int]
+ val source = findField.invoke(schema, sourceId.asInstanceOf[Object])
+ if (source == null) {
+ throw new IllegalStateException(
+ s"partition field $name has no source column with id $sourceId")
+ }
+ val sourceType = getMethod(source.getClass, "type").invoke(source).toString
+ if (sourceType == "float" || sourceType == "double") Some(name -> sourceType) else None
+ }
+ }
+ .toSeq
+ }
+
/**
* Gets the partition spec from an Iceberg table.
*/
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
index 3954e2c67b..4303851df8 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
@@ -176,6 +176,7 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
requireDefaultLocationProvider,
requireFormatVersionAtMostTwo,
requireNoUuidColumns,
+ requireNoFloatingPointPartitionField,
requireNoEncryptionPrefix,
requireNoBloomFilterColumnsEnabled,
requireRowGroupCheckMinRecordCountAtDefault,
@@ -268,6 +269,29 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
}
}
+ // iceberg-rust holds a float partition value as an `OrderedFloat`, whose equality treats -0.0
+ // and 0.0 as one value, and its fanout and clustered writers group rows by that equality.
+ // iceberg-java keeps the two apart, so the native writer would file both under whichever
+ // arrived first, and a read that prunes on the other value would lose rows (#6138). Remove this
+ // rule once the iceberg-rust pin carries a fix for apache/iceberg-rust#3325; #5643 tracks it.
+ private val requireNoFloatingPointPartitionField: TriggerRule = ctx =>
+ IcebergReflection
+ .getOutputSpecIdFromSparkWrite(ctx.sparkWrite)
+ .flatMap(IcebergReflection.getPartitionSpecById(ctx.table, _)) match {
+ case None => Some("could not resolve the output partition spec for type checking")
+ case Some(spec) =>
+ try {
+ IcebergReflection.floatingPointPartitionFields(spec).headOption.map {
+ case (name, typeName) =>
+ s"partition field $name has Iceberg type $typeName, and the native writer does " +
+ "not keep -0.0 and 0.0 partitions apart"
+ }
+ } catch {
+ case e: Exception =>
+ Some(s"could not inspect the output partition spec: ${e.getMessage}")
+ }
+ }
+
private val requireNoEncryptionPrefix: TriggerRule = ctx =>
ctx.properties.keys
.find(_.startsWith(EncryptionPropertyPrefix))
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
index d99691599b..e8aa36e2c5 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
@@ -1336,10 +1336,52 @@ 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") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ Seq("zero_native", "zero_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, f FLOAT, d DOUBLE)
+ USING iceberg PARTITIONED BY (f, d)
+ """)
+ }
+ val values =
+ "(1, CAST('-0.0' AS FLOAT), 0.0D), " +
+ "(2, CAST(0.0 AS FLOAT), 0.0D), " +
+ "(3, CAST(0.0 AS FLOAT), CAST('-0.0' AS DOUBLE))"
+
+ assertNativeWriteDoesNotEngage("zero_native", Seq(1, 2, 3)) {
+ spark.sql(s"INSERT INTO $catalog.$ns.zero_native VALUES $values")
+ }
+ spark.sql(s"INSERT INTO $catalog.$ns.zero_jvm VALUES $values")
+
+ val dirs = partitionDirs(warehouseDir, "zero_native")
+ assert(dirs == partitionDirs(warehouseDir, "zero_jvm"), s"native enabled: $dirs")
+ assert(dirs == Set("f=-0.0/d=0.0", "f=0.0/d=0.0", "f=0.0/d=-0.0"))
+
+ // Iceberg prunes on the partition value, so a merged partition changes what a read returns.
+ def idsWhereDIsZero(table: String): Seq[Int] =
+ spark
+ .sql(s"SELECT id FROM $catalog.$ns.$table WHERE d = 0.0D")
+ .collect()
+ .map(_.getInt(0))
+ .sorted
+ .toSeq
+ assert(idsWhereDIsZero("zero_native") == idsWhereDIsZero("zero_jvm"))
+ }
+ }
+
// iceberg-java renders a `float` or `double` partition value with `Float.toString` /
// `Double.toString`. Rust's `Display` spelled `Double.MAX_VALUE` as 309 digits instead, past the
- // 255-byte limit on one path component (apache/datafusion-comet#5836).
- test("native acceleration: float and double partition paths match iceberg-java") {
+ // 255-byte limit on one path component (apache/datafusion-comet#5836). Float and double identity
+ // partitions now fall back (#6138), so this checks that the fallback keeps iceberg-java's
+ // layout; the unit tests in `iceberg_partition_path.rs` cover how the native writer renders
+ // these values.
+ test("float and double partition paths match iceberg-java") {
assumeNativeAcceleration()
withIcebergCatalog { warehouseDir =>
Seq("float_path_native", "float_path_jvm").foreach { table =>
@@ -1356,7 +1398,7 @@ class CometIcebergWriteActionSuite
"(3, CAST(3.4028235E38 AS FLOAT), CAST(4.9E-324 AS DOUBLE)), " +
"(4, CAST(0.001 AS FLOAT), CAST(1.0E20 AS DOUBLE))"
- assertNativeWriteEngages("float_path_native", Seq(1, 2, 3, 4)) {
+ assertNativeWriteDoesNotEngage("float_path_native", Seq(1, 2, 3, 4)) {
spark.sql(s"INSERT INTO $catalog.$ns.float_path_native VALUES $values")
}
spark.sql(s"INSERT INTO $catalog.$ns.float_path_jvm VALUES $values")
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index 2e2cc3f234..8034ef9675 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -884,6 +884,95 @@ 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.
+ withDetectionCatalog { dir =>
+ createTable(
+ dir,
+ "part_void",
+ partitionSpec = "PARTITIONED BY (amount)",
+ properties = Some("'format-version'='1'"))
+ loadIcebergTable(spark, catalog, ns, "part_void")
+ .asInstanceOf[org.apache.iceberg.Table]
+ .updateSpec()
+ .removeField("amount")
+ .commit()
+ spark.sql(s"REFRESH TABLE $catalog.$ns.part_void")
+ assertSupportLevelIs[Compatible]("part_void")
+ }
+ }
+
test("fall-back: uuid column in the write schema") {
withDetectionCatalog { dir =>
// Spark DDL cannot declare `uuid`, so evolve the schema through the Iceberg API. Spark
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org