This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2c2684a898 [common][spark] Fix hilbert sort keys for date, float and
double columns (#10554)
2c2684a898 is described below
commit 2c2684a898e2ed3f1f19e4a2565cd0fb081de718
Author: YangJie <[email protected]>
AuthorDate: Thu Oct 15 20:52:14 2026 -0400
[common][spark] Fix hilbert sort keys for date, float and double columns
(#10554)
---
.../apache/paimon/sort/hilbert/HilbertIndexer.java | 16 +++++-
.../paimon/sort/hilbert/HilbertIndexerTest.java | 50 +++++++++++++++++++
.../SortCompactActionForAppendTableITCase.java | 58 ++++++++++++++++++++++
.../apache/paimon/spark/sort/SparkHilbertUDF.java | 10 ++--
.../paimon/spark/sort/SparkHilbertUDFTest.java | 48 ++++++++++++++++++
.../spark/procedure/CompactProcedureTestBase.scala | 53 ++++++++++++++++++++
6 files changed, 230 insertions(+), 5 deletions(-)
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 b4d67e48ca..7fb2cc9fc2 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
@@ -93,6 +93,18 @@ public class HilbertIndexer implements Serializable {
functionSet.forEach(RowProcessor::open);
}
+ /**
+ * Maps a double to the curve's 63-bit unsigned domain, keeping its order.
The curve only reads
+ * the low 63 bits of each coordinate, so the raw bits would drop the sign
and put {@code x} and
+ * {@code -x} on the same point. Floats widen to double without changing
their order.
+ */
+ public static long doubleToOrderedLong(double value) {
+ long bits = Double.doubleToLongBits(value);
+ // flip all bits of negatives and only the sign bit of positives, as
in z-order
+ bits ^= (bits >> (Long.SIZE - 1)) | Long.MIN_VALUE;
+ return bits >>> 1;
+ }
+
public byte[] index(InternalRow row) {
Long[] columnLongs = new Long[fieldsIndex.length];
@@ -221,7 +233,7 @@ public class HilbertIndexer implements Serializable {
return (row) ->
row.isNullAt(fieldIndex)
? PRIMITIVE_EMPTY
- :
Double.doubleToLongBits(row.getFloat(fieldIndex));
+ : doubleToOrderedLong(row.getFloat(fieldIndex));
}
@Override
@@ -229,7 +241,7 @@ public class HilbertIndexer implements Serializable {
return (row) ->
row.isNullAt(fieldIndex)
? PRIMITIVE_EMPTY
- :
Double.doubleToLongBits(row.getDouble(fieldIndex));
+ : doubleToOrderedLong(row.getDouble(fieldIndex));
}
@Override
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
index f547edb2e1..74d9ffdcae 100644
---
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
@@ -152,6 +152,56 @@ public class HilbertIndexerTest {
}
}
+ /**
+ * The curve covers one half of the plane before the other, so sorted by
key one of the two
+ * coordinates changes sign exactly once. A key that ignores the sign bit
puts {@code x} and
+ * {@code -x} on the same point and interleaves both signs.
+ */
+ @Test
+ public void testFloatingPointKeysKeepTheSign() {
+ double[] values = {-4.5, -3.5, -2.5, -1.5, 1.5, 2.5, 3.5, 4.5};
+ for (DataType type : new DataType[] {DataTypes.DOUBLE(),
DataTypes.FLOAT()}) {
+ RowType rowType = RowType.of(new DataType[] {type, type}, new
String[] {"x", "y"});
+ HilbertIndexer indexer = new HilbertIndexer(rowType,
Arrays.asList("x", "y"));
+ indexer.open();
+
+ List<double[]> points = new ArrayList<>();
+ for (double x : values) {
+ for (double y : values) {
+ points.add(new double[] {x, y});
+ }
+ }
+ points.sort(
+ (left, right) ->
+ compareUnsigned(
+ indexer.index(row(type, left)),
+ indexer.index(row(type, right))));
+
+ assertThat(indexer.index(row(type, new double[] {1.5, 2.5})))
+ .as("%s", type)
+ .isNotEqualTo(indexer.index(row(type, new double[] {-1.5,
-2.5})));
+ assertThat(Math.min(signChanges(points, 0), signChanges(points,
1)))
+ .as("%s", type)
+ .isEqualTo(1);
+ }
+ }
+
+ private static int signChanges(List<double[]> points, int column) {
+ int changes = 0;
+ for (int i = 1; i < points.size(); i++) {
+ if ((points.get(i - 1)[column] < 0) != (points.get(i)[column] <
0)) {
+ changes++;
+ }
+ }
+ return changes;
+ }
+
+ private static GenericRow row(DataType type, double[] point) {
+ return type.equals(DataTypes.FLOAT())
+ ? GenericRow.of((float) point[0], (float) point[1])
+ : GenericRow.of(point[0], point[1]);
+ }
+
private static BigInteger index(Long[] points) {
long[] data =
Arrays.stream(points).mapToLong(Long::longValue).toArray();
return HilbertCurve.bits(63).dimensions(points.length).index(data);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
index d871c13910..15be7fbde8 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForAppendTableITCase.java
@@ -280,6 +280,64 @@ public class SortCompactActionForAppendTableITCase extends
ActionITCaseBase {
.isLessThan(filesFilterOrder.size() / (double)
filesOrder.size());
}
+ @Test
+ public void testHilbertKeepsNegativeAndPositiveDoublesApart() throws
Exception {
+ catalog.createDatabase(database, true);
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("x", DataTypes.DOUBLE())
+ .column("y", DataTypes.DOUBLE())
+ .option("bucket", "-1")
+ .option("sink.parallelism", "1")
+ .build(),
+ true);
+ // written with alternating signs, so a single sign change can only
come from the sort
+ double[] values = {-4.5, 4.5, -3.5, 3.5, -2.5, 2.5, -1.5, 1.5};
+ BatchWriteBuilder builder = getTable().newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite()) {
+ for (double x : values) {
+ for (double y : values) {
+ write.write(GenericRow.of(x, y));
+ }
+ }
+ commit(write.prepareCommit());
+ }
+
+ hilbert(Arrays.asList("x", "y"));
+
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
+ Assertions.assertThat(files).hasSize(1);
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(files.get(0).partition())
+ .withBucket(files.get(0).bucket())
+
.withDataFiles(Collections.singletonList(files.get(0).file()))
+ .withBucketPath("not used")
+ .build();
+ List<double[]> rows = new ArrayList<>();
+ getTable()
+ .newReadBuilder()
+ .newRead()
+ .createReader(dataSplit)
+ .forEachRemaining(
+ row -> rows.add(new double[] {row.getDouble(0),
row.getDouble(1)}));
+ Assertions.assertThat(rows).hasSize(64);
+
+ // The curve visits one half of the plane before the other, so along
the clustered order
+ // one of the two coordinates changes sign exactly once. A key that
ignores the sign puts
+ // x and -x on the same point and interleaves both signs.
+ int[] signChanges = new int[2];
+ for (int i = 1; i < rows.size(); i++) {
+ for (int c = 0; c < 2; c++) {
+ if ((rows.get(i - 1)[c] < 0) != (rows.get(i)[c] < 0)) {
+ signChanges[c]++;
+ }
+ }
+ }
+ Assertions.assertThat(Math.min(signChanges[0],
signChanges[1])).isEqualTo(1);
+ }
+
@Test
public void testTableConf() throws Exception {
createTable();
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 0e60000389..b28aaadb16 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
@@ -43,6 +43,7 @@ import org.apache.spark.sql.types.TimestampType;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
+import java.sql.Date;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
@@ -149,7 +150,7 @@ public class SparkHilbertUDF implements Serializable {
if (value == null) {
return PRIMITIVE_EMPTY;
}
- return Double.doubleToLongBits((double)
value);
+ return
HilbertIndexer.doubleToOrderedLong(value);
},
DataTypes.LongType)
.withName("FLOAT_ORDERED_BYTES");
@@ -165,7 +166,7 @@ public class SparkHilbertUDF implements Serializable {
if (value == null) {
return PRIMITIVE_EMPTY;
}
- return Double.doubleToLongBits(value);
+ return
HilbertIndexer.doubleToOrderedLong(value);
},
DataTypes.LongType)
.withName("DOUBLE_ORDERED_BYTES");
@@ -247,7 +248,10 @@ public class SparkHilbertUDF implements Serializable {
} else if (type instanceof DecimalType) {
return
longToOrderedLongUDF().apply(column.cast(DataTypes.LongType));
} else if (type instanceof DateType) {
- return
longToOrderedLongUDF().apply(column.cast(DataTypes.LongType));
+ // A date does not cast to bigint: ANSI mode rejects the cast and
legacy mode turns
+ // every date into null. Take the days since epoch instead, as the
indexer does.
+ Column epochDay = functions.datediff(column,
functions.lit(Date.valueOf("1970-01-01")));
+ return
longToOrderedLongUDF().apply(epochDay.cast(DataTypes.LongType));
} else {
throw new IllegalArgumentException(
String.format(
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
index abb24eb731..9b7f49114d 100644
---
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
@@ -28,6 +28,7 @@ import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
+import java.sql.Date;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
@@ -142,4 +143,51 @@ public class SparkHilbertUDFTest {
SparkSession.clearDefaultSession();
}
}
+
+ @Test
+ void testDateColumnMapsToEpochDay() {
+ for (String ansi : new String[] {"true", "false"}) {
+ SparkSession spark =
+ SparkSession.builder()
+ .master("local[1]")
+ .appName("spark-hilbert-udf-date-test")
+ .config("spark.ui.enabled", "false")
+ .config("spark.sql.ansi.enabled", ansi)
+ .getOrCreate();
+ try {
+ StructType schema =
+ new StructType(
+ new StructField[] {
+ new StructField("a", DataTypes.DateType,
true, Metadata.empty())
+ });
+ Dataset<Row> df =
+ spark.createDataFrame(
+ Arrays.asList(
+
RowFactory.create(Date.valueOf("0001-01-01")),
+
RowFactory.create(Date.valueOf("1969-12-31")),
+
RowFactory.create(Date.valueOf("1970-01-01")),
+
RowFactory.create(Date.valueOf("9999-12-31")),
+ RowFactory.create((Date) null)),
+ schema);
+
+ SparkHilbertUDF udf = new SparkHilbertUDF();
+ List<Long> mapped = new ArrayList<>();
+ for (Row row :
+ df.select(udf.sortedLexicographically(df.col("a"),
DataTypes.DateType))
+ .collectAsList()) {
+ mapped.add(row.getLong(0));
+ }
+
+ // The same days since epoch the indexer reads from a DATE
field. Casting the date
+ // to bigint failed analysis under ANSI and gave null, i.e.
the sentinel, otherwise.
+ assertThat(mapped)
+ .as("ansi=%s", ansi)
+ .containsExactly(-719162L, -1L, 0L, 2932896L,
Long.MAX_VALUE);
+ } finally {
+ spark.stop();
+ SparkSession.clearActiveSession();
+ SparkSession.clearDefaultSession();
+ }
+ }
+ }
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 6d3997633a..9364ad824b 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -1532,6 +1532,59 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
}
}
+ test("Paimon Procedure: hilbert sort compact on a date column") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, dt DATE)
+ |TBLPROPERTIES ('bucket'='-1')
+ |""".stripMargin)
+ spark.sql("INSERT INTO T VALUES (1, DATE '2024-01-03'), (2, DATE
'1969-12-31'), (3, null)")
+ spark.sql("INSERT INTO T VALUES (4, DATE '2024-01-01')")
+
+ withSparkSQLConf("spark.sql.ansi.enabled" -> "true") {
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.compact(table => 'T', order_strategy =>
'hilbert', order_by => 'dt,id')"),
+ Row(true) :: Nil)
+ }
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY id"),
+ Row(1, java.sql.Date.valueOf("2024-01-03")) ::
+ Row(2, java.sql.Date.valueOf("1969-12-31")) ::
+ Row(3, null) ::
+ Row(4, java.sql.Date.valueOf("2024-01-01")) :: Nil
+ )
+ }
+ }
+
+ test("Paimon Procedure: hilbert sort compact keeps negative and positive
doubles apart") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (x DOUBLE, y DOUBLE)
+ |TBLPROPERTIES ('bucket'='-1')
+ |""".stripMargin)
+ val values = Seq(-4.5, -3.5, -2.5, -1.5, 1.5, 2.5, 3.5, 4.5)
+ val points = for (x <- values; y <- values) yield s"($x, $y)"
+ spark.sql(s"INSERT INTO T VALUES ${points.mkString(", ")}")
+
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.compact(table => 'T', order_strategy => 'hilbert',
order_by => 'x,y')"),
+ Row(true) :: Nil)
+
+ // The curve visits one half of the plane before the other, so along the
clustered order
+ // one of the two coordinates changes sign exactly once. A key that
ignores the sign puts
+ // x and -x on the same point and interleaves both signs.
+ val rows = spark.sql("SELECT x, y FROM T").collect()
+ Assertions.assertThat(rows.length).isEqualTo(64)
+ def signChanges(column: Int): Int =
+ rows
+ .sliding(2)
+ .count(pair => (pair(0).getDouble(column) < 0) !=
(pair(1).getDouble(column) < 0))
+ Assertions.assertThat(math.min(signChanges(0),
signChanges(1))).isEqualTo(1)
+ }
+ }
+
test("Paimon Procedure: incremental clustering with local-sort mode") {
withTable("T") {
spark.sql(s"""
