huaxingao commented on code in PR #17669:
URL: https://github.com/apache/iceberg/pull/17669#discussion_r3974703022
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2209,6 +2212,105 @@ public void testZOrderUDFWithTimestampNTZType() {
assertThat(zorderBytes).isNotNull().isNotEmpty();
}
+ @TestTemplate
+ public void zOrderUDFEncodesNullValuesAsZeroBytes() {
+ Object[][] nullsByType = {
+ {"CAST(NULL AS BOOLEAN)", DataTypes.BooleanType},
+ {"CAST(NULL AS TINYINT)", DataTypes.ByteType},
+ {"CAST(NULL AS SMALLINT)", DataTypes.ShortType},
+ {"CAST(NULL AS INT)", DataTypes.IntegerType},
+ {"CAST(NULL AS BIGINT)", DataTypes.LongType},
+ {"CAST(NULL AS FLOAT)", DataTypes.FloatType},
+ {"CAST(NULL AS DOUBLE)", DataTypes.DoubleType},
+ {"CAST(NULL AS DATE)", DataTypes.DateType},
+ {"CAST(NULL AS TIMESTAMP)", DataTypes.TimestampType},
+ {"CAST(NULL AS TIMESTAMP_NTZ)", DataTypes.TimestampNTZType},
+ {"CAST(NULL AS STRING)", DataTypes.StringType},
+ {"CAST(NULL AS BINARY)", DataTypes.BinaryType},
+ };
+
+ for (Object[] nullByType : nullsByType) {
+ String literal = (String) nullByType[0];
+ DataType type = (DataType) nullByType[1];
+ SparkZOrderUDF zorderUDF = new SparkZOrderUDF(1, 16, 1024);
+ Dataset<Row> result =
+ spark
+ .sql("SELECT " + literal + " as test_col")
+ .withColumn(
+ "zorder_result",
zorderUDF.sortedLexicographically(col("test_col"), type));
+
+ byte[] zorderBytes =
result.collectAsList().get(0).getAs("zorder_result");
+ assertThat(zorderBytes)
+ .as("A null %s must produce ordered bytes rather than failing",
type.simpleString())
+ .isNotNull()
+ .isNotEmpty();
+ assertThat(zorderBytes)
+ .as("A null %s must sort lowest, as an all-zero byte sequence",
type.simpleString())
+ .containsOnly((byte) 0);
+ }
+ }
+
+ @TestTemplate
+ public void zOrderSortWithNullBooleanValues() {
+ Schema schema =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "flag", Types.BooleanType.get()));
+ Table table =
+ TABLES.create(
+ schema,
+ PartitionSpec.unpartitioned(),
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion)),
+ tableLocation);
+
+ for (int batch = 0; batch < 2; batch++) {
+ spark
+ .createDataFrame(
+ Lists.newArrayList(
+ RowFactory.create(1, true),
+ RowFactory.create(2, null),
+ RowFactory.create(3, false)),
+ SparkSchemaUtil.convert(schema))
+ .write()
+ .format("iceberg")
+ .mode("append")
+ .save(tableLocation);
+ }
+ table.refresh();
+
+ long dataSizeBefore = testDataSize(table);
+ RewriteDataFiles.Result result =
+ basicRewrite(table)
+ .zOrder("id", "flag")
+ .option(SizeBasedFileRewritePlanner.MIN_INPUT_FILES, "1")
+ .execute();
+
+ assertThat(result.rewrittenBytesCount()).isEqualTo(dataSizeBefore);
+ assertThat(result.rewrittenDataFilesCount()).isGreaterThan(0);
+ assertThat(spark.read().format("iceberg").load(tableLocation).filter("flag
IS NULL").count())
+ .isEqualTo(2);
+ }
+
+ @TestTemplate
+ public void zOrderSortWithMismatchedColumnCase() {
+ assertThat(spark.conf().get("spark.sql.caseSensitive"))
+ .as("This test covers the case-insensitive column resolution path")
+ .isEqualTo("false");
Review Comment:
I would not reset config in `@AfterEach` because it runs after every test
method, so a
reset hook would wipe per-class setup. This class is a good example --
`setupSpark()`
disables AQE in `@BeforeAll` because "tests assume that writes generate a
particular
number of files". Reset after each test and AQE comes back on for every
method after the
first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [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]
