[
https://issues.apache.org/jira/browse/FLINK-40693?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40693:
-----------------------------------
Labels: pull-request-available (was: )
> FLOAT/DOUBLE IN and NOT IN disagree with equality for signed zero
> -----------------------------------------------------------------
>
> Key: FLINK-40693
> URL: https://issues.apache.org/jira/browse/FLINK-40693
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.20.1, 2.3.0, 2.2.1, 2.1.3
> Environment: OpenJDK 21.0.12, Linux amd64, Maven 3.9.16. Batch and
> streaming execution, parallelism 1
> Reporter: Sergey Paryshev
> Priority: Major
> Labels: pull-request-available
>
> SQL equality and constant-list IN predicates produce different results for
> FLOAT and DOUBLE signed zero. A singleton IN predicate behaves like equality,
> whereas an IN predicate using the generated hash-set implementation can
> reject an
> equal value with the opposite zero sign.
> This reproduces with a typed in-memory DataStream, without external
> connectors or file formats.
> Steps to reproduce
> Run the following as a Flink Table/DataStream application with the planner
> available:
> {code:java}
> package org.apache.flink.examples.table;
> import org.apache.flink.api.common.RuntimeExecutionMode;
> import org.apache.flink.api.common.typeinfo.Types;
> import
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.table.api.EnvironmentSettings;
> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
> import org.apache.flink.types.Row;
> import org.apache.flink.util.CloseableIterator;
> import java.util.ArrayList;
> import java.util.Arrays;
> import java.util.Collections;
> import java.util.List;
> public class SignedZeroIn {
> public static void main(String[] args) throws Exception {
> final List<Row> rows = Arrays.asList(
> Row.of(1, -0.0f, -0.0d),
> Row.of(1, -0.0f, -0.0d),
> Row.of(2, 0.0f, 0.0d),
> Row.of(3, 1.0f, 1.0d),
> Row.of(4, null, null),
> Row.of(5, 22.0f, 22.0d),
> Row.of(6, Float.NaN, Double.NaN),
> Row.of(7, Float.POSITIVE_INFINITY,
> Double.POSITIVE_INFINITY),
> Row.of(8, Float.NEGATIVE_INFINITY,
> Double.NEGATIVE_INFINITY));
> if (Float.floatToRawIntBits((Float) rows.get(0).getField(1)) !=
> 0x80000000
> || Double.doubleToRawLongBits((Double)
> rows.get(0).getField(2))
> != 0x8000000000000000L) {
> throw new AssertionError("The source must contain negative
> zero");
> }
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
> env.setParallelism(1);
> final StreamTableEnvironment tables =
> StreamTableEnvironment.create(env,
> EnvironmentSettings.inStreamingMode());
> tables.createTemporaryView(
> "T",
> tables.fromDataStream(
> env.fromCollection(
> rows,
> Types.ROW_NAMED(
> new String[] {"id", "f", "d"},
> Types.INT, Types.FLOAT,
> Types.DOUBLE))));
> for (String type : Arrays.asList("FLOAT", "DOUBLE")) {
> final String field = type.equals("FLOAT") ? "f" : "d";
> final String query = "SELECT id FROM T WHERE " + field;
> final String zero = "CAST(0 AS " + type + ")";
> System.out.println(type + " equality: "
> + ids(tables, query + " = " + zero));
> System.out.println(type + " singleton IN: "
> + ids(tables, query + " IN (" + zero + ")"));
> System.out.println(type + " two-element IN: "
> + ids(tables, query + " IN (" + zero
> + ", CAST(2 AS " + type + "))"));
> }
> }
> private static List<Integer> ids(StreamTableEnvironment tables, String
> sql)
> throws Exception {
> final List<Integer> result = new ArrayList<>();
> try (CloseableIterator<Row> rows =
> tables.executeSql(sql).collect()) {
> rows.forEachRemaining(row -> result.add((Integer)
> row.getField(0)));
> }
> Collections.sort(result);
> return result;
> }
> }
> {code}
> Expected result
> All three predicates return the ID multiset [1, 1, 2] for both types.
> Actual result
> Equality and singleton IN return [1, 1, 2].
> The two-element IN returns [2].
> Execution-path verification
> The regression tests inspect generated code and verify
> FloatHashSet/DoubleHashSet membership for the two-element list and a
> 21-element distinct list. Singleton IN uses the equality path.
> SearchOperatorGen selects hash sets for point or complemented-point Sargs,
> without a list-size threshold at that stage. The separate threshold of 20 in
> ConvertToNotInOrInRule is not a general SQL IN threshold.
> Root cause
> Generated numeric equality uses primitive ==. FloatHashSet and
> DoubleHashSet intentionally implement boxed equality, which distinguishes
> positive and negative zero. SQL SEARCH currently uses these sets without
> adapting their zero
> equivalence.
> Complemented SEARCH also affects NOT IN. NULL handling must remain
> three-valued: NULL input yields UNKNOWN, and a nonmatching value with NULL in
> the list yields UNKNOWN.
> The public documentation does not explicitly specify signed-zero equality.
> This fix follows Flink's existing generated numeric equality and preserves
> equivalence across these execution paths.
> Proposed fix
> When preparing a floating-point SEARCH constant set, include both zero
> representations whenever either zero is present, after casting to the common
> type. Preserve both representations during code generation.
> Related work
> FLINK-35276 and PR #28383 address negative-zero sorting and remain open;
> they do not fix SQL membership. Searches did not identify an exact duplicate.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
