diff --git a/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java b/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java index 241dc6100379..21f725015edb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java +++ b/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java @@ -160,7 +160,7 @@ public HProcessFunction visit(BooleanType booleanType) { if (row.isNullAt(fieldIndex)) { return PRIMITIVE_EMPTY; } - return row.getBoolean(fieldIndex) ? PRIMITIVE_EMPTY : 0; + return row.getBoolean(fieldIndex) ? 1L : 0L; }; } diff --git a/paimon-common/src/test/java/org/apache/paimon/sort/hilbert/HilbertIndexerTest.java b/paimon-common/src/test/java/org/apache/paimon/sort/hilbert/HilbertIndexerTest.java new file mode 100644 index 000000000000..238e80cbb5f6 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/sort/hilbert/HilbertIndexerTest.java @@ -0,0 +1,69 @@ +/* + * 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.sort.hilbert; + +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.jupiter.api.Test; + +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link HilbertIndexer}. */ +public class HilbertIndexerTest { + + @Test + public void testBooleanValuesDistinctFromNull() { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.BOOLEAN(), DataTypes.BOOLEAN()}, + new String[] {"a", "b"}); + HilbertIndexer indexer = new HilbertIndexer(rowType, Arrays.asList("a", "b")); + indexer.open(); + + // FALSE, TRUE and NULL are the only three states a boolean column has, and each has to + // land on its own point of the curve. Pinning the exact curve position of each one also + // pins the mapping itself (0 / 1 / the null sentinel), so an inverted mapping that keeps + // the three distinct cannot slip through and desync this from the Spark UDF. + byte[] falseIndex = indexer.index(booleanRow(false)); + byte[] trueIndex = indexer.index(booleanRow(true)); + byte[] nullIndex = indexer.index(booleanRow(null)); + + assertThat(falseIndex).isEqualTo(HilbertIndexer.hilbertCurvePosBytes(new Long[] {0L, 0L})); + assertThat(trueIndex).isEqualTo(HilbertIndexer.hilbertCurvePosBytes(new Long[] {1L, 1L})); + assertThat(nullIndex) + .isEqualTo( + HilbertIndexer.hilbertCurvePosBytes( + new Long[] {Long.MAX_VALUE, Long.MAX_VALUE})); + assertThat(trueIndex).isNotEqualTo(nullIndex); + assertThat(falseIndex).isNotEqualTo(nullIndex); + assertThat(falseIndex).isNotEqualTo(trueIndex); + } + + private static GenericRow booleanRow(Boolean value) { + GenericRow row = new GenericRow(2); + row.setField(0, value); + row.setField(1, value); + return row; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java index ebd72312440c..971e8342e5a6 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java @@ -161,7 +161,14 @@ private UserDefinedFunction doubleToOrderedLongUDF() { private UserDefinedFunction booleanToOrderedLongUDF() { UserDefinedFunction udf = functions - .udf((Boolean value) -> value ? PRIMITIVE_EMPTY : 0, DataTypes.LongType) + .udf( + (Boolean value) -> { + if (value == null) { + return PRIMITIVE_EMPTY; + } + return value ? 1L : 0L; + }, + DataTypes.LongType) .withName("BOOLEAN-LEXICAL-BYTES"); return udf; } diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkHilbertUDFTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkHilbertUDFTest.java new file mode 100644 index 000000000000..d96b68388d89 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/sort/SparkHilbertUDFTest.java @@ -0,0 +1,88 @@ +/* + * 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.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 SparkHilbertUDF}. */ +public class SparkHilbertUDFTest { + + @Test + void testBooleanColumnMapsNullFalseAndTrueToDistinctValues() { + SparkSession spark = + SparkSession.builder() + .master("local[1]") + .appName("spark-hilbert-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); + + SparkHilbertUDF udf = new SparkHilbertUDF(); + List rows = + df.select( + df.col("a"), + udf.sortedLexicographically(df.col("a"), DataTypes.BooleanType) + .as("hilbert")) + .collectAsList(); + + Map mapped = new HashMap<>(); + for (Row row : rows) { + assertThat(row.isNullAt(1)).isFalse(); + mapped.put(row.isNullAt(0) ? null : row.getBoolean(0), row.getLong(1)); + } + + // A null boolean must reach the sentinel without unboxing, and TRUE must not share + // it: Long.MAX_VALUE is what every type in this class uses for null. + assertThat(mapped.get(null)).isEqualTo(Long.MAX_VALUE); + assertThat(mapped.get(Boolean.TRUE)).isEqualTo(1L); + assertThat(mapped.get(Boolean.FALSE)).isEqualTo(0L); + } finally { + spark.stop(); + SparkSession.clearActiveSession(); + SparkSession.clearDefaultSession(); + } + } +}