andygrove commented on code in PR #5729:
URL: https://github.com/apache/datafusion-comet/pull/5729#discussion_r3944411574
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1033,6 +1033,124 @@ class CometIcebergWriteActionSuite
}
}
+ // iceberg-java renders a `timestamptz` partition value with
+ // `DateTimeUtil.microsToIsoTimestamptz` and a `binary` one with base64;
iceberg-rust's
+ // `partition_to_path` renders the first as a `chrono` `DateTime<Utc>`
(space separator, " UTC"
+ // suffix) -- and outright panics on a pre-1970 value with a sub-second
part, since
+ // `microseconds_to_datetimetz` casts a negative remainder to `u32`
+ // (apache/datafusion-comet#5694) -- and the second as uppercase hex. Comet
renders the path
+ // itself; this pins the result against the layout iceberg-java's own writer
produces.
+ test("native acceleration: timestamptz and binary partition paths match
iceberg-java") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ // A UTC session zone makes the stored micros of each literal exact, so
the expected
+ // directory names below are not a function of the machine's zone.
+ withSQLConf("spark.sql.session.timeZone" -> "UTC") {
+ Seq("ts_path_native", "ts_path_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, ts TIMESTAMP, bin BINARY)
+ USING iceberg PARTITIONED BY (ts, bin)
+ """)
+ }
+ // Pre-epoch with a sub-second part (the panic case), pre-epoch on a
whole second, the
+ // epoch itself, and a post-epoch microsecond value.
+ val values =
+ "(1, TIMESTAMP '1969-12-31 23:59:58.5', X'0001FF'), " +
+ "(2, TIMESTAMP '1969-12-31 23:59:58', X'00'), " +
+ "(3, TIMESTAMP '1970-01-01 00:00:00', X''), " +
+ "(4, TIMESTAMP '2024-04-01 19:25:00.123456', X'FF')"
+
+ assertNativeWriteEngages("ts_path_native", Seq(1, 2, 3, 4)) {
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_native VALUES $values")
+ }
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_jvm VALUES $values")
+
+ val nativeDirs = partitionDirs(warehouseDir, "ts_path_native")
+ assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"),
s"native: $nativeDirs")
+ assert(
+
nativeDirs.contains("ts=1969-12-31T23%3A59%3A58.5%2B00%3A00/bin=AAH%2F"),
+ s"native: $nativeDirs")
+ assert(
+ nativeDirs.contains("ts=1970-01-01T00%3A00%3A00%2B00%3A00/bin="),
+ s"native: $nativeDirs")
+ assert(
+
nativeDirs.contains("ts=2024-04-01T19%3A25%3A00.123456%2B00%3A00/bin=%2Fw%3D%3D"),
+ s"native: $nativeDirs")
+
+ // Values round-trip through both readers regardless of how the path
was spelled.
+ Seq("true", "false").foreach { cometEnabled =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> cometEnabled) {
+ val rows =
+ spark.sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_native ORDER
BY id").collect()
+ assert(
+ rows.toSeq ==
+ spark
+ .sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_jvm ORDER BY
id")
+ .collect()
+ .toSeq,
+ s"comet=$cometEnabled: $rows")
+ }
+ }
+ }
+ }
+ }
+
+ // iceberg-java's `UpdatePartitionSpec` keeps a dropped partition field in a
format-version-1
+ // spec as a `void` transform so its field id survives, and
`PartitionSpec#isUnpartitioned` is
+ // "every field is void", not "no fields". The next write therefore runs
through the
+ // unpartitioned writer -- which stamps an empty partition struct -- against
a spec whose fields
+ // are non-empty, and iceberg-rust's `ManifestWriter` used to `zip_eq` the
two and panic across
+ // the JNI boundary (apache/datafusion-comet#5691). Dropping the source
column afterwards then
+ // broke the manifest's `partition_type` resolution as well
(apache/datafusion-comet#5693).
+ test("native acceleration: writes after a V1 partition field is dropped
match iceberg-java") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("evolved_native", "evolved_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, region STRING)
+ USING iceberg TBLPROPERTIES ('format-version'='1')
+ """)
+ }
+
+ def evolve(table: String, write: (String, Seq[Int]) => Unit): Unit = {
+ write("(1, 'us')", Seq(1))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table ADD PARTITION FIELD region
AS region_part")
+ write("(2, 'eu')", Seq(1, 2))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP PARTITION FIELD
region_part")
+ // The spec is now void-only: the #5691 shape.
+ write("(3, 'ap')", Seq(1, 2, 3))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP COLUMN region")
+ // The void field's source column is gone: the #5693 shape.
+ write("(4)", Seq(1, 2, 3, 4))
Review Comment:
Agreed, and thanks for separating 1.10 out. I'd assumed the `UnknownType`
fallback added to `partitionType` covered it, but `javaClasses` only got the
same fallback in 1.11, so 1.10 gets further and still fails. Since that failure
is in the driver-side commit the native path shares with the stock one, gating
is the only honest option: normalising the transport manifest can't rescue it.
Split the source-column drop into its own test behind
`assume(icebergVersionAtLeast(1, 11))` and left the void-only stages
unconditional, so #5691 itself stays covered on 3.4/3.5/4.0. Verified
`CometIcebergWriteActionSuite` locally on spark-3.4 (Iceberg 1.5.2), spark-3.5
(1.8.1), spark-4.0 (1.10.0) and spark-4.1 (1.11.0), plus the other Iceberg
suites in the `scans` group on 3.4 and 4.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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
