From 19ece8657c2b6f42d51f0c3256fbaaa1ad201c8e Mon Sep 17 00:00:00 2001
From: Au_Miner <358671982@qq.com>
Date: Thu, 11 Jun 2026 11:25:14 +0800
Subject: [PATCH] [FLINK-35276][table] Fix incorrect sort order for
 floating-point negative zero in generated comparators

---
 .../table/planner/codegen/GenerateUtils.scala |  6 ++-
 .../codegen/SortCodeGeneratorTest.java        | 40 ++++++++++++++++++-
 2 files changed, 44 insertions(+), 2 deletions(-)

diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
index 9cfae27fe44fe..15555b95d4675 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
@@ -632,7 +632,11 @@ object GenerateUtils {
       val sortUtil =
         classOf[org.apache.flink.table.runtime.operators.sort.SortUtil].getCanonicalName
       s"$sortUtil.compareBinary($leftTerm, $rightTerm)"
-    case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | DATE | TIME_WITHOUT_TIME_ZONE |
+    case FLOAT =>
+      s"Float.compare($leftTerm, $rightTerm)"
+    case DOUBLE =>
+      s"Double.compare($leftTerm, $rightTerm)"
+    case TINYINT | SMALLINT | INTEGER | BIGINT | DATE | TIME_WITHOUT_TIME_ZONE |
         INTERVAL_YEAR_MONTH | INTERVAL_DAY_TIME =>
       s"($leftTerm > $rightTerm ? 1 : $leftTerm < $rightTerm ? -1 : 0)"
     case TIMESTAMP_WITH_TIME_ZONE | MULTISET | MAP | VARIANT | BITMAP =>
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/SortCodeGeneratorTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/SortCodeGeneratorTest.java
index 0dbf1c098860f..fcceaa9dda5a9 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/SortCodeGeneratorTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/SortCodeGeneratorTest.java
@@ -119,6 +119,7 @@ class SortCodeGeneratorTest {
 
     private RowType inputType;
     private SortSpec sortSpec;
+    private BinaryRowData[] customTestData;
 
     private static final DataType INT_ROW_TYPE =
             DataTypes.ROW(DataTypes.FIELD("f0", DataTypes.INT())).bridgedTo(Row.class);
@@ -136,6 +137,26 @@ class SortCodeGeneratorTest {
                     .createComparator(
                             new int[] {0}, new boolean[] {true}, 0, new ExecutionConfig());
 
+    @Test
+    void testFloatNegativeZeroSortKey() throws Exception {
+        inputType =
+                RowType.of(
+                        new TypeInformationRawType<>(Types.INT),
+                        new FloatType(),
+                        new BooleanType());
+        sortSpec =
+                SortSpec.builder()
+                        .addField(0, true, SortUtil.getNullDefaultOrder(true))
+                        .addField(1, true, SortUtil.getNullDefaultOrder(true))
+                        .build();
+        customTestData = buildNegativeZeroFloatTestData();
+        try {
+            testInner();
+        } finally {
+            customTestData = null;
+        }
+    }
+
     @Test
     void testMultiKeys() throws Exception {
         for (int i = 0; i < 100; i++) {
@@ -212,7 +233,22 @@ private BinaryRowData row(int i, Object[][] values) {
         return row;
     }
 
+    private BinaryRowData[] buildNegativeZeroFloatTestData() {
+        Object[][] values = new Object[3][];
+        values[0] = new Object[] {RawValueData.fromObject(0), RawValueData.fromObject(0)};
+        values[1] = new Object[] {0.0f, -0.0f};
+        values[2] = new Object[] {false, true};
+        BinaryRowData[] result = new BinaryRowData[2];
+        for (int i = 0; i < result.length; i++) {
+            result[i] = row(i, values);
+        }
+        return result;
+    }
+
     private BinaryRowData[] getTestData() {
+        if (customTestData != null) {
+            return customTestData;
+        }
         BinaryRowData[] result = new BinaryRowData[RECORD_NUM];
         Object[][] values = new Object[inputType.getFieldCount()][];
         for (int i = 0; i < inputType.getFieldCount(); i++) {
@@ -477,7 +513,9 @@ private void testInner() throws Exception {
 
         List<BinaryRowData> data = Arrays.asList(dataArray.clone());
         List<BinaryRowData> binaryRows = Arrays.asList(dataArray.clone());
-        Collections.shuffle(binaryRows);
+        if (customTestData == null) {
+            Collections.shuffle(binaryRows);
+        }
 
         for (BinaryRowData row : binaryRows) {
             if (!sortBuffer.write(row)) {