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
