Hi everyone,

I'd like to start a discussion on a FLIP that introduces higher-order functions
(functions that take a lambda expression as an argument) to Flink SQL, together
with the first two built-ins that use them: TRANSFORM and ARRAY_FILTER.

FLIP: 
https://docs.google.com/document/d/144P06vspNDwU3nevEluPeOsWBQsaaQUhe8DouiEFqeQ/edit?usp=sharing

Motivation

Users migrating to Flink SQL from Spark, Databricks, Snowflake, DuckDB, and
Presto/Trino expect to manipulate collections (arrays and maps) inline with a
lambda instead of UNNEST + re-aggregate rewrites or bespoke UDFs. The absence of
higher-order collection functions forces verbose query rewrites during migration
and raises time-to-first-query. TRANSFORM(array, x -> x + 1) and
ARRAY_FILTER(array, x -> x > 0) are the two most requested entry points and,
importantly, they can be built on lambda infrastructure that already exists in
Calcite (CALCITE-3679), so the surface area we add on the Flink side is rather
small.

Summary

 •  Introduce a new logical type FUNCTION that describes the type of a lambda
    (its argument types and its result type). This is the type-system foundation
    every higher-order function needs; it is a planning/translation helper type
    and is not a persisted column type.
 •  Add ARRAY_FILTER(array, element -> predicate), which returns a new array
    containing only the elements for which the predicate holds.
 •  Add TRANSFORM(collection, lambda), which applies a lambda to every element 
of
    an array (TRANSFORM(array, x -> expr)) or every entry of a map
    (TRANSFORM(map, (k, v) -> expr)), returning a new array/map.
 •  Reuse Calcite's lambda parsing, validation and RexLambda/FunctionSqlType
    infrastructure (CALCITE-3679) rather than inventing a Flink-specific lambda
    syntax. The lambda arrow syntax x -> expr and (k, v) -> expr is already
    parseable by the Calcite version Flink bundles (1.41.0).

The functions are net-new syntax and are additive: no existing query changes
behavior. There is no new configuration option — the functions are always
available once the release ships.

Examples:

    SELECT ARRAY_FILTER(ARRAY[1, 2, 3, 4], x -> x > 2);              -- [3, 4]
    SELECT TRANSFORM(ARRAY[1, 2, 3], x -> x * 10);                   -- [10, 
20, 30]
    SELECT TRANSFORM(ARRAY['a', 'bb', 'ccc'], s -> CHAR_LENGTH(s));  -- [1, 2, 
3]
    SELECT TRANSFORM(MAP['a', 1, 'b', 2], (k, v) -> v * 100);        -- {a=100, 
b=200}

Thanks,
Dominik Bünzli
Data, Analytics & AI Engineer

Reply via email to