This is an automated email from the ASF dual-hosted git repository. godfrey pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d402fe255cdca37568ad6ac585ac6fbdf24b5e74 Author: lincoln lee <[email protected]> AuthorDate: Thu Aug 18 13:41:55 2022 +0800 [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable by fixing this we can completely avoid create user lookup function instance to determine if async lookup is enabled for execution, and make the exec lookup join node immutable and easier to maintain This closes #20592 --- .../table/planner/hint/LookupJoinHintOptions.java | 11 +- .../plan/nodes/exec/batch/BatchExecLookupJoin.java | 6 +- .../nodes/exec/common/CommonExecLookupJoin.java | 68 ++- .../plan/nodes/exec/spec/LookupJoinHintSpec.java | 239 --------- .../nodes/exec/stream/StreamExecLookupJoin.java | 29 +- .../table/planner/plan/utils/LookupJoinUtil.java | 556 +++++++++++++-------- .../physical/batch/BatchPhysicalLookupJoin.scala | 4 +- .../physical/common/CommonPhysicalLookupJoin.scala | 55 +- .../physical/stream/StreamPhysicalLookupJoin.scala | 36 +- .../stream/StreamPhysicalLookupJoinRule.scala | 14 +- ...pJoinHintWithInvalidPropagationShuttleTest.java | 10 +- .../nodes/exec/serde/AsyncLookupOptionsTest.java | 138 +++++ .../exec/serde/LookupJoinHintSpecSerdeTest.java | 51 -- .../nodes/exec/serde/RetryLookupOptionsTest.java | 80 +++ ...ntSpecTest.java => LookupJoinHintTestUtil.java} | 35 +- .../testAggAndLeftJoinWithTryResolveMode.out | 4 +- .../testJoinTemporalTable.out | 4 +- .../testJoinTemporalTableWithAsyncHint.out | 8 +- .../testJoinTemporalTableWithAsyncHint2.out | 8 +- .../testJoinTemporalTableWithAsyncRetryHint.out | 18 +- .../testJoinTemporalTableWithAsyncRetryHint2.out | 18 +- ...testJoinTemporalTableWithProjectionPushDown.out | 4 +- .../testJoinTemporalTableWithRetryHint.out | 17 +- .../plan/stream/sql/join/LookupJoinTest.xml | 250 ++++++++- .../plan/metadata/FlinkRelMdHandlerTestBase.scala | 4 +- .../plan/stream/sql/NonDeterministicDagTest.scala | 6 +- .../plan/stream/sql/join/LookupJoinTest.scala | 46 +- .../runtime/stream/sql/AsyncLookupJoinITCase.scala | 9 +- 28 files changed, 1013 insertions(+), 715 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java index b634572035d..1afb8ff8f9f 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java @@ -53,9 +53,7 @@ public class LookupJoinHintOptions { public static final ConfigOption<ExecutionConfigOptions.AsyncOutputMode> ASYNC_OUTPUT_MODE = key("output-mode") .enumType(ExecutionConfigOptions.AsyncOutputMode.class) - .defaultValue( - ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE - .defaultValue()) + .noDefaultValue() .withDescription( "Output mode for asynchronous operations which will convert to {@see AsyncDataStream.OutputMode}, ORDERED by default. " + "If set to ALLOW_UNORDERED, will attempt to use {@see AsyncDataStream.OutputMode.UNORDERED} when it does not " @@ -64,17 +62,14 @@ public class LookupJoinHintOptions { public static final ConfigOption<Integer> ASYNC_CAPACITY = key("capacity") .intType() - .defaultValue( - ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY - .defaultValue()) + .noDefaultValue() .withDescription( "The max number of async i/o operation that the async lookup join can trigger."); public static final ConfigOption<Duration> ASYNC_TIMEOUT = key("timeout") .durationType() - .defaultValue( - ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_TIMEOUT.defaultValue()) + .noDefaultValue() .withDescription( "Timeout from first invoke to final completion of asynchronous operation, may include multiple" + " retries, and will be reset in case of failover."); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java index 2c07e3f03ac..6e4ef47ff7b 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java @@ -50,6 +50,7 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin implements BatchEx Map<Integer, LookupJoinUtil.LookupKey> lookupKeys, @Nullable List<RexNode> projectionOnTemporalTable, @Nullable RexNode filterOnTemporalTable, + @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions, InputProperty inputProperty, RowType outputType, String description) { @@ -63,11 +64,12 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin implements BatchEx lookupKeys, projectionOnTemporalTable, filterOnTemporalTable, + asyncLookupOptions, + // batch lookup join does not support retry hint currently + null, ChangelogMode.insertOnly(), Collections.singletonList(inputProperty), outputType, - // batch lookup join does not support hint currently - null, description); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java index b68f0309326..d520f7630ce 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java @@ -24,7 +24,6 @@ import org.apache.flink.api.dag.Transformation; import org.apache.flink.api.java.typeutils.RowTypeInfo; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.runtime.state.KeyGroupRangeAssignment; -import org.apache.flink.streaming.api.datastream.AsyncDataStream; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.streaming.api.functions.async.AsyncFunction; import org.apache.flink.streaming.api.operators.KeyedProcessOperator; @@ -58,7 +57,6 @@ import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext; import org.apache.flink.table.planner.plan.nodes.exec.InputProperty; import org.apache.flink.table.planner.plan.nodes.exec.SingleTransformationTranslator; -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec; import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSpec; import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil; import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable; @@ -80,6 +78,7 @@ import org.apache.flink.table.runtime.operators.join.lookup.AsyncLookupJoinWithC import org.apache.flink.table.runtime.operators.join.lookup.KeyedLookupJoinWrapper; import org.apache.flink.table.runtime.operators.join.lookup.LookupJoinRunner; import org.apache.flink.table.runtime.operators.join.lookup.LookupJoinWithCalcRunner; +import org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy; import org.apache.flink.table.runtime.types.PlannerTypeUtils; import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter; import org.apache.flink.table.runtime.typeutils.InternalSerializers; @@ -111,6 +110,7 @@ import java.util.Optional; import java.util.stream.Collectors; import static org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType; +import static org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.PARTITIONER_TRANSFORMATION; import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTypeFactory; import static org.apache.flink.util.Preconditions.checkArgument; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -167,7 +167,9 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> public static final String FIELD_NAME_FILTER_ON_TEMPORAL_TABLE = "filterOnTemporalTable"; public static final String FIELD_NAME_INPUT_CHANGELOG_MODE = "inputChangelogMode"; - public static final String FIELD_NAME_JOIN_HINT = "joinHint"; + + public static final String FIELD_NAME_ASYNC_OPTIONS = "asyncOptions"; + public static final String FIELD_NAME_RETRY_OPTIONS = "retryOptions"; @JsonProperty(FIELD_NAME_JOIN_TYPE) private final FlinkJoinType joinType; @@ -195,9 +197,13 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> @JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE) private final ChangelogMode inputChangelogMode; - @JsonProperty(FIELD_NAME_JOIN_HINT) + @JsonProperty(FIELD_NAME_ASYNC_OPTIONS) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions; + + @JsonProperty(FIELD_NAME_RETRY_OPTIONS) @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable LookupJoinHintSpec joinHintSpec; + private final @Nullable LookupJoinUtil.RetryLookupOptions retryOptions; protected CommonExecLookupJoin( int id, @@ -210,10 +216,11 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> Map<Integer, LookupJoinUtil.LookupKey> lookupKeys, @Nullable List<RexNode> projectionOnTemporalTable, @Nullable RexNode filterOnTemporalTable, + @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions, + @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, ChangelogMode inputChangelogMode, List<InputProperty> inputProperties, RowType outputType, - @Nullable LookupJoinHintSpec lookupJoinHintSpec, String description) { super(id, context, persistedConfig, inputProperties, outputType, description); checkArgument(inputProperties.size() == 1); @@ -224,7 +231,8 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> this.projectionOnTemporalTable = projectionOnTemporalTable; this.filterOnTemporalTable = filterOnTemporalTable; this.inputChangelogMode = inputChangelogMode; - this.joinHintSpec = lookupJoinHintSpec; + this.asyncLookupOptions = asyncLookupOptions; + this.retryOptions = retryOptions; } public TemporalTableSourceSpec getTemporalTableSourceSpec() { @@ -246,18 +254,23 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> RowType tableSourceRowType = FlinkTypeFactory.toLogicalRowType(temporalTable.getRowType()); RowType resultRowType = (RowType) getOutputType(); validateLookupKeyType(lookupKeys, inputRowType, tableSourceRowType); + boolean isAsyncEnabled = null != asyncLookupOptions; + ResultRetryStrategy retryStrategy = + retryOptions != null ? retryOptions.toRetryStrategy() : null; UserDefinedFunction lookupFunction = LookupJoinUtil.getLookupFunction( temporalTable, lookupKeys.keySet(), planner.getFlinkContext().getClassLoader(), - joinHintSpec, - upsertMaterialize); + isAsyncEnabled, + retryStrategy); UserDefinedFunctionHelper.prepareInstance(config, lookupFunction); boolean isLeftOuterJoin = joinType == FlinkJoinType.LEFT; - boolean isAsyncEnabled = lookupFunction instanceof AsyncTableFunction; + if (isAsyncEnabled) { + assert lookupFunction instanceof AsyncTableFunction; + } Transformation<RowData> inputTransformation = (Transformation<RowData>) inputEdge.translateToPlan(planner); @@ -293,7 +306,8 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> inputRowType, tableSourceRowType, resultRowType, - isLeftOuterJoin); + isLeftOuterJoin, + asyncLookupOptions); } else { operatorFactory = createSyncLookupJoin( @@ -390,6 +404,8 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> keySelector, KeyGroupRangeAssignment.DEFAULT_LOWER_BOUND_MAX_PARALLELISM); Transformation<RowData> partitionedTransform = new PartitionTransformation<>(inputTransformation, partitioner); + createTransformationMeta(PARTITIONER_TRANSFORMATION, "Partitioner", "Partitioner", config) + .fill(partitionedTransform); if (singleParallelism) { setSingletonParallelism(partitionedTransform); } else { @@ -450,20 +466,6 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> } } - private LookupJoinUtil.AsyncLookupOptions getAsyncOptions(ExecNodeConfig config) { - if (joinHintSpec != null) { - // hint first - return new LookupJoinUtil.AsyncLookupOptions( - joinHintSpec.getAsyncCapacity(), - joinHintSpec.getAsyncTimeout(), - joinHintSpec.getAsyncOutputMode()); - } - return new LookupJoinUtil.AsyncLookupOptions( - config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY), - config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_TIMEOUT).toMillis(), - config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE)); - } - @SuppressWarnings("unchecked") private StreamOperatorFactory<RowData> createAsyncLookupJoin( RelOptTable temporalTable, @@ -475,9 +477,8 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> RowType inputRowType, RowType tableSourceRowType, RowType resultRowType, - boolean isLeftOuterJoin) { - - LookupJoinUtil.AsyncLookupOptions asyncLookupOptions = getAsyncOptions(config); + boolean isLeftOuterJoin, + LookupJoinUtil.AsyncLookupOptions asyncLookupOptions) { DataTypeFactory dataTypeFactory = ShortcutUtils.unwrapContext(relBuilder).getCatalogManager().getDataTypeFactory(); @@ -552,16 +553,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> asyncFunc, asyncLookupOptions.asyncTimeout, asyncLookupOptions.asyncBufferCapacity, - convert(asyncLookupOptions.asyncOutputMode)); - } - - private AsyncDataStream.OutputMode convert( - ExecutionConfigOptions.AsyncOutputMode asyncOutputMode) { - if (inputChangelogMode.containsOnly(RowKind.INSERT) - && asyncOutputMode == ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED) { - return AsyncDataStream.OutputMode.UNORDERED; - } - return AsyncDataStream.OutputMode.ORDERED; + asyncLookupOptions.asyncOutputMode); } private StreamOperatorFactory<RowData> createSyncLookupJoin( diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpec.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpec.java deleted file mode 100644 index 796e7a3ea23..00000000000 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpec.java +++ /dev/null @@ -1,239 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.table.planner.plan.nodes.exec.spec; - -import org.apache.flink.configuration.Configuration; -import org.apache.flink.streaming.util.retryable.RetryPredicates; -import org.apache.flink.table.api.config.ExecutionConfigOptions; -import org.apache.flink.table.planner.hint.LookupJoinHintOptions; -import org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy; - -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; - -import org.apache.calcite.rel.hint.RelHint; - -import javax.annotation.Nullable; - -import java.time.Duration; -import java.util.Objects; - -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_CAPACITY; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_LOOKUP; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_OUTPUT_MODE; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_TIMEOUT; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.FIXED_DELAY; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.MAX_ATTEMPTS; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.RETRY_PREDICATE; -import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.RETRY_STRATEGY; -import static org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy.NO_RETRY_STRATEGY; -import static org.apache.flink.util.Preconditions.checkNotNull; - -/** - * LookupJoinHintSpec describes the user specified hint options for lookup join. - * - * <p>This class corresponds to {@link - * org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin} rel node. - */ -@JsonIgnoreProperties(ignoreUnknown = true) -public class LookupJoinHintSpec { - public static final String FIELD_NAME_ASYNC = "async"; - public static final String FIELD_NAME_ASYNC_OUTPUT_MODE = "output-mode"; - public static final String FIELD_NAME_ASYNC_CAPACITY = "capacity"; - public static final String FIELD_NAME_ASYNC_TIMEOUT = "timeout"; - public static final String FIELD_NAME_RETRY_PREDICATE = "retry-predicate"; - public static final String FIELD_NAME_RETRY_STRATEGY = "retry-strategy"; - public static final String FIELD_NAME_RETRY_FIXED_DELAY = "fixed-delay"; - public static final String FIELD_NAME_RETRY_MAX_ATTEMPTS = "max-attempts"; - - @JsonProperty(FIELD_NAME_ASYNC) - @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable Boolean async; - - @JsonProperty(FIELD_NAME_ASYNC_OUTPUT_MODE) - private final ExecutionConfigOptions.AsyncOutputMode asyncOutputMode; - - @JsonProperty(FIELD_NAME_ASYNC_CAPACITY) - private final Integer asyncCapacity; - - @JsonProperty(FIELD_NAME_ASYNC_TIMEOUT) - private final Long asyncTimeout; - - @JsonProperty(FIELD_NAME_RETRY_PREDICATE) - @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable String retryPredicate; - - @JsonProperty(FIELD_NAME_RETRY_STRATEGY) - @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable LookupJoinHintOptions.RetryStrategy retryStrategy; - - @JsonProperty(FIELD_NAME_RETRY_FIXED_DELAY) - @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable Long retryFixedDelay; - - @JsonProperty(FIELD_NAME_RETRY_MAX_ATTEMPTS) - @JsonInclude(JsonInclude.Include.NON_NULL) - private final @Nullable Integer retryMaxAttempts; - - @JsonCreator - public LookupJoinHintSpec( - @JsonProperty(FIELD_NAME_ASYNC) @Nullable Boolean async, - @JsonProperty(FIELD_NAME_ASYNC_OUTPUT_MODE) @Nullable - ExecutionConfigOptions.AsyncOutputMode asyncOutputMode, - @JsonProperty(FIELD_NAME_ASYNC_CAPACITY) Integer asyncCapacity, - @JsonProperty(FIELD_NAME_ASYNC_TIMEOUT) Long asyncTimeout, - @JsonProperty(FIELD_NAME_RETRY_PREDICATE) String retryPredicate, - @JsonProperty(FIELD_NAME_RETRY_STRATEGY) @Nullable - LookupJoinHintOptions.RetryStrategy retryStrategy, - @JsonProperty(FIELD_NAME_RETRY_FIXED_DELAY) @Nullable Long retryFixedDelay, - @JsonProperty(FIELD_NAME_RETRY_MAX_ATTEMPTS) @Nullable Integer retryMaxAttempts) { - this.async = async; - this.asyncOutputMode = checkNotNull(asyncOutputMode); - this.asyncCapacity = checkNotNull(asyncCapacity); - this.asyncTimeout = checkNotNull(asyncTimeout); - this.retryPredicate = retryPredicate; - this.retryStrategy = retryStrategy; - this.retryFixedDelay = retryFixedDelay; - this.retryMaxAttempts = retryMaxAttempts; - } - - @JsonIgnore - @Nullable - public Boolean getAsync() { - return async; - } - - @JsonIgnore - public Boolean isAsync() { - return Boolean.TRUE.equals(async); - } - - @JsonIgnore - public ExecutionConfigOptions.AsyncOutputMode getAsyncOutputMode() { - return asyncOutputMode; - } - - @JsonIgnore - public Integer getAsyncCapacity() { - return asyncCapacity; - } - - @JsonIgnore - public Long getAsyncTimeout() { - return asyncTimeout; - } - - @JsonIgnore - public String getRetryPredicate() { - return retryPredicate; - } - - @JsonIgnore - @Nullable - public LookupJoinHintOptions.RetryStrategy getRetryStrategy() { - return retryStrategy; - } - - @JsonIgnore - @Nullable - public Long getRetryFixedDelay() { - return retryFixedDelay; - } - - @JsonIgnore - @Nullable - public Integer getRetryMaxAttempts() { - return retryMaxAttempts; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - LookupJoinHintSpec that = (LookupJoinHintSpec) o; - return Objects.equals(async, that.async) - && asyncOutputMode == that.asyncOutputMode - && Objects.equals(asyncCapacity, that.asyncCapacity) - && Objects.equals(asyncTimeout, that.asyncTimeout) - && Objects.equals(retryPredicate, that.retryPredicate) - && retryStrategy == that.retryStrategy - && Objects.equals(retryFixedDelay, that.retryFixedDelay) - && Objects.equals(retryMaxAttempts, that.retryMaxAttempts); - } - - @Override - public int hashCode() { - return Objects.hash( - async, - asyncOutputMode, - asyncCapacity, - asyncTimeout, - retryPredicate, - retryStrategy, - retryFixedDelay, - retryMaxAttempts); - } - - /** - * Convert given joinHint to {@link LookupJoinHintSpec}. - * - * @param lookupJoinHint - */ - @JsonIgnore - public static LookupJoinHintSpec fromJoinHint(RelHint lookupJoinHint) { - Configuration conf = Configuration.fromMap(lookupJoinHint.kvOptions); - Duration fixedDelay = conf.get(FIXED_DELAY); - return new LookupJoinHintSpec( - conf.get(ASYNC_LOOKUP), - conf.get(ASYNC_OUTPUT_MODE), - conf.get(ASYNC_CAPACITY), - conf.get(ASYNC_TIMEOUT).toMillis(), - conf.get(RETRY_PREDICATE), - conf.get(RETRY_STRATEGY), - fixedDelay != null ? fixedDelay.toMillis() : null, - conf.get(MAX_ATTEMPTS)); - } - - /** - * Convert this {@link LookupJoinHintSpec} to {@link ResultRetryStrategy} in a best effort - * manner. If invalid {@link LookupJoinHintOptions#RETRY_PREDICATE} or {@link - * LookupJoinHintOptions#RETRY_STRATEGY} is given, then {@link - * ResultRetryStrategy#NO_RETRY_STRATEGY} will return. - */ - @JsonIgnore - public ResultRetryStrategy toRetryStrategy() { - if (null == retryPredicate - || !LookupJoinHintOptions.LOOKUP_MISS_PREDICATE.equalsIgnoreCase(retryPredicate) - || retryStrategy != LookupJoinHintOptions.RetryStrategy.FIXED_DELAY) { - return NO_RETRY_STRATEGY; - } - // retry option values have been validated by hint checker - return ResultRetryStrategy.fixedDelayRetry( - this.retryMaxAttempts, - this.retryFixedDelay, - RetryPredicates.EMPTY_RESULT_PREDICATE); - } -} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java index 6b83b3d380c..5f4ebd7bffb 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java @@ -29,7 +29,6 @@ import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata; import org.apache.flink.table.planner.plan.nodes.exec.InputProperty; import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin; -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec; import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSpec; import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; import org.apache.flink.table.runtime.operators.join.FlinkJoinType; @@ -75,12 +74,13 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin implements Stream Map<Integer, LookupJoinUtil.LookupKey> lookupKeys, @Nullable List<RexNode> projectionOnTemporalTable, @Nullable RexNode filterOnTemporalTable, + boolean lookupKeyContainsPrimaryKey, + boolean upsertMaterialize, + @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions, + @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, ChangelogMode inputChangelogMode, InputProperty inputProperty, RowType outputType, - boolean lookupKeyContainsPrimaryKey, - boolean upsertMaterialize, - @Nullable LookupJoinHintSpec lookupJoinHintSpec, String description) { this( ExecNodeContext.newNodeId(), @@ -92,12 +92,13 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin implements Stream lookupKeys, projectionOnTemporalTable, filterOnTemporalTable, + lookupKeyContainsPrimaryKey, + upsertMaterialize, + asyncLookupOptions, + retryOptions, inputChangelogMode, Collections.singletonList(inputProperty), outputType, - lookupKeyContainsPrimaryKey, - upsertMaterialize, - lookupJoinHintSpec, description); } @@ -115,14 +116,17 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin implements Stream List<RexNode> projectionOnTemporalTable, @JsonProperty(FIELD_NAME_FILTER_ON_TEMPORAL_TABLE) @Nullable RexNode filterOnTemporalTable, + @JsonProperty(FIELD_NAME_LOOKUP_KEY_CONTAINS_PRIMARY_KEY) + boolean lookupKeyContainsPrimaryKey, + @JsonProperty(FIELD_NAME_REQUIRE_UPSERT_MATERIALIZE) boolean upsertMaterialize, + @JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable + LookupJoinUtil.AsyncLookupOptions asyncLookupOptions, + @JsonProperty(FIELD_NAME_RETRY_OPTIONS) @Nullable + LookupJoinUtil.RetryLookupOptions retryOptions, @JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE) @Nullable ChangelogMode inputChangelogMode, @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties, @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType, - @JsonProperty(FIELD_NAME_LOOKUP_KEY_CONTAINS_PRIMARY_KEY) - boolean lookupKeyContainsPrimaryKey, - @JsonProperty(FIELD_NAME_REQUIRE_UPSERT_MATERIALIZE) boolean upsertMaterialize, - @JsonProperty(FIELD_NAME_JOIN_HINT) @Nullable LookupJoinHintSpec lookupJoinHintSpec, @JsonProperty(FIELD_NAME_DESCRIPTION) String description) { super( id, @@ -134,10 +138,11 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin implements Stream lookupKeys, projectionOnTemporalTable, filterOnTemporalTable, + asyncLookupOptions, + retryOptions, inputChangelogMode, inputProperties, outputType, - lookupJoinHintSpec, description); this.lookupKeyContainsPrimaryKey = lookupKeyContainsPrimaryKey; this.upsertMaterialize = upsertMaterialize; diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java index ad4769072ac..55399ba9b45 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java @@ -20,8 +20,13 @@ package org.apache.flink.table.planner.plan.utils; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.io.InputFormat; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.datastream.AsyncDataStream; +import org.apache.flink.streaming.util.retryable.RetryPredicates; +import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.AsyncTableFunctionProvider; import org.apache.flink.table.connector.source.InputFormatProvider; import org.apache.flink.table.connector.source.LookupTableSource; @@ -39,7 +44,7 @@ import org.apache.flink.table.functions.AsyncLookupFunction; import org.apache.flink.table.functions.LookupFunction; import org.apache.flink.table.functions.UserDefinedFunction; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec; +import org.apache.flink.table.planner.hint.LookupJoinHintOptions; import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable; import org.apache.flink.table.planner.plan.schema.TableSourceTable; import org.apache.flink.table.runtime.connector.source.LookupRuntimeProviderContext; @@ -57,9 +62,11 @@ import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; import org.apache.flink.table.sources.LookupableTableSource; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; import org.apache.flink.util.Preconditions; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonSubTypes; @@ -67,15 +74,28 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTyp import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName; import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.hint.RelHint; import org.apache.calcite.rex.RexLiteral; +import javax.annotation.Nullable; + +import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Objects; import java.util.stream.IntStream; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_CAPACITY; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_LOOKUP; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_OUTPUT_MODE; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.ASYNC_TIMEOUT; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.FIXED_DELAY; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.MAX_ATTEMPTS; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.RETRY_PREDICATE; +import static org.apache.flink.table.planner.hint.LookupJoinHintOptions.RETRY_STRATEGY; import static org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy.NO_RETRY_STRATEGY; +import static org.apache.flink.util.Preconditions.checkNotNull; /** Utilities for lookup joins using {@link LookupTableSource}. */ @Internal @@ -165,29 +185,147 @@ public final class LookupJoinUtil { } } - private LookupJoinUtil() { - // no instantiation - } - /** AsyncLookupOptions includes async related options. */ + @JsonIgnoreProperties(ignoreUnknown = true) + @JsonTypeName("AsyncOptions") public static class AsyncLookupOptions { + public static final String FIELD_NAME_CAPACITY = "capacity "; + public static final String FIELD_NAME_TIMEOUT = "timeout"; + public static final String FIELD_NAME_OUTPUT_MODE = "output-mode"; + + @JsonProperty(FIELD_NAME_CAPACITY) public final int asyncBufferCapacity; + + @JsonProperty(FIELD_NAME_TIMEOUT) public final long asyncTimeout; - public final ExecutionConfigOptions.AsyncOutputMode asyncOutputMode; + @JsonProperty(FIELD_NAME_OUTPUT_MODE) + public final AsyncDataStream.OutputMode asyncOutputMode; + + @JsonCreator public AsyncLookupOptions( - int asyncBufferCapacity, - long asyncTimeout, - ExecutionConfigOptions.AsyncOutputMode asyncOutputMode) { + @JsonProperty(FIELD_NAME_CAPACITY) int asyncBufferCapacity, + @JsonProperty(FIELD_NAME_TIMEOUT) long asyncTimeout, + @JsonProperty(FIELD_NAME_OUTPUT_MODE) AsyncDataStream.OutputMode asyncOutputMode) { this.asyncBufferCapacity = asyncBufferCapacity; this.asyncTimeout = asyncTimeout; this.asyncOutputMode = asyncOutputMode; } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + AsyncLookupOptions that = (AsyncLookupOptions) o; + return asyncBufferCapacity == that.asyncBufferCapacity + && asyncTimeout == that.asyncTimeout + && asyncOutputMode == that.asyncOutputMode; + } + + @Override + public int hashCode() { + return Objects.hash(asyncBufferCapacity, asyncTimeout, asyncOutputMode); + } } - private static class LookupFunctionCandidates { - UserDefinedFunction syncLookupFunction; - UserDefinedFunction asyncLookupFunction; + /** RetryOptions includes retry lookup related options. */ + @JsonIgnoreProperties(ignoreUnknown = true) + @JsonTypeName("RetryLookupOptions") + public static class RetryLookupOptions { + public static final String FIELD_NAME_RETRY_PREDICATE = "retry-predicate"; + public static final String FIELD_NAME_RETRY_STRATEGY = "retry-strategy"; + public static final String FIELD_NAME_RETRY_FIXED_DELAY = "fixed-delay"; + public static final String FIELD_NAME_RETRY_MAX_ATTEMPTS = "max-attempts"; + + @JsonProperty(FIELD_NAME_RETRY_PREDICATE) + private final String retryPredicate; + + @JsonProperty(FIELD_NAME_RETRY_STRATEGY) + private final LookupJoinHintOptions.RetryStrategy retryStrategy; + + @JsonProperty(FIELD_NAME_RETRY_FIXED_DELAY) + private final Long retryFixedDelay; + + @JsonProperty(FIELD_NAME_RETRY_MAX_ATTEMPTS) + private final Integer retryMaxAttempts; + + @JsonCreator + public RetryLookupOptions( + @JsonProperty(FIELD_NAME_RETRY_PREDICATE) String retryPredicate, + @JsonProperty(FIELD_NAME_RETRY_STRATEGY) + LookupJoinHintOptions.RetryStrategy retryStrategy, + @JsonProperty(FIELD_NAME_RETRY_FIXED_DELAY) Long retryFixedDelay, + @JsonProperty(FIELD_NAME_RETRY_MAX_ATTEMPTS) Integer retryMaxAttempts) { + this.retryPredicate = checkNotNull(retryPredicate); + this.retryStrategy = checkNotNull(retryStrategy); + this.retryFixedDelay = checkNotNull(retryFixedDelay); + this.retryMaxAttempts = checkNotNull(retryMaxAttempts); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + RetryLookupOptions that = (RetryLookupOptions) o; + return Objects.equals(retryPredicate, that.retryPredicate) + && retryStrategy == that.retryStrategy + && Objects.equals(retryFixedDelay, that.retryFixedDelay) + && Objects.equals(retryMaxAttempts, that.retryMaxAttempts); + } + + @Override + public int hashCode() { + return Objects.hash(retryPredicate, retryStrategy, retryFixedDelay, retryMaxAttempts); + } + + + @Nullable + public static RetryLookupOptions fromJoinHint(@Nullable RelHint lookupJoinHint) { + if (null != lookupJoinHint) { + Configuration conf = Configuration.fromMap(lookupJoinHint.kvOptions); + Duration fixedDelay = conf.get(FIXED_DELAY); + if (fixedDelay != null) { + return new RetryLookupOptions( + conf.get(RETRY_PREDICATE), + conf.get(RETRY_STRATEGY), + fixedDelay.toMillis(), + conf.get(MAX_ATTEMPTS)); + } + } + return null; + } + + /** + * Convert this {@link RetryLookupOptions} to {@link ResultRetryStrategy} in the best effort + * manner. If invalid {@link LookupJoinHintOptions#RETRY_PREDICATE} or {@link + * LookupJoinHintOptions#RETRY_STRATEGY} is given, then {@link + * ResultRetryStrategy#NO_RETRY_STRATEGY} will return. + */ + @JsonIgnore + @SuppressWarnings("unchecked") + public ResultRetryStrategy toRetryStrategy() { + if (!LookupJoinHintOptions.LOOKUP_MISS_PREDICATE.equalsIgnoreCase(retryPredicate) + || retryStrategy != LookupJoinHintOptions.RetryStrategy.FIXED_DELAY) { + return NO_RETRY_STRATEGY; + } + // retry option values have been validated by hint checker + return ResultRetryStrategy.fixedDelayRetry( + this.retryMaxAttempts, + this.retryFixedDelay, + RetryPredicates.EMPTY_RESULT_PREDICATE); + } + } + + private LookupJoinUtil() { + // no instantiation } /** Gets lookup keys sorted by index in ascending order. */ @@ -197,78 +335,161 @@ public final class LookupJoinUtil { return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray(); } + public static AsyncLookupOptions getMergedAsyncOptions( + RelHint lookupHint, TableConfig config, ChangelogMode inputChangelogMode) { + Configuration confFromHint; + if (lookupHint == null) { + confFromHint = new Configuration(); + } else { + confFromHint = Configuration.fromMap(lookupHint.kvOptions); + } + return new AsyncLookupOptions( + coalesce( + confFromHint.get(ASYNC_CAPACITY), + config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY)), + coalesce( + confFromHint.get(ASYNC_TIMEOUT), + config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_TIMEOUT)) + .toMillis(), + convert( + inputChangelogMode, + coalesce( + confFromHint.get(ASYNC_OUTPUT_MODE), + config.get( + ExecutionConfigOptions + .TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE)))); + } + /** - * Gets lookup function (async or sync) from temporal table according to the given lookup keys - * with considering {@link LookupJoinHintSpec} and required upsertMaterialize. Note: if required - * upsertMaterialize is true, will return synchronous lookup function only, otherwise prefers - * asynchronous lookup function except there's a hint option 'async' = 'false', will raise an - * error if both candidates not found. + * This method determines whether async lookup is enabled according to the given lookup keys + * with considering lookup {@link RelHint} and required upsertMaterialize. Note: it will not + * create the function instance to avoid potential heavy cost during optimization phase. if + * required upsertMaterialize is true, will return synchronous lookup function only, otherwise + * prefers asynchronous lookup function except there's a hint option 'async' = 'false', will + * raise an error if both candidates not found. * * <pre>{@code - * 1. if upsertMaterialize == true : require sync lookup or else error + * 1. if upsertMaterialize == true : return false * * 2. preferAsync = except there is a hint option 'async' = 'false' * if (preferAsync) { - * async lookup != null ? async : sync or else error + * return asyncFound ? true : false * } else { - * sync lookup != null ? sync : async or else error + * return syncFound ? false : true * } * }</pre> */ - public static UserDefinedFunction getLookupFunction( + public static boolean isAsyncLookup( RelOptTable temporalTable, Collection<Integer> lookupKeys, - ClassLoader classLoader, - LookupJoinHintSpec joinHintSpec, + RelHint lookupHint, boolean upsertMaterialize) { - // async & sync lookup candidates - LookupFunctionCandidates lookupFunctionCandidates = new LookupFunctionCandidates(); - // prefer (not require) by default - boolean asyncLookup = LookupJoinUtil.preferAsync(joinHintSpec); - boolean require = false; + boolean preferAsync = preferAsync(lookupHint); if (upsertMaterialize) { // upsertMaterialize only works on sync lookup mode, async lookup is unsupported. - require = true; - asyncLookup = false; + return false; + } + boolean syncFound = false; + boolean asyncFound = false; + if (temporalTable instanceof TableSourceTable) { + int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys); + LookupTableSource.LookupRuntimeProvider provider = + createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder); + if (provider instanceof LookupFunctionProvider + || provider instanceof TableFunctionProvider) { + syncFound = true; + } + if (provider instanceof AsyncLookupFunctionProvider + || provider instanceof AsyncTableFunctionProvider) { + asyncFound = true; + } + } else if (temporalTable instanceof LegacyTableSourceTable) { + LegacyTableSourceTable<?> legacyTableSourceTable = + (LegacyTableSourceTable<?>) temporalTable; + LookupableTableSource<?> tableSource = + (LookupableTableSource<?>) legacyTableSourceTable.tableSource(); + if (tableSource.isAsyncEnabled()) { + asyncFound = true; + } else { + syncFound = true; + } } + if (!syncFound && !asyncFound) { + throw new TableException( + String.format( + "table %s is neither TableSourceTable not LegacyTableSourceTable", + temporalTable.getQualifiedName())); + } + return preferAsync ? asyncFound : !syncFound; + } + /** + * Gets required lookup function (async or sync) from temporal table , will raise an error if + * specified lookup function instance not found. + */ + public static UserDefinedFunction getLookupFunction( + RelOptTable temporalTable, + Collection<Integer> lookupKeys, + ClassLoader classLoader, + boolean async, + ResultRetryStrategy retryStrategy) { + UserDefinedFunction lookupFunction = null; int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys); if (temporalTable instanceof TableSourceTable) { - findLookupFunctionFromNewSource( - (TableSourceTable) temporalTable, - lookupKeyIndicesInOrder, - joinHintSpec, - classLoader, - lookupFunctionCandidates); - } - if (temporalTable instanceof LegacyTableSourceTable) { - findLookupFunctionFromLegacySource( - (LegacyTableSourceTable<?>) temporalTable, - lookupKeyIndicesInOrder, - lookupFunctionCandidates); + lookupFunction = + findLookupFunctionFromNewSource( + (TableSourceTable) temporalTable, + lookupKeyIndicesInOrder, + retryStrategy, + async, + classLoader); + } else if (temporalTable instanceof LegacyTableSourceTable) { + lookupFunction = + findLookupFunctionFromLegacySource( + (LegacyTableSourceTable<?>) temporalTable, + lookupKeyIndicesInOrder, + async); } - UserDefinedFunction selectLookupFunction = - selectLookupFunction( - lookupFunctionCandidates.asyncLookupFunction, - lookupFunctionCandidates.syncLookupFunction, - require, - asyncLookup); - - if (null == selectLookupFunction) { + if (null == lookupFunction) { StringBuilder errorMsg = new StringBuilder(); - if (require) { - errorMsg.append("Required ") - .append(asyncLookup ? "async" : "sync") - .append(" lookup function by planner, but "); - } - errorMsg.append("table ") + errorMsg.append("Required ") + .append(async ? "async" : "sync") + .append(" lookup function by planner, but table ") .append(temporalTable.getQualifiedName()) .append( "does not offer a valid lookup function neither as TableSourceTable nor LegacyTableSourceTable"); throw new TableException(errorMsg.toString()); } - return selectLookupFunction; + return lookupFunction; + } + + /** + * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async + * option in hint is false. + */ + private static boolean preferAsync(@Nullable RelHint lookupHint) { + // async option has no default value, prefer async except async option is false + if (null == lookupHint) { + return true; + } + Configuration conf = Configuration.fromMap(lookupHint.kvOptions); + Boolean async = conf.get(ASYNC_LOOKUP); + return null == async || async; + } + + private static <T> T coalesce(T t1, T t2) { + return t1 != null ? t1 : t2; + } + + private static AsyncDataStream.OutputMode convert( + ChangelogMode inputChangelogMode, + ExecutionConfigOptions.AsyncOutputMode asyncOutputMode) { + if (inputChangelogMode.containsOnly(RowKind.INSERT) + && asyncOutputMode == ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED) { + return AsyncDataStream.OutputMode.UNORDERED; + } + return AsyncDataStream.OutputMode.ORDERED; } /** @@ -276,13 +497,10 @@ public final class LookupJoinUtil { * LookupFunction is supported. */ private static LookupFunction wrapSyncRetryDelegator( - LookupFunctionProvider provider, LookupJoinHintSpec joinHintSpec) { - if (joinHintSpec != null) { - ResultRetryStrategy retryStrategy = joinHintSpec.toRetryStrategy(); - if (retryStrategy != NO_RETRY_STRATEGY) { - return new RetryableLookupFunctionDelegator( - provider.createLookupFunction(), joinHintSpec.toRetryStrategy()); - } + LookupFunctionProvider provider, ResultRetryStrategy retryStrategy) { + if (retryStrategy != null && retryStrategy != NO_RETRY_STRATEGY) { + return new RetryableLookupFunctionDelegator( + provider.createLookupFunction(), retryStrategy); } return provider.createLookupFunction(); } @@ -292,109 +510,74 @@ public final class LookupJoinUtil { * Note: only AsyncLookupFunction is supported. */ private static AsyncLookupFunction wrapASyncRetryDelegator( - AsyncLookupFunctionProvider provider, LookupJoinHintSpec joinHintSpec) { - if (joinHintSpec != null) { - ResultRetryStrategy retryStrategy = joinHintSpec.toRetryStrategy(); - if (retryStrategy != NO_RETRY_STRATEGY) { - return new RetryableAsyncLookupFunctionDelegator( - provider.createAsyncLookupFunction(), joinHintSpec.toRetryStrategy()); - } + AsyncLookupFunctionProvider provider, ResultRetryStrategy retryStrategy) { + if (retryStrategy != null && retryStrategy != NO_RETRY_STRATEGY) { + return new RetryableAsyncLookupFunctionDelegator( + provider.createAsyncLookupFunction(), retryStrategy); } return provider.createAsyncLookupFunction(); } - private static void findLookupFunctionFromNewSource( + private static UserDefinedFunction findLookupFunctionFromNewSource( TableSourceTable temporalTable, int[] lookupKeyIndicesInOrder, - LookupJoinHintSpec joinHintSpec, - ClassLoader classLoader, - LookupFunctionCandidates lookupFunctionCandidates) { - UserDefinedFunction syncLookupFunction = null; - UserDefinedFunction asyncLookupFunction = null; - - // TODO: support nested lookup keys in the future, - // currently we only support top-level lookup keys - int[][] indices = - IntStream.of(lookupKeyIndicesInOrder) - .mapToObj(i -> new int[] {i}) - .toArray(int[][]::new); - - LookupTableSource tableSource = (LookupTableSource) temporalTable.tableSource(); - LookupRuntimeProviderContext providerContext = new LookupRuntimeProviderContext(indices); + ResultRetryStrategy retryStrategy, + boolean async, + ClassLoader classLoader) { LookupTableSource.LookupRuntimeProvider provider = - tableSource.getLookupRuntimeProvider(providerContext); - - if (provider instanceof LookupFunctionProvider) { - if (provider instanceof PartialCachingLookupProvider) { - PartialCachingLookupProvider partialCachingLookupProvider = - (PartialCachingLookupProvider) provider; - syncLookupFunction = - new CachingLookupFunction( - partialCachingLookupProvider.getCache(), - wrapSyncRetryDelegator(partialCachingLookupProvider, joinHintSpec)); - } else if (provider instanceof FullCachingLookupProvider) { - FullCachingLookupProvider fullCachingLookupProvider = - (FullCachingLookupProvider) provider; - RowType tableSourceRowType = - FlinkTypeFactory.toLogicalRowType(temporalTable.getRowType()); - LookupFullCache fullCache = - createFullCache( - fullCachingLookupProvider, - lookupKeyIndicesInOrder, - classLoader, - tableSourceRowType); - // retry on fullCachingLookupFunction is meaningless - syncLookupFunction = - new CachingLookupFunction( - fullCache, fullCachingLookupProvider.createLookupFunction()); - } else { - syncLookupFunction = - wrapSyncRetryDelegator((LookupFunctionProvider) provider, joinHintSpec); + createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder); + + if (async) { + if (provider instanceof AsyncLookupFunctionProvider) { + if (provider instanceof PartialCachingAsyncLookupProvider) { + PartialCachingAsyncLookupProvider partialCachingLookupProvider = + (PartialCachingAsyncLookupProvider) provider; + return new CachingAsyncLookupFunction( + partialCachingLookupProvider.getCache(), + wrapASyncRetryDelegator(partialCachingLookupProvider, retryStrategy)); + } else { + return wrapASyncRetryDelegator( + (AsyncLookupFunctionProvider) provider, retryStrategy); + } } - } - if (provider instanceof AsyncLookupFunctionProvider) { - if (provider instanceof PartialCachingAsyncLookupProvider) { - PartialCachingAsyncLookupProvider partialCachingLookupProvider = - (PartialCachingAsyncLookupProvider) provider; - asyncLookupFunction = - new CachingAsyncLookupFunction( - partialCachingLookupProvider.getCache(), - wrapASyncRetryDelegator( - partialCachingLookupProvider, joinHintSpec)); - } else { - asyncLookupFunction = - wrapASyncRetryDelegator( - (AsyncLookupFunctionProvider) provider, joinHintSpec); + if (provider instanceof AsyncTableFunctionProvider) { + return ((AsyncTableFunctionProvider<?>) provider).createAsyncTableFunction(); + } + } else { + if (provider instanceof LookupFunctionProvider) { + if (provider instanceof PartialCachingLookupProvider) { + PartialCachingLookupProvider partialCachingLookupProvider = + (PartialCachingLookupProvider) provider; + return new CachingLookupFunction( + partialCachingLookupProvider.getCache(), + wrapSyncRetryDelegator(partialCachingLookupProvider, retryStrategy)); + } else if (provider instanceof FullCachingLookupProvider) { + FullCachingLookupProvider fullCachingLookupProvider = + (FullCachingLookupProvider) provider; + RowType tableSourceRowType = + FlinkTypeFactory.toLogicalRowType(temporalTable.getRowType()); + LookupFullCache fullCache = + createFullCache( + fullCachingLookupProvider, + lookupKeyIndicesInOrder, + classLoader, + tableSourceRowType); + // retry on fullCachingLookupFunction is meaningless + return new CachingLookupFunction( + fullCache, fullCachingLookupProvider.createLookupFunction()); + } else { + return wrapSyncRetryDelegator((LookupFunctionProvider) provider, retryStrategy); + } + } + if (provider instanceof TableFunctionProvider) { + return ((TableFunctionProvider<?>) provider).createTableFunction(); } } - if (provider instanceof TableFunctionProvider) { - syncLookupFunction = ((TableFunctionProvider<?>) provider).createTableFunction(); - } - if (provider instanceof AsyncTableFunctionProvider) { - asyncLookupFunction = - ((AsyncTableFunctionProvider<?>) provider).createAsyncTableFunction(); - } - setLookupFunctions(lookupFunctionCandidates, asyncLookupFunction, syncLookupFunction); - } - - private static void setLookupFunctions( - LookupFunctionCandidates lookupFunctionCandidates, - UserDefinedFunction asyncLookupFunction, - UserDefinedFunction syncLookupFunction) { - if (asyncLookupFunction != null) { - lookupFunctionCandidates.asyncLookupFunction = asyncLookupFunction; - } - if (syncLookupFunction != null) { - lookupFunctionCandidates.syncLookupFunction = syncLookupFunction; - } + return null; } - private static void findLookupFunctionFromLegacySource( - LegacyTableSourceTable temporalTable, - int[] lookupKeyIndicesInOrder, - LookupFunctionCandidates lookupFunctionCandidates) { - UserDefinedFunction syncLookupFunction = null; - UserDefinedFunction asyncLookupFunction = null; + private static UserDefinedFunction findLookupFunctionFromLegacySource( + LegacyTableSourceTable temporalTable, int[] lookupKeyIndicesInOrder, boolean async) { String[] lookupFieldNamesInOrder = IntStream.of(lookupKeyIndicesInOrder) .mapToObj(temporalTable.getRowType().getFieldNames()::get) @@ -403,70 +586,25 @@ public final class LookupJoinUtil { (LegacyTableSourceTable<?>) temporalTable; LookupableTableSource<?> tableSource = (LookupableTableSource<?>) legacyTableSourceTable.tableSource(); - if (tableSource.isAsyncEnabled()) { - asyncLookupFunction = tableSource.getAsyncLookupFunction(lookupFieldNamesInOrder); - } - syncLookupFunction = tableSource.getLookupFunction(lookupFieldNamesInOrder); - setLookupFunctions(lookupFunctionCandidates, asyncLookupFunction, syncLookupFunction); - } - - private static UserDefinedFunction selectLookupFunction( - UserDefinedFunction asyncLookupFunction, - UserDefinedFunction syncLookupFunction, - boolean require, - boolean async) { - if (require) { - return async ? asyncLookupFunction : syncLookupFunction; - } else { - if (async) { - // prefer async - return null != asyncLookupFunction ? asyncLookupFunction : syncLookupFunction; - } - // prefer sync - return null != syncLookupFunction ? syncLookupFunction : asyncLookupFunction; + // respect the definition of LookupableTableSource#isAsyncEnabled + if (async && tableSource.isAsyncEnabled()) { + return tableSource.getAsyncLookupFunction(lookupFieldNamesInOrder); } - } - - public static boolean preferAsync(LookupJoinHintSpec lookupJoinHintSpec) { - // async option has no default value, prefer async except async option is false - return null == lookupJoinHintSpec - || null == lookupJoinHintSpec.getAsync() - || lookupJoinHintSpec.isAsync(); - } - - public static boolean isAsyncLookup( - RelOptTable temporalTable, - Collection<Integer> lookupKeys, - LookupJoinHintSpec lookupJoinHintSpec) { - boolean preferAsync = preferAsync(lookupJoinHintSpec); - if (temporalTable instanceof TableSourceTable) { - LookupTableSource.LookupRuntimeProvider provider = - getLookupRuntimeProvider(temporalTable, lookupKeys); - return preferAsync - && (provider instanceof AsyncLookupFunctionProvider - || provider instanceof AsyncTableFunctionProvider); - } else if (temporalTable instanceof LegacyTableSourceTable) { - LegacyTableSourceTable<?> legacyTableSourceTable = - (LegacyTableSourceTable<?>) temporalTable; - LookupableTableSource<?> lookupableTableSource = - (LookupableTableSource<?>) legacyTableSourceTable.tableSource(); - return preferAsync && lookupableTableSource.isAsyncEnabled(); + if (!async && !tableSource.isAsyncEnabled()) { + return tableSource.getLookupFunction(lookupFieldNamesInOrder); } - throw new TableException( - String.format( - "table %s is neither TableSourceTable not LegacyTableSourceTable", - temporalTable.getQualifiedName())); + return null; } - private static LookupTableSource.LookupRuntimeProvider getLookupRuntimeProvider( - RelOptTable temporalTable, Collection<Integer> lookupKeys) { - int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys); + private static LookupTableSource.LookupRuntimeProvider createLookupRuntimeProvider( + RelOptTable temporalTable, int[] lookupKeyIndicesInOrder) { // TODO: support nested lookup keys in the future, // currently we only support top-level lookup keys int[][] indices = IntStream.of(lookupKeyIndicesInOrder) .mapToObj(i -> new int[] {i}) .toArray(int[][]::new); + LookupTableSource tableSource = (LookupTableSource) ((TableSourceTable) temporalTable).tableSource(); LookupRuntimeProviderContext providerContext = new LookupRuntimeProviderContext(indices); diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLookupJoin.scala index a98b4613349..1dfbe46cf64 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLookupJoin.scala @@ -24,7 +24,6 @@ import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSp import org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin import org.apache.flink.table.planner.plan.utils.{FlinkRexUtil, JoinTypeUtil} import org.apache.flink.table.planner.utils.JavaScalaConversionUtil -import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig import org.apache.calcite.plan.{RelOptCluster, RelOptTable, RelTraitSet} import org.apache.calcite.rel.RelNode @@ -75,13 +74,14 @@ class BatchPhysicalLookupJoin( } new BatchExecLookupJoin( - unwrapTableConfig(this), + tableConfig, JoinTypeUtil.getFlinkJoinType(joinType), remainingCondition.orNull, new TemporalTableSourceSpec(temporalTable), allLookupKeys.map(item => (Int.box(item._1), item._2)).asJava, projectionOnTemporalTable, filterOnTemporalTable, + asyncOptions.orNull, InputProperty.DEFAULT, FlinkTypeFactory.toLogicalRowType(getRowType), getRelDetailedDescription) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala index c5fde386f73..36c9211262c 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala @@ -17,23 +17,27 @@ */ package org.apache.flink.table.planner.plan.nodes.physical.common -import org.apache.flink.table.api.TableException +import org.apache.flink.table.api.{TableConfig, TableException} import org.apache.flink.table.catalog.{ObjectIdentifier, UniqueConstraint} +import org.apache.flink.table.connector.ChangelogMode import org.apache.flink.table.planner.calcite.FlinkTypeFactory import org.apache.flink.table.planner.plan.nodes.FlinkRelNode -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRel import org.apache.flink.table.planner.plan.schema.{IntermediateRelTable, LegacyTableSourceTable, TableSourceTable} -import org.apache.flink.table.planner.plan.utils.{ExpressionFormat, JoinTypeUtil, LookupJoinUtil, RelExplainUtil} +import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, ExpressionFormat, JoinTypeUtil, LookupJoinUtil, RelExplainUtil} import org.apache.flink.table.planner.plan.utils.ExpressionFormat.ExpressionFormat import org.apache.flink.table.planner.plan.utils.LookupJoinUtil._ import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat +import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig import org.apache.flink.table.runtime.types.PlannerTypeUtils import org.apache.calcite.plan.{RelOptCluster, RelOptTable, RelTraitSet} +import org.apache.calcite.plan.hep.HepRelVertex import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField} import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel} import org.apache.calcite.rel.core.{JoinInfo, JoinRelType, TableScan} +import org.apache.calcite.rel.hint.RelHint import org.apache.calcite.rex._ import org.apache.calcite.sql.{SqlExplainLevel, SqlKind} import org.apache.calcite.sql.fun.SqlStdOperatorTable @@ -83,7 +87,8 @@ abstract class CommonPhysicalLookupJoin( val calcOnTemporalTable: Option[RexProgram], val joinInfo: JoinInfo, val joinType: JoinRelType, - val lookupHintSpec: Option[LookupJoinHintSpec] = Option.empty[LookupJoinHintSpec]) + val lookupHint: Option[RelHint] = Option.empty[RelHint], + val upsertMaterialize: Boolean = false) extends SingleRel(cluster, traitSet, inputRel) with FlinkRelNode { @@ -108,6 +113,27 @@ abstract class CommonPhysicalLookupJoin( "e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)") } + lazy val isAsyncEnabled: Boolean = LookupJoinUtil.isAsyncLookup( + temporalTable, + allLookupKeys.keys.map(Int.box).toList.asJava, + lookupHint.orNull, + upsertMaterialize) + + lazy val retryOptions: Option[RetryLookupOptions] = + Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull)) + + lazy val inputChangelogMode: ChangelogMode = getInputChangelogMode(getInput) + + lazy val tableConfig: TableConfig = unwrapTableConfig(this); + + lazy val asyncOptions: Option[AsyncLookupOptions] = if (isAsyncEnabled) { + Option.apply( + LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode)) + } else { + // do not create asyncOptions if async is not enabled + Option.empty[AsyncLookupOptions] + } + override def deriveRowType(): RelDataType = { val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory] val rightType = if (calcOnTemporalTable.isDefined) { @@ -125,7 +151,7 @@ abstract class CommonPhysicalLookupJoin( } override def explainTerms(pw: RelWriter): RelWriter = { - val inputFieldNames = inputRel.getRowType.getFieldNames.asScala.toArray + val inputFieldNames = getInput.getRowType.getFieldNames.asScala.toArray val tableFieldNames = temporalTable.getRowType.getFieldNames val resultFieldNames = getRowType.getFieldNames.asScala.toArray val whereString = calcOnTemporalTable match { @@ -161,14 +187,6 @@ abstract class CommonPhysicalLookupJoin( case t: LegacyTableSourceTable[_] => t.tableIdentifier } - // The lookup function maybe not the final choice at runtime because lack of upsert materialize - // info here. This can be consistent after planner offers enough info here. - val isAsyncEnabled: Boolean = - LookupJoinUtil.isAsyncLookup( - temporalTable, - allLookupKeys.keys.map(Int.box).toList.asJava, - lookupHintSpec.orNull) - super .explainTerms(pw) .item("table", tableIdentifier.asSummaryString()) @@ -185,6 +203,17 @@ abstract class CommonPhysicalLookupJoin( pw.getDetailLevel), remainingCondition.isDefined) .item("select", selection) + .itemIf("upsertMaterialize", "true", upsertMaterialize) + } + + private def getInputChangelogMode(rel: RelNode): ChangelogMode = rel match { + case streamPhysicalRel: StreamPhysicalRel => + ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly()) + case hepRelVertex: HepRelVertex => + // there are multi hep-programs in PHYSICAL_REWRITE phase, this would be invoked during + // hep-optimization, so need to deal with HepRelVertex + getInputChangelogMode(hepRelVertex.getCurrentRel) + case _ => ChangelogMode.insertOnly() } /** Gets the remaining join condition which is used */ diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLookupJoin.scala index 6ea85096dc0..54cf2c9ce84 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLookupJoin.scala @@ -19,16 +19,16 @@ package org.apache.flink.table.planner.plan.nodes.physical.stream import org.apache.flink.table.planner.calcite.FlinkTypeFactory import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty} -import org.apache.flink.table.planner.plan.nodes.exec.spec.{LookupJoinHintSpec, TemporalTableSourceSpec} +import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSpec import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLookupJoin import org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin -import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, FlinkRexUtil, JoinTypeUtil} +import org.apache.flink.table.planner.plan.utils.{FlinkRexUtil, JoinTypeUtil} import org.apache.flink.table.planner.utils.JavaScalaConversionUtil -import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig import org.apache.calcite.plan.{RelOptCluster, RelOptTable, RelTraitSet} -import org.apache.calcite.rel.{RelNode, RelWriter} +import org.apache.calcite.rel.RelNode import org.apache.calcite.rel.core.{JoinInfo, JoinRelType} +import org.apache.calcite.rel.hint.RelHint import org.apache.calcite.rex.RexProgram import java.util @@ -44,8 +44,8 @@ class StreamPhysicalLookupJoin( tableCalcProgram: Option[RexProgram], joinInfo: JoinInfo, joinType: JoinRelType, - lookupHintSpec: Option[LookupJoinHintSpec], - val upsertMaterialize: Boolean = false) + lookupHint: Option[RelHint], + upsertMaterialize: Boolean) extends CommonPhysicalLookupJoin( cluster, traitSet, @@ -54,7 +54,8 @@ class StreamPhysicalLookupJoin( tableCalcProgram, joinInfo, joinType, - lookupHintSpec) + lookupHint, + upsertMaterialize) with StreamPhysicalRel { override def requireWatermark: Boolean = false @@ -68,7 +69,7 @@ class StreamPhysicalLookupJoin( tableCalcProgram, joinInfo, joinType, - lookupHintSpec, + lookupHint, upsertMaterialize ) } @@ -82,7 +83,7 @@ class StreamPhysicalLookupJoin( tableCalcProgram, joinInfo, joinType, - lookupHintSpec, + lookupHint, upsertMaterialize ) } @@ -95,29 +96,22 @@ class StreamPhysicalLookupJoin( case _ => (null, null) } - val inputChangelogMode = - ChangelogPlanUtils.getChangelogMode(getInput.asInstanceOf[StreamPhysicalRel]).get new StreamExecLookupJoin( - unwrapTableConfig(this), + tableConfig, JoinTypeUtil.getFlinkJoinType(joinType), remainingCondition.orNull, new TemporalTableSourceSpec(temporalTable), allLookupKeys.map(item => (Int.box(item._1), item._2)).asJava, projectionOnTemporalTable, filterOnTemporalTable, + lookupKeyContainsPrimaryKey(), + upsertMaterialize, + asyncOptions.orNull, + retryOptions.orNull, inputChangelogMode, InputProperty.DEFAULT, FlinkTypeFactory.toLogicalRowType(getRowType), - lookupKeyContainsPrimaryKey(), - upsertMaterialize, - lookupHintSpec.orNull, getRelDetailedDescription) } - - override def explainTerms(pw: RelWriter): RelWriter = { - super - .explainTerms(pw) - .itemIf("upsertMaterialize", "true", upsertMaterialize) - } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLookupJoinRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLookupJoinRule.scala index 85328ea075b..e0c77dbec8c 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLookupJoinRule.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLookupJoinRule.scala @@ -19,13 +19,13 @@ package org.apache.flink.table.planner.plan.rules.physical.stream import org.apache.flink.table.planner.hint.JoinStrategy import org.apache.flink.table.planner.plan.nodes.FlinkConventions -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec import org.apache.flink.table.planner.plan.nodes.logical._ import org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin import org.apache.flink.table.planner.plan.rules.physical.common.{BaseSnapshotOnCalcTableScanRule, BaseSnapshotOnTableScanRule} import org.apache.calcite.plan.{RelOptRule, RelOptTable} +import org.apache.calcite.rel.hint.RelHint import org.apache.calcite.rex.RexProgram /** @@ -80,14 +80,15 @@ object StreamPhysicalLookupJoinRule { val convInput = RelOptRule.convert(input, requiredTrait) - val lookupJoinHint = join.getHints + val lookupRelHint = join.getHints .stream() .filter(hint => JoinStrategy.isLookupHint(hint.hintName)) .findFirst() - val lookupJoinHintSpec = if (lookupJoinHint.isPresent) { - Option.apply(LookupJoinHintSpec.fromJoinHint(lookupJoinHint.get())) + + val lookupHint = if (lookupRelHint.isPresent) { + Option.apply(lookupRelHint.get()) } else { - Option.empty[LookupJoinHintSpec] + Option.empty[RelHint] } new StreamPhysicalLookupJoin( @@ -98,6 +99,7 @@ object StreamPhysicalLookupJoinRule { calcProgram, joinInfo, join.getJoinType, - lookupJoinHintSpec) + lookupHint, + false) } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/alias/ClearLookupJoinHintWithInvalidPropagationShuttleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/alias/ClearLookupJoinHintWithInvalidPropagationShuttleTest.java index 87d39da661c..a261f514452 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/alias/ClearLookupJoinHintWithInvalidPropagationShuttleTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/alias/ClearLookupJoinHintWithInvalidPropagationShuttleTest.java @@ -21,7 +21,7 @@ package org.apache.flink.table.planner.alias; import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; import org.apache.flink.table.planner.hint.FlinkHints; -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpecTest; +import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintTestUtil; import org.apache.flink.table.planner.utils.TableTestUtil; import org.apache.calcite.rel.RelNode; @@ -78,7 +78,7 @@ public class ClearLookupJoinHintWithInvalidPropagationShuttleTest JoinRelType.INNER, builder.equals(builder.field(2, 0, "a"), builder.field(2, 1, "a"))) .project(builder.field(1, 0, "a")) - .hints(LookupJoinHintSpecTest.getLookupJoinHint("lookup", false, true)) + .hints(LookupJoinHintTestUtil.getLookupJoinHint("lookup", false, true)) .build(); verifyRelPlan(root); } @@ -103,7 +103,7 @@ public class ClearLookupJoinHintWithInvalidPropagationShuttleTest JoinRelType.INNER, builder.equals(builder.field(2, 0, "a"), builder.field(2, 1, "a"))) .project(builder.field(1, 0, "a")) - .hints(LookupJoinHintSpecTest.getLookupJoinHint("lookup", false, true)) + .hints(LookupJoinHintTestUtil.getLookupJoinHint("lookup", false, true)) .hints(RelHint.builder(FlinkHints.HINT_ALIAS).hintOption("t1").build()) .scan("src") .snapshot(builder.getRexBuilder().makeCall(FlinkSqlOperatorTable.PROCTIME)) @@ -111,7 +111,7 @@ public class ClearLookupJoinHintWithInvalidPropagationShuttleTest JoinRelType.INNER, builder.equals(builder.field(2, 0, "a"), builder.field(2, 1, "a"))) .project(builder.field(1, 0, "a")) - .hints(LookupJoinHintSpecTest.getLookupJoinHint("lookup", true, false)) + .hints(LookupJoinHintTestUtil.getLookupJoinHint("lookup", true, false)) .build(); verifyRelPlan(root); } @@ -142,7 +142,7 @@ public class ClearLookupJoinHintWithInvalidPropagationShuttleTest JoinRelType.INNER, builder.equals(builder.field(2, 0, "a"), builder.field(2, 1, "a"))) .project(builder.field(1, 0, "a")) - .hints(LookupJoinHintSpecTest.getLookupJoinHint("lookup", true, true)) + .hints(LookupJoinHintTestUtil.getLookupJoinHint("lookup", true, true)) .build(); verifyRelPlan(root); } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java new file mode 100644 index 00000000000..df1369322c6 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.nodes.exec.serde; + +import org.apache.flink.streaming.api.datastream.AsyncDataStream; +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.planner.hint.JoinStrategy; +import org.apache.flink.table.planner.hint.LookupJoinHintOptions; +import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintTestUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; + +import org.apache.calcite.rel.hint.RelHint; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.table.planner.plan.nodes.exec.serde.JsonSerdeTestUtil.testJsonRoundTrip; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Tests for {@link LookupJoinUtil.AsyncLookupOptions}. */ +public class AsyncLookupOptionsTest { + + @Test + void testSerdeAsyncLookupOptions() throws IOException { + LookupJoinUtil.AsyncLookupOptions asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.completeLookupHint, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.lookupHintWithAsync, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.lookupHintWithRetry, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.lookupHintWithTableOnly, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class); + } + + @Test + void testAsyncLookupOptions() { + LookupJoinUtil.AsyncLookupOptions asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.completeLookupHint, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + assertTrue(asyncLookupOptions.asyncOutputMode == AsyncDataStream.OutputMode.UNORDERED); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.completeLookupHint, + TableConfig.getDefault(), + ChangelogMode.all()); + assertTrue(asyncLookupOptions.asyncOutputMode == AsyncDataStream.OutputMode.ORDERED); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + LookupJoinHintTestUtil.lookupHintWithTableOnly, + TableConfig.getDefault(), + ChangelogMode.insertOnly()); + assertTrue(asyncLookupOptions.asyncOutputMode == AsyncDataStream.OutputMode.ORDERED); + assertTrue( + asyncLookupOptions.asyncTimeout + == ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_TIMEOUT + .defaultValue() + .toMillis()); + assertTrue( + asyncLookupOptions.asyncBufferCapacity + == ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY + .defaultValue()); + + TableConfig userConf = TableConfig.getDefault(); + userConf.set( + ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE, + ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED); + userConf.set(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY, 300); + Map<String, String> kvOptions = new HashMap<>(); + kvOptions.put(LookupJoinHintOptions.ASYNC_LOOKUP.key(), "true"); + kvOptions.put(LookupJoinHintOptions.ASYNC_CAPACITY.key(), "1000"); + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + RelHint.builder(JoinStrategy.LOOKUP.getJoinHintName()) + .hintOptions(kvOptions) + .build(), + userConf, + ChangelogMode.insertOnly()); + assertTrue(asyncLookupOptions.asyncOutputMode == AsyncDataStream.OutputMode.UNORDERED); + assertTrue( + asyncLookupOptions.asyncTimeout + == ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_TIMEOUT + .defaultValue() + .toMillis()); + assertTrue(asyncLookupOptions.asyncBufferCapacity == 1000); + + asyncLookupOptions = + LookupJoinUtil.getMergedAsyncOptions( + RelHint.builder(JoinStrategy.LOOKUP.getJoinHintName()) + .hintOptions(kvOptions) + .build(), + userConf, + ChangelogMode.all()); + assertTrue(asyncLookupOptions.asyncOutputMode == AsyncDataStream.OutputMode.ORDERED); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupJoinHintSpecSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupJoinHintSpecSerdeTest.java deleted file mode 100644 index bb5b820f338..00000000000 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupJoinHintSpecSerdeTest.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.table.planner.plan.nodes.exec.serde; - -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec; -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpecTest; - -import org.junit.jupiter.api.Test; - -import java.io.IOException; - -import static org.apache.flink.table.planner.plan.nodes.exec.serde.JsonSerdeTestUtil.testJsonRoundTrip; - -/** Serde tests for {@link LookupJoinHintSpec}. */ -public class LookupJoinHintSpecSerdeTest { - - @Test - void testJoinSpecSerde() throws IOException { - LookupJoinHintSpec lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(LookupJoinHintSpecTest.completeLookupHint); - testJsonRoundTrip(lookupJoinHintSpec, LookupJoinHintSpec.class); - - lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(LookupJoinHintSpecTest.lookupHintWithAsync); - testJsonRoundTrip(lookupJoinHintSpec, LookupJoinHintSpec.class); - - lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(LookupJoinHintSpecTest.lookupHintWithRetry); - testJsonRoundTrip(lookupJoinHintSpec, LookupJoinHintSpec.class); - - lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(LookupJoinHintSpecTest.lookupHintWithTableOnly); - testJsonRoundTrip(lookupJoinHintSpec, LookupJoinHintSpec.class); - } -} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RetryLookupOptionsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RetryLookupOptionsTest.java new file mode 100644 index 00000000000..820c114a519 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RetryLookupOptionsTest.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.nodes.exec.serde; + +import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintTestUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; +import org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; + +import static org.apache.flink.table.planner.plan.nodes.exec.serde.JsonSerdeTestUtil.testJsonRoundTrip; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Tests for {@link LookupJoinUtil.RetryLookupOptions}. */ +public class RetryLookupOptionsTest { + + @Test + void testSerdeRetryLookupOptions() throws IOException { + LookupJoinUtil.RetryLookupOptions retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.completeLookupHint); + testJsonRoundTrip(retryLookupOptions, LookupJoinUtil.RetryLookupOptions.class); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithAsync); + testJsonRoundTrip(retryLookupOptions, LookupJoinUtil.RetryLookupOptions.class); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithRetry); + testJsonRoundTrip(retryLookupOptions, LookupJoinUtil.RetryLookupOptions.class); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithTableOnly); + testJsonRoundTrip(retryLookupOptions, LookupJoinUtil.RetryLookupOptions.class); + } + + @Test + void testToRetryStrategy() { + LookupJoinUtil.RetryLookupOptions retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.completeLookupHint); + assertTrue(retryLookupOptions.toRetryStrategy() != ResultRetryStrategy.NO_RETRY_STRATEGY); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithAsync); + assertTrue(retryLookupOptions == null); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithRetry); + assertTrue(retryLookupOptions.toRetryStrategy() != ResultRetryStrategy.NO_RETRY_STRATEGY); + + retryLookupOptions = + LookupJoinUtil.RetryLookupOptions.fromJoinHint( + LookupJoinHintTestUtil.lookupHintWithTableOnly); + assertTrue(retryLookupOptions == null); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpecTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintTestUtil.java similarity index 67% rename from flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpecTest.java rename to flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintTestUtil.java index 0aa0aae4013..9dc113c560a 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpecTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintTestUtil.java @@ -20,18 +20,14 @@ package org.apache.flink.table.planner.plan.nodes.exec.spec; import org.apache.flink.table.planner.hint.JoinStrategy; import org.apache.flink.table.planner.hint.LookupJoinHintOptions; -import org.apache.flink.table.runtime.operators.join.lookup.ResultRetryStrategy; import org.apache.calcite.rel.hint.RelHint; -import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; -import static org.junit.jupiter.api.Assertions.assertTrue; - -/** Tests for {@link LookupJoinHintSpec}. */ -public class LookupJoinHintSpecTest { +/** Test util for lookup hint. */ +public class LookupJoinHintTestUtil { public static RelHint completeLookupHint = getLookupJoinHint(true, true); public static RelHint lookupHintWithAsync = getLookupJoinHint(true, false); public static RelHint lookupHintWithRetry = getLookupJoinHint(false, true); @@ -68,31 +64,4 @@ public class LookupJoinHintSpecTest { kvOptions.put(LookupJoinHintOptions.FIXED_DELAY.key(), "155 ms"); kvOptions.put(LookupJoinHintOptions.MAX_ATTEMPTS.key(), "10"); } - - @Test - void testJoinHintToRetryStrategy() { - LookupJoinHintSpec lookupJoinHintSpec = LookupJoinHintSpec.fromJoinHint(completeLookupHint); - assertTrue(lookupJoinHintSpec.toRetryStrategy() != ResultRetryStrategy.NO_RETRY_STRATEGY); - } - - @Test - void testJoinHintWithTableOnlyToRetryStrategy() { - LookupJoinHintSpec lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(lookupHintWithTableOnly); - assertTrue(lookupJoinHintSpec.toRetryStrategy() == ResultRetryStrategy.NO_RETRY_STRATEGY); - } - - @Test - void testJoinHintWithAsyncToRetryStrategy() { - LookupJoinHintSpec lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(lookupHintWithAsync); - assertTrue(lookupJoinHintSpec.toRetryStrategy() == ResultRetryStrategy.NO_RETRY_STRATEGY); - } - - @Test - void testJoinHintWithRetryToRetryStrategy() { - LookupJoinHintSpec lookupJoinHintSpec = - LookupJoinHintSpec.fromJoinHint(lookupHintWithRetry); - assertTrue(lookupJoinHintSpec.toRetryStrategy() != ResultRetryStrategy.NO_RETRY_STRATEGY); - } } diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testAggAndLeftJoinWithTryResolveMode.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testAggAndLeftJoinWithTryResolveMode.out index 983e5e01506..a378e26e29e 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testAggAndLeftJoinWithTryResolveMode.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testAggAndLeftJoinWithTryResolveMode.out @@ -304,6 +304,8 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, + "requireUpsertMaterialize" : true, "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER" ], "inputProperties" : [ { "requiredDistribution" : { @@ -313,8 +315,6 @@ "priority" : 0 } ], "outputType" : "ROW<`a` INT, `id` INT, `name` VARCHAR(2147483647), `age` INT>", - "lookupKeyContainsPrimaryKey" : false, - "requireUpsertMaterialize" : true, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], async=[false], lookup=[id=a], select=[a, id, name, age], upsertMaterialize=[true])" }, { "id" : 9, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTable.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTable.out index ecd654ad000..1f8477aefb7 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTable.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTable.out @@ -258,6 +258,7 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,7 +304,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, @@ -473,4 +473,4 @@ }, "shuffleMode" : "PIPELINED" } ] -} +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint.out index 5fb580acb2f..24d394854c1 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint.out @@ -258,6 +258,7 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,13 +304,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, - "joinHint" : { - "async" : true, - "output-mode" : "ALLOW_UNORDERED", - "capacity" : 100, - "timeout" : 180000 - }, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint2.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint2.out index 82c38b72f98..24d394854c1 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint2.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncHint2.out @@ -258,6 +258,7 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,13 +304,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, - "joinHint" : { - "async" : true, - "output-mode" : "ORDERED", - "capacity" : 1000, - "timeout" : 600000 - }, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint.out index a8b0f6a4131..2da272a7bb6 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint.out @@ -258,6 +258,13 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, + "retryOptions" : { + "retry-predicate" : "lookup_miss", + "retry-strategy" : "FIXED_DELAY", + "fixed-delay" : 10000, + "max-attempts" : 3 + }, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,17 +310,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, - "joinHint" : { - "async" : true, - "output-mode" : "ORDERED", - "capacity" : 100, - "timeout" : 180000, - "retry-predicate" : "lookup_miss", - "retry-strategy" : "FIXED_DELAY", - "fixed-delay" : 10000, - "max-attempts" : 3 - }, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint2.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint2.out index 936696bc74e..2da272a7bb6 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint2.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithAsyncRetryHint2.out @@ -258,6 +258,13 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, + "retryOptions" : { + "retry-predicate" : "lookup_miss", + "retry-strategy" : "FIXED_DELAY", + "fixed-delay" : 10000, + "max-attempts" : 3 + }, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,17 +310,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, - "joinHint" : { - "async" : true, - "output-mode" : "ORDERED", - "capacity" : 1000, - "timeout" : 600000, - "retry-predicate" : "lookup_miss", - "retry-strategy" : "FIXED_DELAY", - "fixed-delay" : 10000, - "max-attempts" : 3 - }, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithProjectionPushDown.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithProjectionPushDown.out index b4f2cb7d22d..c3b8ae21edd 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithProjectionPushDown.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithProjectionPushDown.out @@ -262,6 +262,7 @@ "type" : "INT" } ], "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -301,7 +302,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id])" }, { "id" : 5, @@ -457,4 +457,4 @@ }, "shuffleMode" : "PIPELINED" } ] -} +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithRetryHint.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithRetryHint.out index c8dc6c5ba8a..2da272a7bb6 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithRetryHint.out +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest_jsonplan/testJoinTemporalTableWithRetryHint.out @@ -258,6 +258,13 @@ }, "projectionOnTemporalTable" : null, "filterOnTemporalTable" : null, + "lookupKeyContainsPrimaryKey" : false, + "retryOptions" : { + "retry-predicate" : "lookup_miss", + "retry-strategy" : "FIXED_DELAY", + "fixed-delay" : 10000, + "max-attempts" : 3 + }, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ { "requiredDistribution" : { @@ -303,16 +310,6 @@ "fieldType" : "INT" } ] }, - "lookupKeyContainsPrimaryKey" : false, - "joinHint" : { - "output-mode" : "ORDERED", - "capacity" : 100, - "timeout" : 180000, - "retry-predicate" : "lookup_miss", - "retry-strategy" : "FIXED_DELAY", - "fixed-delay" : 10000, - "max-attempts" : 3 - }, "description" : "LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age])" }, { "id" : 5, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml index 4f6a308b9b5..e5f6474d247 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml @@ -16,6 +16,252 @@ See the License for the specific language governing permissions and limitations under the License. --> <Root> + <TestCase name="testAggAndLeftJoinAllowUnordered[LegacyTableSource=false]"> + <Resource name="explain"> + <![CDATA[== Abstract Syntax Tree == +LogicalSink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- LogicalProject(a=[$0], name=[$4], age=[$5]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{0, 2}]) + :- LogicalProject(a=[$1], c=[$2], proctime=[PROCTIME()]) + : +- LogicalAggregate(group=[{0}], a=[MAX($1)], c=[COUNT($2)]) + : +- LogicalProject(b=[$1], a=[$0], c=[$2]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalFilter(condition=[=($cor0.a, $0)]) + +- LogicalSnapshot(period=[$cor0.proctime]) + +- LogicalTableScan(table=[[default_catalog, default_database, AsyncLookupTable]]) + +== Optimized Physical Plan == +Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- Calc(select=[a, name, age]) + +- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100]) + +- Calc(select=[a]) + +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) + +- Exchange(distribution=[hash[b]]) + +- Calc(select=[b, a, c]) + +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) + +== Optimized Execution Plan == +Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- Calc(select=[a, name, age]) + +- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100]) + +- Calc(select=[a]) + +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) + +- Exchange(distribution=[hash[b]]) + +- Calc(select=[b, a, c]) + +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) + +== Physical Execution Plan == +{ + "nodes" : [ { + "id" : , + "type" : "Source: Collection Source", + "pact" : "Data Source", + "contents" : "Source: Collection Source", + "parallelism" : 1 + }, { + "id" : , + "type" : "SourceConversion[]", + "pact" : "Operator", + "contents" : "[]:SourceConversion(table=[default_catalog.default_database.MyTable], fields=[a, b, c, proctime, rowtime])", + "parallelism" : 1, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[b, a, c])", + "parallelism" : 1, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "GroupAggregate[]", + "pact" : "Operator", + "contents" : "[]:GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "HASH", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[a])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "LookupJoin[]", + "pact" : "Operator", + "contents" : "[]:LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[a, name, age])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Sink: Sink1[]", + "pact" : "Data Sink", + "contents" : "[]:Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + } ] +}]]> + </Resource> + </TestCase> + <TestCase name="testAggAndLeftJoinAllowUnordered[LegacyTableSource=true]"> + <Resource name="explain"> + <![CDATA[== Abstract Syntax Tree == +LogicalSink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- LogicalProject(a=[$0], name=[$4], age=[$5]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{0, 2}]) + :- LogicalProject(a=[$1], c=[$2], proctime=[PROCTIME()]) + : +- LogicalAggregate(group=[{0}], a=[MAX($1)], c=[COUNT($2)]) + : +- LogicalProject(b=[$1], a=[$0], c=[$2]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalFilter(condition=[=($cor0.a, $0)]) + +- LogicalSnapshot(period=[$cor0.proctime]) + +- LogicalTableScan(table=[[default_catalog, default_database, AsyncLookupTable, source: [TestTemporalTable(id, name, age)]]]) + +== Optimized Physical Plan == +Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- Calc(select=[a, name, age]) + +- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100]) + +- Calc(select=[a]) + +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) + +- Exchange(distribution=[hash[b]]) + +- Calc(select=[b, a, c]) + +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) + +== Optimized Execution Plan == +Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) ++- Calc(select=[a, name, age]) + +- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100]) + +- Calc(select=[a]) + +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) + +- Exchange(distribution=[hash[b]]) + +- Calc(select=[b, a, c]) + +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) + +== Physical Execution Plan == +{ + "nodes" : [ { + "id" : , + "type" : "Source: Collection Source", + "pact" : "Data Source", + "contents" : "Source: Collection Source", + "parallelism" : 1 + }, { + "id" : , + "type" : "SourceConversion[]", + "pact" : "Operator", + "contents" : "[]:SourceConversion(table=[default_catalog.default_database.MyTable], fields=[a, b, c, proctime, rowtime])", + "parallelism" : 1, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[b, a, c])", + "parallelism" : 1, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "GroupAggregate[]", + "pact" : "Operator", + "contents" : "[]:GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "HASH", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[a])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "LookupJoin[]", + "pact" : "Operator", + "contents" : "[]:LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[LeftOuterJoin], lookup=[id=a], select=[a, id, name, age], async=[ORDERED, 180000ms, 100])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Calc[]", + "pact" : "Operator", + "contents" : "[]:Calc(select=[a, name, age])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + }, { + "id" : , + "type" : "Sink: Sink1[]", + "pact" : "Data Sink", + "contents" : "[]:Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age])", + "parallelism" : 4, + "predecessors" : [ { + "id" : , + "ship_strategy" : "FORWARD", + "side" : "second" + } ] + } ] +}]]> + </Resource> + </TestCase> <TestCase name="testAvoidAggregatePushDown[LegacyTableSource=false]"> <Resource name="sql"> <![CDATA[ @@ -193,7 +439,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam <Resource name="optimized exec plan"> <![CDATA[ Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id, name, age]) -+- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age]) ++- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[InnerJoin], async=[true], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> </Resource> @@ -215,7 +461,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam <Resource name="optimized exec plan"> <![CDATA[ Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id, name, age]) -+- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[InnerJoin], async=[false], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age]) ++- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[InnerJoin], async=[true], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> </Resource> diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala index bf83987afa3..bfb983ca094 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala @@ -33,7 +33,6 @@ import org.apache.flink.table.planner.plan.logical.{LogicalWindow, _} import org.apache.flink.table.planner.plan.nodes.FlinkConventions import org.apache.flink.table.planner.plan.nodes.calcite._ import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalWindowTableFunction -import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintSpec import org.apache.flink.table.planner.plan.nodes.logical._ import org.apache.flink.table.planner.plan.nodes.physical.batch._ import org.apache.flink.table.planner.plan.nodes.physical.stream._ @@ -2627,7 +2626,8 @@ class FlinkRelMdHandlerTestBase { None, JoinInfo.of(ImmutableIntList.of(0), ImmutableIntList.of(0)), JoinRelType.INNER, - Option.empty[LookupJoinHintSpec] + Option.empty[RelHint], + false ) (batchLookupJoin, streamLookupJoin) } diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala index cd4009a1321..c86bb1c07e4 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala @@ -22,7 +22,7 @@ import org.apache.flink.api.java.tuple import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink} import org.apache.flink.table.api._ -import org.apache.flink.table.api.config.OptimizerConfigOptions +import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions} import org.apache.flink.table.api.config.OptimizerConfigOptions.NonDeterministicUpdateStrategy import org.apache.flink.table.api.internal.TableEnvironmentInternal import org.apache.flink.table.data.RowData @@ -58,6 +58,10 @@ class NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUp OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY, nonDeterministicUpdateStrategy) + // for json plan test + util.tableEnv.getConfig + .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, Int.box(4)) + util.addTableSource[(Int, Long, String, Boolean)]("T", 'a, 'b, 'c, 'd) util.addDataStream[(Int, String, Long)]("T1", 'a, 'b, 'c, 'proctime.proctime, 'rowtime.rowtime) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala index d7ebfcb93e2..30c1f61a490 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala @@ -22,7 +22,7 @@ import org.apache.flink.core.testutils.FlinkMatchers.containsMessage import org.apache.flink.streaming.api.datastream.DataStream import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment import org.apache.flink.table.api._ -import org.apache.flink.table.api.config.OptimizerConfigOptions +import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions} import org.apache.flink.table.api.internal.TableEnvironmentInternal import org.apache.flink.table.data.RowData import org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_TYPE @@ -103,6 +103,18 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri |) |""".stripMargin) } + util.addTable(""" + |CREATE TABLE Sink1 ( + | a int, + | name varchar, + | age int + |) with ( + | 'connector' = 'values', + | 'sink-insert-only' = 'false' + |)""".stripMargin) + // for json plan test + util.tableEnv.getConfig + .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, Int.box(4)) } @Test @@ -181,6 +193,7 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri } util.addDataStream[(Int, String, Long, Timestamp)]("T", 'a, 'b, 'c, 'ts, 'proctime.proctime) createLookupTable("LookupTable1", new InvalidTableFunctionResultType) + expectExceptionThrown( "SELECT * FROM T JOIN LookupTable1 " + "FOR SYSTEM_TIME AS OF T.proctime AS D ON T.a = D.id AND T.b = D.name AND T.ts = D.ts", @@ -558,23 +571,11 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri @Test def testAggAndAllConstantLookupKeyWithTryResolveMode(): Unit = { - util.getStreamEnv.setParallelism(4) // expect lookup join using single parallelism due to all constant lookup key util.tableEnv.getConfig.set( OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY, OptimizerConfigOptions.NonDeterministicUpdateStrategy.TRY_RESOLVE) - util.addTable(""" - |CREATE TABLE Sink1 ( - | `id` INT, - | `name` STRING, - | `age` INT - |) WITH ( - | 'connector' = 'values', - | 'sink-insert-only' = 'false' - |) - |""".stripMargin) - val sql = """ |INSERT INTO Sink1 @@ -791,6 +792,25 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri util.verifyExecPlan(sql) } + @Test + def testAggAndLeftJoinAllowUnordered(): Unit = { + util.tableEnv.getConfig.set( + ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE, + ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED) + + val stmt = util.tableEnv.asInstanceOf[TestingTableEnvironment].createStatementSet() + stmt.addInsertSql( + """ + |INSERT INTO Sink1 + |SELECT T.a, D.name, D.age + |FROM (SELECT max(a) a, count(c) c, PROCTIME() proctime FROM MyTable GROUP BY b) T + |LEFT JOIN AsyncLookupTable + |FOR SYSTEM_TIME AS OF T.proctime AS D ON T.a = D.id + |""".stripMargin) + + util.verifyExplain(stmt, ExplainDetail.JSON_EXECUTION_PLAN) + } + // ========================================================================================== private def createLookupTable(tableName: String, lookupFunction: UserDefinedFunction): Unit = { diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala index 06b6d5daa27..ecf8aab94fe 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala @@ -311,12 +311,9 @@ class AsyncLookupJoinITCase( @Test def testAggAndAsyncLeftJoinWithTryResolveMode(): Unit = { // will require a sync lookup function because input has update on TRY_RESOLVE mode - // only legacy source can provide both sync and async functions - if (!legacyTableSource) { - thrown.expectMessage( - "Required sync lookup function by planner, but table [default_catalog, default_database, user_table]does not offer a valid lookup function") - thrown.expect(classOf[TableException]) - } + // there's no test sources that have both sync and async lookup functions + thrown.expectMessage("Required sync lookup function by planner") + thrown.expect(classOf[TableException]) tEnv.getConfig.set( OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY, OptimizerConfigOptions.NonDeterministicUpdateStrategy.TRY_RESOLVE)
