You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "twalthr (via GitHub)" <gi...@apache.org> on 2025/01/24 10:22:18 UTC
[PR] [FLINK-37180][table] Support running stateless PTFs [flink]
twalthr opened a new pull request, #26076:
URL: https://github.com/apache/flink/pull/26076
## What is the purpose of the change
Will follow...
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1928574617
##########
flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java:
##########
@@ -177,6 +177,14 @@ public List<FunctionTestStep> getSetupFunctionTestSteps() {
.collect(Collectors.toList());
}
+ /** Convenience method to avoid casting. It assumes that the order of steps is not important. */
Review Comment:
I usually faced adj + noun rather than noun + noun... not sure how correct it is...
```suggestion
/** Convenient method to avoid casting. It assumes that the order of steps is not important. */
```
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944450426
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java:
##########
@@ -135,6 +142,25 @@ private static void checkReservedArgs(List<StaticArgument> staticArgs) {
}
}
+ private static void checkMultipleTableArgs(List<StaticArgument> staticArgs) {
+ if (staticArgs.stream().filter(arg -> arg.is(StaticArgumentTrait.TABLE)).count() > 1) {
Review Comment:
given that in most cases this is just 2 or 3, limiting make the code just more complex
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944571585
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java:
##########
@@ -163,40 +167,31 @@ protected RelDataType deriveRowType() {
return rowType;
}
- public List<StaticArgument> getProvidedInputArgs() {
- final RexCall call = (RexCall) scan.getCall();
- final List<RexNode> operands = call.getOperands();
- final BridgingSqlFunction.WithTableFunction function =
- (BridgingSqlFunction.WithTableFunction) call.getOperator();
- final List<StaticArgument> declaredArgs =
- function.getTypeInference()
- .getStaticArguments()
- .orElseThrow(IllegalStateException::new);
- // This logic filters out optional tables for which an input is missing. It returns tables
- // in the same order as provided inputs of this RelNode.
- return Ord.zip(declaredArgs).stream()
- .filter(arg -> arg.e.is(StaticArgumentTrait.TABLE))
- .filter(arg -> operands.get(arg.i) instanceof RexTableArgCall)
- .map(arg -> arg.e)
- .collect(Collectors.toList());
- }
-
/**
- * An important part of {@link ProcessTableFunction} is the mandatory unique identifier. Even if
- * the PTF has no state entries, state or timers might be added later. So a PTF should serve as
- * an identifiable black box for the optimizer. UIDs ensure that.
+ * An important part of {@link ProcessTableFunction} is the mandatory unique identifier for PTFs
Review Comment:
Yes, this is similar to non-keyed datastream operators. We can't check everything during validation because state and time is available after specialization. Esp. for built-in functions. Specialization happens in StreamExec node. But since this is introducing stateless PTFs. Let's postpone this discussion to stateful ones.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1928588289
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java:
##########
@@ -135,6 +142,25 @@ private static void checkReservedArgs(List<StaticArgument> staticArgs) {
}
}
+ private static void checkMultipleTableArgs(List<StaticArgument> staticArgs) {
+ if (staticArgs.stream().filter(arg -> arg.is(StaticArgumentTrait.TABLE)).count() > 1) {
Review Comment:
```suggestion
if (staticArgs.stream().filter(arg -> arg.is(StaticArgumentTrait.TABLE)).limit(2).count() > 1) {
```
nit: would it make sense to add limit like?
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26076:
URL: https://github.com/apache/flink/pull/26076#issuecomment-2612248449
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "799d81532fc4c2f752e97b8aeef09b5007f96f5d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "799d81532fc4c2f752e97b8aeef09b5007f96f5d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 799d81532fc4c2f752e97b8aeef09b5007f96f5d 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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944576029
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.stream;
+
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.AtomicTypeWrappingFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoArgsFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoCreatingFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ScalarArgsFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowPassThroughFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetOptionalPartitionFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetPassThroughFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetUpdatingArgFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TestProcessTableFunctionBase;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedTableAsRowFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedTableAsSetFunction;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+/** {@link TableTestProgram} definitions for testing {@link StreamExecProcessTableFunction}. */
+public class ProcessTableFunctionTestPrograms {
+
+ private static final String BASIC_VALUES =
+ "CREATE VIEW t AS SELECT * FROM (VALUES ('Bob', 12), ('Alice', 42)) AS T(name, score)";
+
+ private static final String UPDATING_VALUES =
+ "CREATE VIEW t AS SELECT name, COUNT(*) FROM "
+ + "(VALUES ('Bob', 12), ('Alice', 42), ('Bob', 14)) AS T(name, score) "
+ + "GROUP BY name";
+
+ /** Corresponds to {@link TestProcessTableFunctionBase}. */
+ private static final String BASE_SINK_SCHEMA = "`out` STRING, `count` INT";
+
+ /** Corresponds to {@link TestProcessTableFunctionBase}. */
+ private static final String KEYED_BASE_SINK_SCHEMA = "`name` STRING, `out` STRING, `count` INT";
+
+ /** Corresponds to {@link TestProcessTableFunctionBase}. */
+ private static final String PASS_THROUGH_BASE_SINK_SCHEMA =
+ "`name` STRING, `score` INT, `out` STRING, `count` INT";
+
+ public static final TableTestProgram PROCESS_SCALAR_ARGS =
+ TableTestProgram.of("process-scalar-args", "no table as input")
+ .setupTemporarySystemFunction("f", ScalarArgsFunction.class)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(BASE_SINK_SCHEMA)
+ .consumedValues("+I[{42, true}, 1]", "+I[{42, true}, 2]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(i => 42, b => CAST('TRUE' AS BOOLEAN))")
+ .build();
+
+ public static final TableTestProgram PROCESS_TABLE_AS_ROW =
+ TableTestProgram.of("process-row", "table with row semantics")
+ .setupTemporarySystemFunction("f", TableAsRowFunction.class)
+ .setupSql(BASIC_VALUES)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[{+I[Bob, 12], 1}, 1]",
+ "+I[{+I[Bob, 12], 1}, 2]",
+ "+I[{+I[Alice, 42], 1}, 1]",
+ "+I[{+I[Alice, 42], 1}, 2]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t, i => 1)")
+ .build();
+
+ public static final TableTestProgram PROCESS_TYPED_TABLE_AS_ROW =
+ TableTestProgram.of("process-typed-row", "typed table with row semantics")
+ .setupTemporarySystemFunction("f", TypedTableAsRowFunction.class)
+ .setupSql(BASIC_VALUES)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[{User(s='Bob', i=12), 1}, 1]",
+ "+I[{User(s='Bob', i=12), 1}, 2]",
+ "+I[{User(s='Alice', i=42), 1}, 1]",
+ "+I[{User(s='Alice', i=42), 1}, 2]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f(u => TABLE t, i => 1)")
+ .build();
+
+ public static final TableTestProgram PROCESS_TABLE_AS_SET =
+ TableTestProgram.of("process-set", "table with set semantics")
+ .setupTemporarySystemFunction("f", TableAsSetFunction.class)
+ .setupSql(BASIC_VALUES)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {+I[Bob, 12], 1}, 1]",
+ "+I[Bob, {+I[Bob, 12], 1}, 2]",
Review Comment:
I'm also annoyed by it nowadays. I will remove this.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944571585
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java:
##########
@@ -163,40 +167,31 @@ protected RelDataType deriveRowType() {
return rowType;
}
- public List<StaticArgument> getProvidedInputArgs() {
- final RexCall call = (RexCall) scan.getCall();
- final List<RexNode> operands = call.getOperands();
- final BridgingSqlFunction.WithTableFunction function =
- (BridgingSqlFunction.WithTableFunction) call.getOperator();
- final List<StaticArgument> declaredArgs =
- function.getTypeInference()
- .getStaticArguments()
- .orElseThrow(IllegalStateException::new);
- // This logic filters out optional tables for which an input is missing. It returns tables
- // in the same order as provided inputs of this RelNode.
- return Ord.zip(declaredArgs).stream()
- .filter(arg -> arg.e.is(StaticArgumentTrait.TABLE))
- .filter(arg -> operands.get(arg.i) instanceof RexTableArgCall)
- .map(arg -> arg.e)
- .collect(Collectors.toList());
- }
-
/**
- * An important part of {@link ProcessTableFunction} is the mandatory unique identifier. Even if
- * the PTF has no state entries, state or timers might be added later. So a PTF should serve as
- * an identifiable black box for the optimizer. UIDs ensure that.
+ * An important part of {@link ProcessTableFunction} is the mandatory unique identifier for PTFs
Review Comment:
Yes, this is similar to non-keyed datastream operators. We can't check everything during validation because state and time is available after specialization. Esp. for built-in functions. Specialization happens in StreamExec node, but since this is introducing stateless PTFs. Let's postpone this discussion to stateful ones.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944594056
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.RowKind;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Serializable representation of {@link TableSemantics} that will serve {@link
+ * ProcessTableFunction.Context#tableSemanticsFor(String)} during runtime.
+ */
+public class RuntimeTableSemantics implements TableSemantics, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String argName;
+ private final int inputIndex;
+ private final DataType dataType;
+ private final int[] partitionByColumns;
+ private final byte[] expectedChanges;
+ private final boolean passColumnsThrough;
+ private final boolean hasSetSemantics;
+
+ private transient ChangelogMode changelogMode;
+
+ public RuntimeTableSemantics(
+ String argName,
+ int inputIndex,
+ DataType dataType,
+ int[] partitionByColumns,
+ byte[] expectedChanges,
+ boolean passColumnsThrough,
+ boolean hasSetSemantics) {
+ this.argName = argName;
+ this.inputIndex = inputIndex;
+ this.dataType = dataType;
+ this.partitionByColumns = partitionByColumns;
+ this.expectedChanges = expectedChanges;
+ this.passColumnsThrough = passColumnsThrough;
+ this.hasSetSemantics = hasSetSemantics;
+ }
+
+ public String getArgName() {
+ return argName;
+ }
+
+ public int getInputIndex() {
+ return inputIndex;
+ }
+
+ public boolean passColumnsThrough() {
+ return passColumnsThrough;
+ }
+
+ public boolean hasSetSemantics() {
+ return hasSetSemantics;
+ }
+
+ @Override
+ public DataType dataType() {
+ return dataType;
+ }
+
+ @Override
+ public int[] partitionByColumns() {
+ return partitionByColumns;
+ }
+
+ @Override
+ public int[] orderByColumns() {
+ return new int[0];
+ }
+
+ @Override
+ public int timeColumn() {
+ return -1;
+ }
+
+ @Override
+ public List<String> coPartitionArgs() {
+ return List.of();
+ }
Review Comment:
We don't need to throw an exception here. The return value is correct. A time columns and co partition args are not present.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944592839
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.test.program.ConfigOptionTestStep;
+import org.apache.flink.table.test.program.FunctionTestStep;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.SqlTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.table.test.program.TableTestProgramRunner;
+import org.apache.flink.table.test.program.TestStep;
+import org.apache.flink.table.test.program.TestStep.TestKind;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Base for tests that do not fall into the {@link RestoreTestBase} category, but use the {@link
Review Comment:
Yes it was, but I just wanted to make it clearer when to use RestoreTestBase and when SemanticTestBase.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #26076:
URL: https://github.com/apache/flink/pull/26076
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1944557382
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/OperatorBindingCallContext.java:
##########
@@ -85,7 +107,7 @@ public boolean isArgumentLiteral(int pos) {
@Override
public boolean isArgumentNull(int pos) {
- return binding.isOperandNull(pos, false);
+ return binding.isOperandNull(pos, false) || isDefault(pos);
Review Comment:
Yes, default values will be passed as null into UDFs.
--
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
Re: [PR] [FLINK-37180][table] Support running stateless PTFs [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26076:
URL: https://github.com/apache/flink/pull/26076#discussion_r1928537953
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java:
##########
@@ -57,10 +56,15 @@
@Internal
public class SystemTypeInference {
- private static final List<StaticArgument> PROCESS_TABLE_FUNCTION_SYSTEM_ARGS =
+ public static final List<StaticArgument> PROCESS_TABLE_FUNCTION_SYSTEM_ARGS =
List.of(StaticArgument.scalar("uid", DataTypes.STRING(), true));
- /** Format of unique identifiers for {@link ProcessTableFunction}. */
+ /**
+ * Format of unique identifiers for {@link ProcessTableFunction}.
+ *
+ * <p>Leading digits are not allowed. This also prevents that a custom PTF uid can infer with
Review Comment:
```suggestion
* <p>Leading digits are not allowed. This also prevents that a custom PTF uid can interfere with
```
--
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