This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5af0f51cfe4 [FLINK-40693][table] Align floating-point `IN` predicates
with equality for signed zero
5af0f51cfe4 is described below
commit 5af0f51cfe4ba089f4df28852d30b67e0ca2127d
Author: ParyshevSergey <[email protected]>
AuthorDate: Mon Sep 28 13:48:56 2026 +0700
[FLINK-40693][table] Align floating-point `IN` predicates with equality for
signed zero
---
.../planner/expressions/ScalarOperatorsTest.scala | 35 +++++++++++++
.../planner/runtime/batch/sql/CalcITCase.scala | 46 ++++++++++++++++
.../planner/runtime/stream/sql/CalcITCase.scala | 61 +++++++++++++++++++++-
.../runtime/util/collections/DoubleHashSet.java | 8 +--
.../runtime/util/collections/FloatHashSet.java | 8 +--
5 files changed, 149 insertions(+), 9 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
index 44fac362f8f..5d22b4e757c 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
@@ -21,6 +21,8 @@ import org.apache.flink.table.api.{DataTypes,
LiteralStringExpression, Unresolve
import org.apache.flink.table.planner.expressions.utils.ScalarOperatorsTestBase
import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
class ScalarOperatorsTest extends ScalarOperatorsTestBase {
@@ -62,6 +64,39 @@ class ScalarOperatorsTest extends ScalarOperatorsTestBase {
)
}
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = Array("FLOAT", "DOUBLE"))
+ def testInWithSignedZero(numericType: String): Unit = {
+ // Negate the runtime zero so constant folding cannot erase its sign.
+ val zero = s"CAST(f7 AS $numericType)"
+ val negativeZero = s"CAST(-f7 AS $numericType)"
+ val pair = s"CAST(0 AS $numericType), CAST(2 AS $numericType)"
+ val negativePair = s"CAST('-0.0' AS $numericType), CAST(2 AS $numericType)"
+ val longList = (Seq(0) ++ (2 to 21)).mkString(", ")
+
+ testSqlApi(s"$negativeZero = $zero", "TRUE")
+ testSqlApi(s"$negativeZero IN (0)", "TRUE")
+ testSqlApi(s"$negativeZero IN ($pair)", "TRUE")
+ testSqlApi(s"$negativeZero NOT IN ($pair)", "FALSE")
+ testSqlApi(s"$zero IN ($negativePair)", "TRUE")
+ testSqlApi(s"$zero NOT IN ($negativePair)", "FALSE")
+ testSqlApi(s"$negativeZero IN (CAST(0 AS $numericType), $negativePair)",
"TRUE")
+ testSqlApi(s"$negativeZero IN ($longList)", "TRUE")
+ testSqlApi(s"$negativeZero NOT IN ($longList)", "FALSE")
+ testSqlApi(s"$negativeZero IN ($pair, NULL)", "TRUE")
+ testSqlApi(s"$negativeZero NOT IN ($pair, NULL)", "FALSE")
+ testSqlApi(s"$negativeZero IN (1, 2)", "FALSE")
+ testSqlApi(s"$negativeZero NOT IN (1, 2)", "TRUE")
+ testSqlApi(s"$negativeZero IN (1, 2, NULL)", "NULL")
+ testSqlApi(s"$negativeZero NOT IN (1, 2, NULL)", "NULL")
+ testSqlApi(s"CAST(f4 AS $numericType) IN ($pair)", "FALSE")
+ testSqlApi(s"CAST(f4 AS $numericType) NOT IN ($pair)", "TRUE")
+ testSqlApi(s"CAST(f14 AS $numericType) IN ($pair)", "NULL")
+ testSqlApi(s"CAST(f14 AS $numericType) NOT IN ($pair)", "NULL")
+ testSqlApi(s"(CAST(f14 AS $numericType) IN ($pair)) IS TRUE", "FALSE")
+ testSqlApi(s"(CAST(f14 AS $numericType) IN ($pair)) IS NOT FALSE", "TRUE")
+ }
+
@Test
def testCompareOperator(): Unit = {
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index 1ffedf2b0ae..909a66ea90a 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -1706,6 +1706,52 @@ class CalcITCase extends BatchTestBase {
)
}
+ @Test
+ def testFloatingPointInWithSignedZero(): Unit = {
+ registerCollection(
+ "SignedZeros",
+ Seq(
+ row(1, -0.0f, -0.0d),
+ row(1, -0.0f, -0.0d),
+ row(2, 0.0f, 0.0d),
+ row(3, 1.0f, 1.0d),
+ row(4, null, null),
+ row(5, 2.0f, 2.0d),
+ row(6, Float.NaN, Double.NaN),
+ row(7, Float.PositiveInfinity, Double.PositiveInfinity),
+ row(8, Float.NegativeInfinity, Double.NegativeInfinity)
+ ),
+ new RowTypeInfo(Types.INT, Types.FLOAT, Types.DOUBLE),
+ "id, f, d"
+ )
+
+ checkResult(
+ "SELECT id FROM SignedZeros WHERE f IN (0, 2) AND d IN (0, 2)",
+ Seq(row(1), row(1), row(2), row(5)))
+
+ for (field <- Seq("f", "d")) {
+ checkResult(
+ s"""
+ |SELECT id, $field = 0,
+ | $field IN (0, 2), $field NOT IN (0, 2),
+ | $field IN (0, 2, NULL), $field NOT IN (0, 2, NULL)
+ |FROM SignedZeros
+ |""".stripMargin,
+ Seq(
+ row(1, true, true, false, true, false),
+ row(1, true, true, false, true, false),
+ row(2, true, true, false, true, false),
+ row(3, false, false, true, null, null),
+ row(4, null, null, null, null, null),
+ row(5, false, true, false, true, false),
+ row(6, false, false, true, null, null),
+ row(7, false, false, true, null, null),
+ row(8, false, false, true, null, null)
+ )
+ )
+ }
+ }
+
@Test
def testSearch(): Unit = {
val myTableDataId = TestValuesTableFactory.registerData(
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
index a8c90fe77b8..f885f582a28 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
@@ -39,7 +39,7 @@ import org.apache.flink.test.util.TestBaseUtils
import org.apache.flink.types.Row
import org.apache.flink.util.CollectionUtil
-import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
+import org.assertj.core.api.Assertions.{assertThat, assertThatList,
assertThatThrownBy}
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.extension.RegisterExtension
@@ -348,6 +348,65 @@ class CalcITCase extends StreamingTestBase {
assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
}
+ @Test
+ def testFloatingPointInWithSignedZero(): Unit = {
+ val dataId = TestValuesTableFactory.registerData(
+ Seq(
+ row(1, -0.0f, -0.0d),
+ row(1, -0.0f, -0.0d),
+ row(2, 0.0f, 0.0d),
+ row(3, 1.0f, 1.0d),
+ row(4, null, null),
+ row(5, 2.0f, 2.0d),
+ row(6, Float.NaN, Double.NaN),
+ row(7, Float.PositiveInfinity, Double.PositiveInfinity),
+ row(8, Float.NegativeInfinity, Double.NegativeInfinity)
+ ))
+ tEnv.executeSql(s"""
+ |CREATE TABLE SignedZeros (id INT, f FLOAT, d DOUBLE)
WITH (
+ | 'connector' = 'values',
+ | 'data-id' = '$dataId',
+ | 'bounded' = 'true'
+ |)
+ |""".stripMargin)
+
+ val filtered = tEnv
+ .executeSql("SELECT id FROM SignedZeros WHERE f IN (0, 2) AND d IN (0,
2)")
+ .collect()
+ try {
+ assertThatList(CollectionUtil.iteratorToList(filtered))
+ .containsExactlyInAnyOrder(row(1), row(1), row(2), row(5))
+ } finally {
+ filtered.close()
+ }
+
+ for (field <- Seq("f", "d")) {
+ val projected = tEnv
+ .executeSql(s"""
+ |SELECT id, $field = 0,
+ | $field IN (0, 2), $field NOT IN (0, 2),
+ | $field IN (0, 2, NULL), $field NOT IN (0, 2, NULL)
+ |FROM SignedZeros
+ |""".stripMargin)
+ .collect()
+ try {
+
assertThatList(CollectionUtil.iteratorToList(projected)).containsExactlyInAnyOrder(
+ row(1, true, true, false, true, false),
+ row(1, true, true, false, true, false),
+ row(2, true, true, false, true, false),
+ row(3, false, false, true, null, null),
+ row(4, null, null, null, null, null),
+ row(5, false, true, false, true, false),
+ row(6, false, false, true, null, null),
+ row(7, false, false, true, null, null),
+ row(8, false, false, true, null, null)
+ )
+ } finally {
+ projected.close()
+ }
+ }
+ }
+
@Test
def testLongProjectionList(): Unit = {
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/DoubleHashSet.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/DoubleHashSet.java
index bce2d931b5f..d32931a0bc9 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/DoubleHashSet.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/DoubleHashSet.java
@@ -37,9 +37,9 @@ public class DoubleHashSet extends OptimizableHashSet {
this(DEFAULT_INITIAL_SIZE, DEFAULT_LOAD_FACTOR);
}
- /** See {@link Double#equals(Object)}. */
+ /** See {@link Double#equals(Object)}, except that both signs of zero are
considered equal. */
public boolean add(final double k) {
- long longKey = Double.doubleToLongBits(k);
+ final long longKey = k == 0.0d ? 0L : Double.doubleToLongBits(k);
if (longKey == 0L) {
if (this.containsZero) {
return false;
@@ -75,9 +75,9 @@ public class DoubleHashSet extends OptimizableHashSet {
return true;
}
- /** See {@link Double#equals(Object)}. */
+ /** See {@link Double#equals(Object)}, except that both signs of zero are
considered equal. */
public boolean contains(final double k) {
- long longKey = Double.doubleToLongBits(k);
+ final long longKey = k == 0.0d ? 0L : Double.doubleToLongBits(k);
if (longKey == 0L) {
return this.containsZero;
} else {
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/FloatHashSet.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/FloatHashSet.java
index 4f21c5859d8..b5634abc1ed 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/FloatHashSet.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/collections/FloatHashSet.java
@@ -37,9 +37,9 @@ public class FloatHashSet extends OptimizableHashSet {
this(DEFAULT_INITIAL_SIZE, DEFAULT_LOAD_FACTOR);
}
- /** See {@link Float#equals(Object)}. */
+ /** See {@link Float#equals(Object)}, except that both signs of zero are
considered equal. */
public boolean add(final float k) {
- int intKey = Float.floatToIntBits(k);
+ final int intKey = k == 0.0f ? 0 : Float.floatToIntBits(k);
if (intKey == 0) {
if (this.containsZero) {
return false;
@@ -73,9 +73,9 @@ public class FloatHashSet extends OptimizableHashSet {
return true;
}
- /** See {@link Float#equals(Object)}. */
+ /** See {@link Float#equals(Object)}, except that both signs of zero are
considered equal. */
public boolean contains(final float k) {
- int intKey = Float.floatToIntBits(k);
+ final int intKey = k == 0.0f ? 0 : Float.floatToIntBits(k);
if (intKey == 0) {
return this.containsZero;
} else {
