diff --git a/paimon-common/src/main/java/org/apache/paimon/sort/zorder/ZIndexer.java b/paimon-common/src/main/java/org/apache/paimon/sort/zorder/ZIndexer.java index 1d40fe75e776..1f21811ed391 100644 --- a/paimon-common/src/main/java/org/apache/paimon/sort/zorder/ZIndexer.java +++ b/paimon-common/src/main/java/org/apache/paimon/sort/zorder/ZIndexer.java @@ -209,7 +209,8 @@ public ZProcessFunction visit(BooleanType booleanType) { return NULL_BYTES; } ZOrderByteUtils.reuse(reuse, PRIMITIVE_BUFFER_SIZE); - reuse.put(0, (byte) (row.getBoolean(fieldIndex) ? -127 : 0)); + // FALSE must not encode to the all-zero NULL_BYTES sentinel. + reuse.put(0, (byte) (row.getBoolean(fieldIndex) ? -127 : 1)); return reuse.array(); }; } diff --git a/paimon-common/src/test/java/org/apache/paimon/sort/zorder/TestZOrderByteUtil.java b/paimon-common/src/test/java/org/apache/paimon/sort/zorder/TestZOrderByteUtil.java index bd59230ae682..cfab2a7b7de2 100644 --- a/paimon-common/src/test/java/org/apache/paimon/sort/zorder/TestZOrderByteUtil.java +++ b/paimon-common/src/test/java/org/apache/paimon/sort/zorder/TestZOrderByteUtil.java @@ -18,15 +18,22 @@ package org.apache.paimon.sort.zorder; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + import org.junit.Test; import org.testcontainers.shaded.com.google.common.primitives.UnsignedBytes; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Comparator; import java.util.Random; import static org.apache.paimon.utils.RandomUtil.randomBytes; import static org.apache.paimon.utils.RandomUtil.randomString; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -409,4 +416,35 @@ public void testByteTruncateOrFill() { byteCompare)); } } + + @Test + public void testBooleanDistinctFromNullSentinel() { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.BOOLEAN(), DataTypes.BOOLEAN()}, + new String[] {"a", "b"}); + ZIndexer indexer = new ZIndexer(rowType, Arrays.asList("a", "b")); + indexer.open(); + + byte[] nullBytes = zvalue(indexer, null); + byte[] falseBytes = zvalue(indexer, false); + byte[] trueBytes = zvalue(indexer, true); + + // The three states have to be pairwise distinct, and NULL is the all-zero sentinel, so + // the unsigned order it puts them in is NULL, then FALSE, then TRUE. + Comparator unsigned = UnsignedBytes.lexicographicalComparator(); + assertThat(unsigned.compare(nullBytes, falseBytes)).isNegative(); + assertThat(unsigned.compare(falseBytes, trueBytes)).isNegative(); + assertThat(nullBytes).isNotEqualTo(falseBytes); + assertThat(falseBytes).isNotEqualTo(trueBytes); + assertThat(nullBytes).isNotEqualTo(trueBytes); + } + + /** {@code index()} hands back its internal buffer, so each result is copied out of it. */ + private static byte[] zvalue(ZIndexer indexer, Boolean value) { + GenericRow row = new GenericRow(2); + row.setField(0, value); + row.setField(1, value); + return indexer.index(row).clone(); + } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkZOrderUDF.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkZOrderUDF.java index 65bf62970687..fbd944e9a13d 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkZOrderUDF.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkZOrderUDF.java @@ -254,7 +254,8 @@ private UserDefinedFunction booleanToOrderedBytesUDF() { inputBuffer( position, ZOrderByteUtils.PRIMITIVE_BUFFER_SIZE); - buffer.put(0, (byte) (value ? -127 : 0)); + // FALSE must not encode to the all-zero sentinel. + buffer.put(0, (byte) (value ? -127 : 1)); return buffer.array(); }, DataTypes.BinaryType) diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkZOrderUDFTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkZOrderUDFTest.java new file mode 100644 index 000000000000..ef07b0f0cba8 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkZOrderUDFTest.java @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.sort; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.functions; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link SparkZOrderUDF}. */ +public class SparkZOrderUDFTest { + + @Test + void testBooleanColumnKeepsFalseOffTheNullSentinel() { + SparkSession spark = + SparkSession.builder() + .master("local[1]") + .appName("spark-zorder-udf-test") + .config("spark.ui.enabled", "false") + .getOrCreate(); + try { + StructType schema = + new StructType( + new StructField[] { + new StructField("a", DataTypes.BooleanType, true, Metadata.empty()) + }); + Dataset df = + spark.createDataFrame( + Arrays.asList( + RowFactory.create(true), + RowFactory.create(false), + RowFactory.create((Boolean) null)), + schema); + + SparkZOrderUDF udf = new SparkZOrderUDF(1, 8, Integer.MAX_VALUE); + List rows = + df.select( + df.col("a"), + // The UDF hands back a per-column buffer it reuses for every + // row, so the bytes are turned into hex inside Spark rather + // than collected as arrays that all alias one another. + functions + .hex( + udf.sortedLexicographically( + df.col("a"), DataTypes.BooleanType)) + .as("zvalue")) + .collectAsList(); + + Map mapped = new HashMap<>(); + for (Row row : rows) { + mapped.put(row.isNullAt(0) ? null : row.getBoolean(0), row.getString(1)); + } + + // NULL is the all-zero sentinel, so FALSE has to be something else, and TRUE keeps + // the high bit that puts it above both in unsigned order. + assertThat(mapped.get(null)).isEqualTo("0000000000000000"); + assertThat(mapped.get(Boolean.FALSE)).isEqualTo("0100000000000000"); + assertThat(mapped.get(Boolean.TRUE)).isEqualTo("8100000000000000"); + } finally { + spark.stop(); + SparkSession.clearActiveSession(); + SparkSession.clearDefaultSession(); + } + } +}