Arvid Heise created FLINK-40909:
-----------------------------------
Summary: Support filter push down for element access on ARRAY, MAP
and VARIANT (ITEM)
Key: FLINK-40909
URL: https://issues.apache.org/jira/browse/FLINK-40909
Project: Flink
Issue Type: Improvement
Components: Table SQL / Planner
Affects Versions: 2.4.0
Reporter: Arvid Heise
Assignee: Arvid Heise
Predicates on an element of an ARRAY, MAP or VARIANT column are never offered
to a source, even if the source implements SupportsFilterPushDown:
{code:sql}
CREATE TABLE src (a ARRAY<INT>, m MAP<STRING, INT>, v VARIANT) WITH (...);
SELECT * FROM src WHERE a[1] = 3;
SELECT * FROM src WHERE m['k'] = 3;
SELECT * FROM src WHERE CAST(v['k'] AS INT) = 3;
{code}
All three conjuncts stay in the Calc above the scan; applyFilters receives
nothing.
h3. Cause
SQL element access is Calcite's ITEM operator.
RexNodeToExpressionConverter#visitCall (RexNodeExtractor.scala) only maps OR,
AND, CAST and TRY_CAST explicitly; every other operator is resolved by name
through the function catalog. There is no built-in named "ITEM" (the
corresponding built-in is BuiltInFunctionDefinitions.AT, named "at"), so the
lookup returns None and FlinkRexUtil#extractPredicates classifies the conjunct
as unconvertible.
The reverse direction already exists: DirectConvertRule maps AT ->
FlinkSqlOperatorTable.ITEM, so Table API at() is planned and executed as ITEM.
FLINK-37924 extended both SqlItemOperator and AT to VARIANT.
h3. Proposal
Add an ITEM case to RexNodeToExpressionConverter#visitCall that produces
CallExpression(AT, operands, outputType) when the first operand is ARRAY, MAP
or VARIANT, i.e. exactly the types AT's input type strategy accepts.
Keep returning None for the other ITEM operands:
* ROW / STRUCTURED_TYPE with a string key (r['f']): handled by ITEM in codegen
(generateDot), but rejected by AT. A ROW with a numeric index is already turned
into a field access by StandardConvertletTable#convertItem. Mapping r['f'] to a
NestedFieldReferenceExpression (FLIP-356) could be a follow-up.
* ANY (dynamic star).
h3. Compatibility
* Sources implementing SupportsFilterPushDown start receiving AT calls. Sources
that do not recognize a call are expected to return it as remaining, so results
do not change, but the plans of sources that accept such filters do.
* The converted expression keeps the RexCall's return type; this must agree
with AT's ITEM_AT output strategy (notably nullability of MAP / VARIANT
access), otherwise FilterPushDownSpec fails to re-resolve the predicate when a
compiled plan is restored.
h3. Tests
* PushFilterIntoTableSourceScanRuleTest: ARRAY index (1-based), MAP key,
VARIANT key and index are pushed; ROW string key is not. Teach TestValues'
FilterUtils to evaluate AT so the values source can accept them.
* Compiled plan round trip of a pushed ITEM filter (JSON plan restore).
* IT on the values source showing identical results with and without push down.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
