org.apache.flink.table.planner.plan.nodes.exec.common

Class CommonExecLookupJoin

java.lang.Object

org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase<org.apache.flink.table.data.RowData>

org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin

All Implemented Interfaces:

ExecNode<org.apache.flink.table.data.RowData>, ExecNodeTranslator<org.apache.flink.table.data.RowData>, FusionCodegenExecNode

Direct Known Subclasses:

BatchExecLookupJoin, StreamExecLookupJoin

public abstract class CommonExecLookupJoin
extends ExecNodeBase<org.apache.flink.table.data.RowData>

Base ExecNode for temporal table join which shares most methods.

For a lookup join query:

 SELECT T.id, T.content, D.age
 FROM T JOIN userTable FOR SYSTEM_TIME AS OF T.proctime AS D
 ON T.content = concat(D.name, '!') AND D.age = 11 AND T.id = D.id
 WHERE D.name LIKE 'Jack%'
 

The LookupJoin physical node encapsulates the following RelNode tree:

      Join (l.name = r.name)
    /     \
 RelNode  Calc (concat(name, "!") as name, name LIKE 'Jack%')
           |
        DimTable (lookup-keys: age=11, id=l.id)
     (age, id, name)
 

• lookupKeys: [$0=11, $1=l.id] ($0 and $1 is the indexes of age and id in dim table)
• calcOnTemporalTable: calc on temporal table rows before join
• joinCondition: join condition on temporal table rows after calc

The workflow of lookup join:

1) lookup records dimension table using the lookup-keys
2) project & filter on the lookup-ed records
3) join left input record and lookup-ed records
4) only outputs the rows which match to the condition

Field Summary

Fields

Modifier and Type	Field and Description
static String	FIELD_NAME_ASYNC_OPTIONS
static String	FIELD_NAME_FILTER_ON_TEMPORAL_TABLE
static String	FIELD_NAME_INPUT_CHANGELOG_MODE
static String	FIELD_NAME_JOIN_TYPE
static String	FIELD_NAME_LOOKUP_KEYS
static String	FIELD_NAME_PRE_FILTER_CONDITION
static String	FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE
static String	FIELD_NAME_REMAINING_JOIN_CONDITION
static String	FIELD_NAME_RETRY_OPTIONS
static String	FIELD_NAME_TEMPORAL_TABLE
static String	LOOKUP_JOIN_MATERIALIZE_TRANSFORMATION
static String	LOOKUP_JOIN_TRANSFORMATION

Fields inherited from interface org.apache.flink.table.planner.plan.nodes.exec.ExecNode

FIELD_NAME_CONFIGURATION, FIELD_NAME_DESCRIPTION, FIELD_NAME_ID, FIELD_NAME_INPUT_PROPERTIES, FIELD_NAME_OUTPUT_TYPE, FIELD_NAME_STATE, FIELD_NAME_TYPE

Constructor Summary

Constructors

Modifier

Constructor and Description

protected

CommonExecLookupJoin(int id, ExecNodeContext context, org.apache.flink.configuration.ReadableConfig persistedConfig, org.apache.flink.table.runtime.operators.join.FlinkJoinType joinType, org.apache.calcite.rex.RexNode preFilterCondition, org.apache.calcite.rex.RexNode remainingJoinCondition, TemporalTableSourceSpec temporalTableSourceSpec, Map<Integer,LookupJoinUtil.LookupKey> lookupKeys, List<org.apache.calcite.rex.RexNode> projectionOnTemporalTable, org.apache.calcite.rex.RexNode filterOnTemporalTable, LookupJoinUtil.AsyncLookupOptions asyncLookupOptions, LookupJoinUtil.RetryLookupOptions retryOptions, org.apache.flink.table.connector.ChangelogMode inputChangelogMode, List<InputProperty> inputProperties, org.apache.flink.table.types.logical.RowType outputType, String description)

Method Summary

Modifier and Type	Method and Description
protected org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData>	createJoinTransformation(PlannerBase planner, ExecNodeConfig config, boolean upsertMaterialize, boolean lookupKeyContainsPrimaryKey)
protected org.apache.flink.streaming.api.functions.ProcessFunction<org.apache.flink.table.data.RowData,org.apache.flink.table.data.RowData>	createSyncLookupJoinFunction(org.apache.calcite.plan.RelOptTable temporalTable, ExecNodeConfig config, ClassLoader classLoader, Map<Integer,LookupJoinUtil.LookupKey> allLookupKeys, org.apache.flink.table.functions.TableFunction<?> syncLookupFunction, RelBuilder relBuilder, org.apache.flink.table.types.logical.RowType inputRowType, org.apache.flink.table.types.logical.RowType tableSourceRowType, org.apache.flink.table.types.logical.RowType resultRowType, boolean isLeftOuterJoin, boolean isObjectReuseEnabled)
protected abstract org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData>	createSyncLookupJoinWithState(org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData> inputTransformation, org.apache.calcite.plan.RelOptTable temporalTable, ExecNodeConfig config, ClassLoader classLoader, Map<Integer,LookupJoinUtil.LookupKey> allLookupKeys, org.apache.flink.table.functions.TableFunction<?> syncLookupFunction, RelBuilder relBuilder, org.apache.flink.table.types.logical.RowType inputRowType, org.apache.flink.table.types.logical.RowType tableSourceRowType, org.apache.flink.table.types.logical.RowType resultRowType, boolean isLeftOuterJoin, boolean isObjectReuseEnabled, boolean lookupKeyContainsPrimaryKey)
protected org.apache.calcite.rel.type.RelDataType	getProjectionOutputRelDataType(RelBuilder relBuilder)
protected org.apache.flink.table.types.logical.RowType	getRightOutputRowType(org.apache.calcite.rel.type.RelDataType projectionOutputRelDataType, org.apache.flink.table.types.logical.RowType tableSourceRowType)
TemporalTableSourceSpec	getTemporalTableSourceSpec()
protected void	validateLookupKeyType(Map<Integer,LookupJoinUtil.LookupKey> lookupKeys, org.apache.flink.table.types.logical.RowType inputRowType, org.apache.flink.table.types.logical.RowType tableSourceRowType)

Methods inherited from class org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase

accept, createFormattedTransformationDescription, createFormattedTransformationName, createTransformationDescription, createTransformationMeta, createTransformationMeta, createTransformationName, createTransformationUid, getContextFromAnnotation, getDescription, getId, getInputEdges, getInputProperties, getOutputType, getPersistedConfig, getSimplifiedName, getTransformation, inputsContainSingleton, replaceInputEdge, setCompiled, setInputEdges, supportFusionCodegen, translateToFusionCodegenSpec, translateToFusionCodegenSpecInternal, translateToPlan, translateToPlanInternal

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

LOOKUP_JOIN_TRANSFORMATION

public static final String LOOKUP_JOIN_TRANSFORMATION

See Also:

Constant Field Values

LOOKUP_JOIN_MATERIALIZE_TRANSFORMATION

public static final String LOOKUP_JOIN_MATERIALIZE_TRANSFORMATION

See Also:

Constant Field Values

FIELD_NAME_JOIN_TYPE

public static final String FIELD_NAME_JOIN_TYPE

See Also:

Constant Field Values

FIELD_NAME_PRE_FILTER_CONDITION

public static final String FIELD_NAME_PRE_FILTER_CONDITION

See Also:

Constant Field Values

FIELD_NAME_REMAINING_JOIN_CONDITION

public static final String FIELD_NAME_REMAINING_JOIN_CONDITION

See Also:

Constant Field Values

FIELD_NAME_TEMPORAL_TABLE

public static final String FIELD_NAME_TEMPORAL_TABLE

See Also:

Constant Field Values

FIELD_NAME_LOOKUP_KEYS

public static final String FIELD_NAME_LOOKUP_KEYS

See Also:

Constant Field Values

FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE

public static final String FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE

See Also:

Constant Field Values

FIELD_NAME_FILTER_ON_TEMPORAL_TABLE

public static final String FIELD_NAME_FILTER_ON_TEMPORAL_TABLE

See Also:

Constant Field Values

FIELD_NAME_INPUT_CHANGELOG_MODE

public static final String FIELD_NAME_INPUT_CHANGELOG_MODE

See Also:

Constant Field Values

FIELD_NAME_ASYNC_OPTIONS

public static final String FIELD_NAME_ASYNC_OPTIONS

See Also:

Constant Field Values

FIELD_NAME_RETRY_OPTIONS

public static final String FIELD_NAME_RETRY_OPTIONS

See Also:

Constant Field Values

Constructor Detail

CommonExecLookupJoin

protected CommonExecLookupJoin(int id,
                               ExecNodeContext context,
                               org.apache.flink.configuration.ReadableConfig persistedConfig,
                               org.apache.flink.table.runtime.operators.join.FlinkJoinType joinType,
                               @Nullable
                               org.apache.calcite.rex.RexNode preFilterCondition,
                               @Nullable
                               org.apache.calcite.rex.RexNode remainingJoinCondition,
                               TemporalTableSourceSpec temporalTableSourceSpec,
                               Map<Integer,LookupJoinUtil.LookupKey> lookupKeys,
                               @Nullable
                               List<org.apache.calcite.rex.RexNode> projectionOnTemporalTable,
                               @Nullable
                               org.apache.calcite.rex.RexNode filterOnTemporalTable,
                               @Nullable
                               LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
                               @Nullable
                               LookupJoinUtil.RetryLookupOptions retryOptions,
                               org.apache.flink.table.connector.ChangelogMode inputChangelogMode,
                               List<InputProperty> inputProperties,
                               org.apache.flink.table.types.logical.RowType outputType,
                               String description)

Method Detail

getTemporalTableSourceSpec

public TemporalTableSourceSpec getTemporalTableSourceSpec()

createJoinTransformation

protected org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData> createJoinTransformation(PlannerBase planner,
                                                                                                                ExecNodeConfig config,
                                                                                                                boolean upsertMaterialize,
                                                                                                                boolean lookupKeyContainsPrimaryKey)

createSyncLookupJoinWithState

protected abstract org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData> createSyncLookupJoinWithState(org.apache.flink.api.dag.Transformation<org.apache.flink.table.data.RowData> inputTransformation,
                                                                                                                              org.apache.calcite.plan.RelOptTable temporalTable,
                                                                                                                              ExecNodeConfig config,
                                                                                                                              ClassLoader classLoader,
                                                                                                                              Map<Integer,LookupJoinUtil.LookupKey> allLookupKeys,
                                                                                                                              org.apache.flink.table.functions.TableFunction<?> syncLookupFunction,
                                                                                                                              RelBuilder relBuilder,
                                                                                                                              org.apache.flink.table.types.logical.RowType inputRowType,
                                                                                                                              org.apache.flink.table.types.logical.RowType tableSourceRowType,
                                                                                                                              org.apache.flink.table.types.logical.RowType resultRowType,
                                                                                                                              boolean isLeftOuterJoin,
                                                                                                                              boolean isObjectReuseEnabled,
                                                                                                                              boolean lookupKeyContainsPrimaryKey)

validateLookupKeyType

protected void validateLookupKeyType(Map<Integer,LookupJoinUtil.LookupKey> lookupKeys,
                                     org.apache.flink.table.types.logical.RowType inputRowType,
                                     org.apache.flink.table.types.logical.RowType tableSourceRowType)

getProjectionOutputRelDataType

protected org.apache.calcite.rel.type.RelDataType getProjectionOutputRelDataType(RelBuilder relBuilder)

getRightOutputRowType

protected org.apache.flink.table.types.logical.RowType getRightOutputRowType(org.apache.calcite.rel.type.RelDataType projectionOutputRelDataType,
                                                                             org.apache.flink.table.types.logical.RowType tableSourceRowType)

createSyncLookupJoinFunction

protected org.apache.flink.streaming.api.functions.ProcessFunction<org.apache.flink.table.data.RowData,org.apache.flink.table.data.RowData> createSyncLookupJoinFunction(org.apache.calcite.plan.RelOptTable temporalTable,
                                                                                                                                                                         ExecNodeConfig config,
                                                                                                                                                                         ClassLoader classLoader,
                                                                                                                                                                         Map<Integer,LookupJoinUtil.LookupKey> allLookupKeys,
                                                                                                                                                                         org.apache.flink.table.functions.TableFunction<?> syncLookupFunction,
                                                                                                                                                                         RelBuilder relBuilder,
                                                                                                                                                                         org.apache.flink.table.types.logical.RowType inputRowType,
                                                                                                                                                                         org.apache.flink.table.types.logical.RowType tableSourceRowType,
                                                                                                                                                                         org.apache.flink.table.types.logical.RowType resultRowType,
                                                                                                                                                                         boolean isLeftOuterJoin,
                                                                                                                                                                         boolean isObjectReuseEnabled)