You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/16 07:44:54 UTC
[GitHub] [flink] lincoln-lil opened a new pull request, #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
lincoln-lil opened a new pull request, #20592:
URL: https://github.com/apache/flink/pull/20592
## What is the purpose of the change
This is a minor improvement for the new LOOKUP join hint which updates the description of lookup join transformation with async params (async output-mode, time-out, buffer-capacity) and retry strategy for easier debugging. Using a simplified format: values only, without param names
e.g.,
```
async-params=[ORDERED, 180000ms, 100]
retry=[FIXED_DELAY, 10000ms, 3])
```
## Brief change log
add updateDescription for CommonExecLookupJoin
## Verifying this change
LookupJoinTest with verifying json exec plan
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
- The serializers: (no )
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948639908
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
+ public static AsyncLookupOptions getMergedAsyncOptions(
Review Comment:
It's better we can add some UTs the verify the logic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948706335
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java:
##########
@@ -246,18 +254,23 @@ protected Transformation<RowData> createJoinTransformation(
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;
Review Comment:
This never happens in runtime by design, but to prevent incorrectly changes in LookupJoinUtil#isAsyncLookup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on PR #20592:
URL: https://github.com/apache/flink/pull/20592#issuecomment-1217806914
@godfreyhe thanks for your comments! I've updated the pr with the changes to make exec lookup join node immutable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948708427
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private LookupJoinHintOptions() {}
public static final ConfigOption<Integer> ASYNC_CAPACITY =
key("capacity")
.intType()
- .defaultValue(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY
- .defaultValue())
+ .noDefaultValue()
Review Comment:
I'll add more tests in AsyncLookupOptionsTest to cover this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe closed pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
godfreyhe closed pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
URL: https://github.com/apache/flink/pull/20592
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948688414
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private LookupJoinHintOptions() {}
public static final ConfigOption<Integer> ASYNC_CAPACITY =
key("capacity")
.intType()
- .defaultValue(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY
- .defaultValue())
+ .noDefaultValue()
Review Comment:
the former version has default value references to related ExecutionConfigOptions, but will have incorrect behavior when users configured the corresponding option (differs from default one), and not explicit configured via hint, e.g.,
user set
```
'table.exec.async-lookup.output-mode': ALLOW_UNORDERED
```
and use a LOOKUP hint
```
LOOKUP('table'='dim1', 'async'='true', 'timeout'='300s')
```
the expected final options on lookup table 'dim1' should be
```
output-mode=ALLOW_UNORDERED,
timeout=300s,
buffer-capacity=100
```
but will be wrongly with default value in hint options here:
```
output-mode=ORDERED,
timeout=300s,
buffer-capacity=100
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948940034
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -165,29 +183,177 @@ public int hashCode() {
}
}
- 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);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode + ", " + asyncTimeout + "ms, " + asyncBufferCapacity;
+ }
+ }
+
+ /** 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);
+ }
+
+ @Override
+ public String toString() {
+ return retryPredicate
+ + ", "
+ + retryStrategy
+ + ", "
+ + retryFixedDelay
+ + "ms, "
+ + retryMaxAttempts;
+ }
+
+ public String getRetryPredicate() {
+ return retryPredicate;
+ }
+
+ public LookupJoinHintOptions.RetryStrategy getRetryStrategy() {
+ return retryStrategy;
+ }
+
+ public Long getRetryFixedDelay() {
+ return retryFixedDelay;
+ }
+
+ public Integer getRetryMaxAttempts() {
+ return retryMaxAttempts;
+ }
+
+ public static RetryLookupOptions fromJoinHint(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
+ 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);
Review Comment:
yes, this was because the RetryPredicates.EMPTY_RESULT_PREDICATE can not have a concrete type, I added a unchecked annotation here.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
Review Comment:
ok
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -108,6 +112,31 @@ abstract class CommonPhysicalLookupJoin(
"e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)")
}
+ lazy val isAsyncEnabled = LookupJoinUtil.isAsyncLookup(
+ temporalTable,
+ allLookupKeys.keys.map(Int.box).toList.asJava,
+ lookupHint.orNull,
+ upsertMaterialize)
+
+ lazy val retryOptions =
+ Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull))
+
+ lazy val inputChangelogMode = getInput match {
+ case streamPhysicalRel: StreamPhysicalRel =>
+ ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly())
+ case _ => ChangelogMode.insertOnly()
+ }
+
+ lazy val tableConfig = unwrapTableConfig(this);
+
+ lazy val asyncOptions = if (isAsyncEnabled) {
+ Option.apply(
+ LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode))
+ } else {
+ // do not create asyncOptions if async is not enabled
+ Option.empty[AsyncLookupOptions]
+ }
Review Comment:
ok
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
+ if (temporalTable instanceof TableSourceTable) {
+ LookupTableSource.LookupRuntimeProvider provider =
+ createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder);
+ if (provider instanceof LookupFunctionProvider
+ || provider instanceof TableFunctionProvider) {
+ syncFound = true;
+ }
+ if (provider instanceof AsyncLookupFunctionProvider
+ || provider instanceof AsyncTableFunctionProvider) {
+ asyncFound = true;
+ }
+ }
+ if (temporalTable instanceof LegacyTableSourceTable) {
Review Comment:
yes, a table instance can not be both TableSourceTable and LegacyTableSourceTable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948689432
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
+ public static AsyncLookupOptions getMergedAsyncOptions(
Review Comment:
covered by AsyncLookupOptionsTest#testAsyncLookupOptions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948796456
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private LookupJoinHintOptions() {}
public static final ConfigOption<Integer> ASYNC_CAPACITY =
key("capacity")
.intType()
- .defaultValue(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY
- .defaultValue())
+ .noDefaultValue()
Review Comment:
make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948688751
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -108,6 +112,31 @@ abstract class CommonPhysicalLookupJoin(
"e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)")
}
+ val isAsyncEnabled = LookupJoinUtil.isAsyncLookup(
+ temporalTable,
+ allLookupKeys.keys.map(Int.box).toList.asJava,
+ lookupHint.orNull,
+ upsertMaterialize)
+
+ val retryOptions = Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull))
+
+ val inputChangelogMode = getInput match {
+ case streamPhysicalRel: StreamPhysicalRel =>
+ ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly())
+ case _ => ChangelogMode.insertOnly()
+ }
+
+ val tableConfig = unwrapTableConfig(this);
+
+ val asyncOptions = isAsyncEnabled match {
+ case true =>
+ Option.apply(
+ LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode))
+ case false =>
+ // do not create asyncOptions if async is not enabled
+ Option.empty[AsyncLookupOptions]
+ }
Review Comment:
ok
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
+ public static AsyncLookupOptions getMergedAsyncOptions(
Review Comment:
convered by AsyncLookupOptionsTest#testAsyncLookupOptions
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private LookupJoinHintOptions() {}
public static final ConfigOption<Integer> ASYNC_CAPACITY =
key("capacity")
.intType()
- .defaultValue(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY
- .defaultValue())
+ .noDefaultValue()
Review Comment:
the former version has default value references from related ExecutionConfigOptions, but will incorrect behavior when users configured the corresponding option (differs from default one), and not explicit configured via hint, e.g.,
user set
```
'table.exec.async-lookup.output-mode': ALLOW_UNORDERED
```
and use a LOOKUP hint
```
LOOKUP('table'='dim1', 'async'='true', 'timeout'='300s')
```
the expected final options on lookup table 'dim1' should be
```
output-mode=ALLOW_UNORDERED,
timeout=300s,
buffer-capacity=100
```
but will be wrongly with default value in hint options here:
```
output-mode=ORDERED,
timeout=300s,
buffer-capacity=100
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20592:
URL: https://github.com/apache/flink/pull/20592#issuecomment-1216280716
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "6d9cdc5826e22a925996d802a3c77dfb6526bb77",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6d9cdc5826e22a925996d802a3c77dfb6526bb77",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6d9cdc5826e22a925996d802a3c77dfb6526bb77 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20592: [FLINK-28987][table-planner] Fix incorrect async to sync lookup fallback path of LegacyTableSourceTable
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948806639
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -165,29 +183,177 @@ public int hashCode() {
}
}
- 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);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode + ", " + asyncTimeout + "ms, " + asyncBufferCapacity;
+ }
+ }
+
+ /** 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);
+ }
+
+ @Override
+ public String toString() {
+ return retryPredicate
+ + ", "
+ + retryStrategy
+ + ", "
+ + retryFixedDelay
+ + "ms, "
+ + retryMaxAttempts;
+ }
+
+ public String getRetryPredicate() {
+ return retryPredicate;
+ }
+
+ public LookupJoinHintOptions.RetryStrategy getRetryStrategy() {
+ return retryStrategy;
+ }
+
+ public Long getRetryFixedDelay() {
+ return retryFixedDelay;
+ }
+
+ public Integer getRetryMaxAttempts() {
+ return retryMaxAttempts;
+ }
+
+ public static RetryLookupOptions fromJoinHint(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
+ 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);
Review Comment:
nit: there is a warning in IDE
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
+ public static AsyncLookupOptions getMergedAsyncOptions(
+ RelHint lookupHint, TableConfig config, ChangelogMode inputChangelogMode) {
Review Comment:
@Nullable RelHint lookupHint
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -165,29 +183,177 @@ public int hashCode() {
}
}
- 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);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode + ", " + asyncTimeout + "ms, " + asyncBufferCapacity;
+ }
+ }
+
+ /** 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);
+ }
+
+ @Override
+ public String toString() {
+ return retryPredicate
+ + ", "
+ + retryStrategy
+ + ", "
+ + retryFixedDelay
+ + "ms, "
+ + retryMaxAttempts;
+ }
+
+ public String getRetryPredicate() {
+ return retryPredicate;
+ }
+
+ public LookupJoinHintOptions.RetryStrategy getRetryStrategy() {
+ return retryStrategy;
+ }
+
+ public Long getRetryFixedDelay() {
+ return retryFixedDelay;
+ }
+
+ public Integer getRetryMaxAttempts() {
+ return retryMaxAttempts;
+ }
+
+ public static RetryLookupOptions fromJoinHint(RelHint lookupJoinHint) {
Review Comment:
nit: add `@Nullable ` annotation for the parameter
change the return type to Optional< RetryLookupOptions> ?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -108,6 +112,31 @@ abstract class CommonPhysicalLookupJoin(
"e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)")
}
+ lazy val isAsyncEnabled = LookupJoinUtil.isAsyncLookup(
+ temporalTable,
+ allLookupKeys.keys.map(Int.box).toList.asJava,
+ lookupHint.orNull,
+ upsertMaterialize)
+
+ lazy val retryOptions =
+ Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull))
+
+ lazy val inputChangelogMode = getInput match {
+ case streamPhysicalRel: StreamPhysicalRel =>
+ ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly())
+ case _ => ChangelogMode.insertOnly()
+ }
+
+ lazy val tableConfig = unwrapTableConfig(this);
+
+ lazy val asyncOptions = if (isAsyncEnabled) {
+ Option.apply(
+ LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode))
+ } else {
+ // do not create asyncOptions if async is not enabled
+ Option.empty[AsyncLookupOptions]
+ }
Review Comment:
nit: add field type to let IDE happy
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
+ if (temporalTable instanceof TableSourceTable) {
+ LookupTableSource.LookupRuntimeProvider provider =
+ createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder);
+ if (provider instanceof LookupFunctionProvider
+ || provider instanceof TableFunctionProvider) {
+ syncFound = true;
+ }
+ if (provider instanceof AsyncLookupFunctionProvider
+ || provider instanceof AsyncTableFunctionProvider) {
+ asyncFound = true;
+ }
+ }
+ 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()));
+ }
+ if (preferAsync) {
+ return asyncFound ? true : false;
}
+ return syncFound ? false : true;
Review Comment:
`return preferAsync ? asyncFound : !syncFound;`
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
+ if (temporalTable instanceof TableSourceTable) {
+ LookupTableSource.LookupRuntimeProvider provider =
+ createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder);
+ if (provider instanceof LookupFunctionProvider
+ || provider instanceof TableFunctionProvider) {
+ syncFound = true;
+ }
+ if (provider instanceof AsyncLookupFunctionProvider
+ || provider instanceof AsyncTableFunctionProvider) {
+ asyncFound = true;
+ }
+ }
+ 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()));
+ }
+ if (preferAsync) {
+ return asyncFound ? true : false;
}
+ return syncFound ? false : true;
+ }
+ /**
+ * 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);
+ lookupFunction =
+ findLookupFunctionFromNewSource(
+ (TableSourceTable) temporalTable,
+ lookupKeyIndicesInOrder,
+ retryStrategy,
+ async,
+ classLoader);
}
if (temporalTable instanceof LegacyTableSourceTable) {
Review Comment:
else if
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
Review Comment:
move this line into the following `if` branch
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(RelHint lookupHint) {
Review Comment:
public -> private
@Nullable RelHint lookupHint
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -165,29 +183,177 @@ public int hashCode() {
}
}
- 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);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode + ", " + asyncTimeout + "ms, " + asyncBufferCapacity;
+ }
+ }
+
+ /** 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);
+ }
+
+ @Override
+ public String toString() {
+ return retryPredicate
+ + ", "
+ + retryStrategy
+ + ", "
+ + retryFixedDelay
+ + "ms, "
+ + retryMaxAttempts;
+ }
+
+ public String getRetryPredicate() {
+ return retryPredicate;
+ }
+
+ public LookupJoinHintOptions.RetryStrategy getRetryStrategy() {
+ return retryStrategy;
+ }
+
+ public Long getRetryFixedDelay() {
+ return retryFixedDelay;
+ }
+
+ public Integer getRetryMaxAttempts() {
+ return retryMaxAttempts;
+ }
+
+ public static RetryLookupOptions fromJoinHint(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
+ public ResultRetryStrategy toRetryStrategy() {
+ if (null == retryPredicate
Review Comment:
retryPredicate will not be null, it has been checked in constructor
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
+ }
+
/**
- * 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 = LookupJoinUtil.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;
+ int[] lookupKeyIndicesInOrder = getOrderedLookupKeys(lookupKeys);
+ if (temporalTable instanceof TableSourceTable) {
+ LookupTableSource.LookupRuntimeProvider provider =
+ createLookupRuntimeProvider(temporalTable, lookupKeyIndicesInOrder);
+ if (provider instanceof LookupFunctionProvider
+ || provider instanceof TableFunctionProvider) {
+ syncFound = true;
+ }
+ if (provider instanceof AsyncLookupFunctionProvider
+ || provider instanceof AsyncTableFunctionProvider) {
+ asyncFound = true;
+ }
+ }
+ if (temporalTable instanceof LegacyTableSourceTable) {
Review Comment:
else if
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
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))));
+ }
+
+ /**
+ * Evaluates if prefer async lookup by given lookup {@link RelHint}. Returns true except async
+ * option in hint is false.
+ */
+ public static boolean preferAsync(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.booleanValue();
Review Comment:
boxing is no needed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r947394831
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java:
##########
@@ -411,6 +426,55 @@ private Transformation<RowData> createSyncLookupJoinWithState(
return transform;
}
+ /**
+ * Update description of the transformation for easier debugging during job running. Will do:
+ *
+ * <p>1. correct async lookup mode if fallback to sync lookup
+ *
+ * <p>2. append async lookup parameters
+ *
+ * <p>3. append retry strategy
+ */
+ private String updateDescription(
Review Comment:
could we make this class immutable ? Many field values(such as: isAsyncEnabled, asyncLookupOptions) can be determined when translating StreamPhysicalLookupJoin to StreamExecLookupJoin
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
lincoln-lil commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r947725826
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java:
##########
@@ -411,6 +426,55 @@ private Transformation<RowData> createSyncLookupJoinWithState(
return transform;
}
+ /**
+ * Update description of the transformation for easier debugging during job running. Will do:
+ *
+ * <p>1. correct async lookup mode if fallback to sync lookup
+ *
+ * <p>2. append async lookup parameters
+ *
+ * <p>3. append retry strategy
+ */
+ private String updateDescription(
Review Comment:
After confirmed that we can determine if async lookup will be enabled for runtime by avoid creating lookup function instance, we should do the change to make exec lookup join node immutable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20592: [FLINK-28987][table-planner] Updates description of lookup join transformation with async params and retry strategy for easier debugging
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20592:
URL: https://github.com/apache/flink/pull/20592#discussion_r948635867
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -108,6 +112,31 @@ abstract class CommonPhysicalLookupJoin(
"e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)")
}
+ val isAsyncEnabled = LookupJoinUtil.isAsyncLookup(
+ temporalTable,
+ allLookupKeys.keys.map(Int.box).toList.asJava,
+ lookupHint.orNull,
+ upsertMaterialize)
+
+ val retryOptions = Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull))
+
+ val inputChangelogMode = getInput match {
+ case streamPhysicalRel: StreamPhysicalRel =>
+ ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly())
+ case _ => ChangelogMode.insertOnly()
+ }
+
+ val tableConfig = unwrapTableConfig(this);
+
+ val asyncOptions = isAsyncEnabled match {
+ case true =>
+ Option.apply(
+ LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode))
+ case false =>
+ // do not create asyncOptions if async is not enabled
+ Option.empty[AsyncLookupOptions]
+ }
Review Comment:
mark them as lazy ?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private LookupJoinHintOptions() {}
public static final ConfigOption<Integer> ASYNC_CAPACITY =
key("capacity")
.intType()
- .defaultValue(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY
- .defaultValue())
+ .noDefaultValue()
Review Comment:
for most cases, the default value could meet users' requirement, so I prefer we can give the default value
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java:
##########
@@ -64,17 +62,14 @@ private 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()
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java:
##########
@@ -246,18 +254,23 @@ protected Transformation<RowData> createJoinTransformation(
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;
Review Comment:
does this check need for online environment ?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java:
##########
@@ -197,92 +363,177 @@ public static int[] getOrderedLookupKeys(Collection<Integer> allLookupKeys) {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
+ public static AsyncLookupOptions getMergedAsyncOptions(
Review Comment:
It's better we can add some UTs the verify the logica
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -108,6 +112,31 @@ abstract class CommonPhysicalLookupJoin(
"e.g., ON T1.id = T2.id && pythonUdf(T1.a, T2.b)")
}
+ val isAsyncEnabled = LookupJoinUtil.isAsyncLookup(
+ temporalTable,
+ allLookupKeys.keys.map(Int.box).toList.asJava,
+ lookupHint.orNull,
+ upsertMaterialize)
+
+ val retryOptions = Option.apply(LookupJoinUtil.RetryLookupOptions.fromJoinHint(lookupHint.orNull))
+
+ val inputChangelogMode = getInput match {
+ case streamPhysicalRel: StreamPhysicalRel =>
+ ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly())
+ case _ => ChangelogMode.insertOnly()
+ }
+
+ val tableConfig = unwrapTableConfig(this);
+
+ val asyncOptions = isAsyncEnabled match {
Review Comment:
nit: `if-else` looks simpler
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org