GGraziadei commented on code in PR #17669:
URL: https://github.com/apache/iceberg/pull/17669#discussion_r3972430790
##########
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:
Fixed, thanks. Out of curiosity for future changes: since the session is
shared across tests, would it make sense to reset the SQL conf in an
`@AfterEach` in `TestBase`, so that a test forgetting to use `withSQLConf`
cannot leak config into the others?
--
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]
