[common][spark] Fix boolean hilbert value colliding with the null sentinel

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;
             };
         }
 

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;
+    }
+}

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;
     }

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<Row> df =
+                    spark.createDataFrame(
+                            Arrays.asList(
+                                    RowFactory.create(true),
+                                    RowFactory.create(false),
+                                    RowFactory.create((Boolean) null)),
+                            schema);
+
+            SparkHilbertUDF udf = new SparkHilbertUDF();
+            List<Row> rows =
+                    df.select(
+                                    df.col("a"),
+                                    udf.sortedLexicographically(df.col("a"), DataTypes.BooleanType)
+                                            .as("hilbert"))
+                            .collectAsList();
+
+            Map<Boolean, Long> 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();
+        }
+    }
+}