[jira] [Created] (FLINK-39088) Preserve upsert keys for explicit (injective) CAST calls to avoid unnecessary SinkUpsertMaterializer

["Gustavo de Morais (Jira)" <ji...@apache.org>]

Gustavo de Morais created FLINK-39088:
-----------------------------------------

             Summary: Preserve upsert keys for explicit (injective) CAST calls to avoid unnecessary SinkUpsertMaterializer
                 Key: FLINK-39088
                 URL: https://issues.apache.org/jira/browse/FLINK-39088
             Project: Flink
          Issue Type: Improvement
          Components: Table SQL / Planner
    Affects Versions: 2.2.0
            Reporter: Gustavo de Morais
            Assignee: Gustavo de Morais
             Fix For: 2.3.0


Currently, when a CAST operation is applied to a primary key or upsert key column in a streaming query, Flink's optimizer loses track of the key even when the cast is provably injective (one-to-one mapping). This causes the planner to insert a SinkUpsertMaterializer operator unnecessarily, adding state overhead and reducing performance.For example:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
CREATE TABLE source (
  id INT PRIMARY KEY NOT ENFORCED,
  name STRING
) WITH ('connector' = 'kafka', ...);
 
CREATE TABLE sink (
  id STRING PRIMARY KEY NOT ENFORCED,
  name STRING  
) WITH ('connector' = 'jdbc', ...);
 
INSERT INTO sink SELECT CAST(id AS STRING), name FROM source;
 
 
 
 
 
 
 
 
 
In this case, CAST(INT AS STRING) is injective - every distinct integer maps to a distinct string representation. The upsert key should be preserved through this cast. However, the current implementation treats any cast as potentially key-destroying, requiring a SinkUpsertMaterializer to re-establish the key before writing to the sink.Solution:Extend the key-tracking logic in FlinkRelMdUniqueKeys to recognize injective casts as key-preserving operations. An explicit cast is considered injective when every distinct input value maps to a distinct output value.The following explicit casts to VARCHAR/CHAR will be recognized as injective: * Integer types: TINYINT, SMALLINT, INTEGER, BIGINT

 * Floating point types: FLOAT, DOUBLE

 * BOOLEAN

 * DATE

 * Timestamp types: TIMESTAMP, TIMESTAMP_LTZ, TIMESTAMP_TZ



--
This message was sent by Atlassian Jira
(v8.20.10#820010)