Tracking Issue: DynamicFilter operator

[jon-chuang]

Design Doc: https://singularity-data.quip.com/AE06Ao1kAIaZ/RFC-Dynamic-Filter-A-New-Streaming-Operator

Logical:

• StreamDynamicFilter (feat(frontend): StreamDynamicFilter #3515)
  • Not sure if we need BatchDynamicFilter?
  • Hash distribution on left_col
  • Optimize scalar subqueries into LogicalDynamicFilter. Enable tpch q11, q22 planner tests
• proto for StreamDynamicFilter

Executor:

• Design of in-memory state for DynamicFilterExecutor (no StateTable yet) - (feat(executor): Basic in-application-memory DynamicFilterExecutor, pass tpch q11, q22 e2e tests  #3681)
  • For the comparison operators: <, <=, >, >= cache looks like a BTreeMap.
  • For ==, <> it could be BTreeMap or HashMap.
  • From proto
  • Enable tpch q11, q22 e2e tests
• DynamicFilterBroadcastExecutor (broadcasts the RHS value)
• StateTable representation for DynamicFilter (feat(executor): Use TableCatalog and StateTable in DynamicFilter (only writes) #3757)
  • For this, we need the vnode key encoding (Tracking: take vnode into consideration for key ordering #3316) to be ready
  • We require vnode range + distribution key scan
  • Encoding (outer side): table_id | vnode | left_col | left_pk
• Cache
  • cache read policy
  • Cache eviction policy (feat(executor): Managed DynamicFilter cache eviction based on epoch #6540)
• Unit tests
• Optimization: Monotonically-increasing pointers for DynamicFilter (similar to append-only optimizations)
• Parametrize by output_indices once ready in chunk builder (https://github.com/singularity-data/risingwave/pull/3588/files)

---

Questions:

1. Do we support dynamic filter for more than numeric data types? (can support I64, F64, NaiveDate/Time, String?, Bytes?).
2. Will the frontend ensure that the left and right types are the same datatype? (we could do a cast on the RHS scalar subquery result)?
3. Can we support only ==, <> and not Cmp operators for some datatypes? I guess the determination for this can be done in the frontend.
4. Shouldn't we use the anti/semi join for ==, <>? (we don't need broadcast for these)

References:

1. Range joins in DuckDB

[st1page]

there is some discussion about the detailed behavior for scalar subqueries in #2279. and some question still exists for the DynamicFilter

1. what should we do if the DynamicFilter gets more than one row? should we throw an error and stop all the streaming jobs? or do not generate the DynamicFilter if the side "could" get more than one row?
2. can we introduce the "LogicalCardinality", "max1row", "min1row", "exact1row" or other similar properties, and then consider the DynamicFilter rewriting as a general optimization for LogicalJoin?

[jon-chuang]

what should we do if the DynamicFilter gets more than one row? should we throw an error and stop all the streaming jobs? or do not generate the DynamicFilter if the side "could" get more than one row?

This is a good question. Actually, SQL does not restrict a non-equijoin join to only be evaluated on a single row. So as a matter of fact, scalar subqueries need to be explicitly identified as those which must return exactly one row (i.e. contains a simple agg). Using the exact1row and max1row properties can indeed facilitate this query rewriting.

However, perhaps as a first approximation, we could simply pattern match on simpleagg with a single column.

[st1page]

Will the frontend ensure that the left and right types are the same datatype? (we could do a cast on the RHS scalar subquery result)?

yes.

Shouldn't we use the anti/semi join for ==, <>? (we don't need broadcast for these)

I think so, and the DynamicFilter can focus on the comparison operators: <, <=, >, >=

[fuyufjh]

Just brought it back to the project to keep it under tracking. Ping @jon-chuang Any plan for it?