This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9bf67dddaa9 [FLINK-40844][table] Only EXPR$0 when a PTF's whole output
is empty
9bf67dddaa9 is described below
commit 9bf67dddaa92c6404df686e2a00d9c083228c350
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Wed Oct 7 15:39:57 2026 +0100
[FLINK-40844][table] Only EXPR$0 when a PTF's whole output is empty
Fixed by #29335
---
.../table/types/inference/SystemTypeInference.java | 15 ++++---
.../stream/ProcessTableFunctionSemanticTests.java | 2 +
.../stream/ProcessTableFunctionTestPrograms.java | 48 ++++++++++++++++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 16 ++++++++
.../plan/stream/sql/ProcessTableFunctionTest.java | 25 +++++++++++
5 files changed, 98 insertions(+), 8 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
index a4eadc77752..057ee0cb42d 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
@@ -403,16 +403,15 @@ public class SystemTypeInference {
}
private List<Field> deriveFunctionOutputFields(DataType
functionDataType) {
- final List<DataType> fieldTypes =
DataType.getFieldDataTypes(functionDataType);
- final List<String> fieldNames =
DataType.getFieldNames(functionDataType);
-
- if (fieldTypes.isEmpty()) {
- // Before the system type inference was introduced, SQL and
- // Table API chose a different default field name.
- // EXPR$0 is chosen for best-effort backwards compatibility for
- // SQL users.
+ final boolean isScalarOutput =
+
!LogicalTypeChecks.isCompositeType(functionDataType.getLogicalType());
+ if (isScalarOutput) {
+ // Scalar output has no field name of its own; EXPR$0 is kept
for
+ // backwards compatibility with the pre-system-type-inference
default.
return List.of(DataTypes.FIELD("EXPR$0", functionDataType));
}
+ final List<DataType> fieldTypes =
DataType.getFieldDataTypes(functionDataType);
+ final List<String> fieldNames =
DataType.getFieldNames(functionDataType);
return IntStream.range(0, fieldTypes.size())
.mapToObj(pos -> DataTypes.FIELD(fieldNames.get(pos),
fieldTypes.get(pos)))
.collect(Collectors.toList());
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index 728ae11cb12..b6ac9d59a01 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -66,6 +66,8 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY,
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY_TABLE_API,
ProcessTableFunctionTestPrograms.PROCESS_ATOMIC_WRAPPING,
+ ProcessTableFunctionTestPrograms.PROCESS_EMPTY_OUTPUT,
+
ProcessTableFunctionTestPrograms.PROCESS_EMPTY_OUTPUT_ROWTIME_ONLY,
ProcessTableFunctionTestPrograms.PROCESS_CONTEXT,
ProcessTableFunctionTestPrograms.PROCESS_POJO_STATE,
ProcessTableFunctionTestPrograms.PROCESS_DEFAULT_POJO_STATE,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index e846ef52ac6..28f87801a01 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -30,6 +30,8 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EagerAndValueViewStateTimeFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyOutputFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyOutputRowSemanticFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ImplicitCastingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalDayArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalYearArgFunction;
@@ -794,6 +796,52 @@ public class ProcessTableFunctionTestPrograms {
.runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name)")
.build();
+ public static final TableTestProgram PROCESS_EMPTY_OUTPUT =
+ TableTestProgram.of(
+ "process-empty-output",
+ "empty function output with pass-through and
rowtime columns")
+ .setupTemporarySystemFunction("f",
EmptyOutputFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("`name` STRING", "`rowtime`
TIMESTAMP_LTZ(3)")
+ .consumedValues(
+ "+I[Bob, 1970-01-01T00:00:00Z]",
+ "+I[Alice,
1970-01-01T00:00:00.001Z]",
+ "+I[Bob,
1970-01-01T00:00:00.002Z]",
+ "+I[Bob,
1970-01-01T00:00:00.003Z]",
+ "+I[Bob,
1970-01-01T00:00:00.004Z]",
+ "+I[Bob,
1970-01-01T00:00:00.005Z]",
+ "+I[Bob,
1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM "
+ + "f(r => TABLE t PARTITION BY name,
on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_EMPTY_OUTPUT_ROWTIME_ONLY =
+ TableTestProgram.of(
+ "process-empty-output-rowtime-only",
+ "empty function output with rowtime column but no
partition by")
+ .setupTemporarySystemFunction("f",
EmptyOutputRowSemanticFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("`rowtime` TIMESTAMP_LTZ(3)")
+ .consumedValues(
+ "+I[1970-01-01T00:00:00Z]",
+ "+I[1970-01-01T00:00:00.001Z]",
+ "+I[1970-01-01T00:00:00.002Z]",
+ "+I[1970-01-01T00:00:00.003Z]",
+ "+I[1970-01-01T00:00:00.004Z]",
+ "+I[1970-01-01T00:00:00.005Z]",
+ "+I[1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM "
+ + "f(r => TABLE t, on_time =>
DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram PROCESS_CONTEXT =
TableTestProgram.of("process-context", "outputs values from
function context")
.setupTemporarySystemFunction("f", ContextFunction.class)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index 7808afc4d58..127ee8fb424 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -203,6 +203,22 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ @DataTypeHint("ROW<>")
+ public static class EmptyOutputFunction extends ProcessTableFunction<Row> {
+ public void eval(@ArgumentHint(SET_SEMANTIC_TABLE) Row r) {
+ collect(Row.of());
+ }
+ }
+
+ /** Testing function with an empty output and no pass-through columns. */
+ @DataTypeHint("ROW<>")
+ public static class EmptyOutputRowSemanticFunction extends
ProcessTableFunction<Row> {
+ public void eval(@ArgumentHint(ROW_SEMANTIC_TABLE) Row r) {
+ collect(Row.of());
+ }
+ }
+
@DataTypeHint("ROW<`out` STRING>")
public abstract static class AppendProcessTableFunctionBase extends
ProcessTableFunction<Row> {
protected void collectObjects(Object... objects) {
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
index 7500bef9c98..cb90b207b8b 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
@@ -31,6 +31,8 @@ import org.apache.flink.table.functions.UserDefinedFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.AppendProcessTableFunctionBase;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyOutputFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyOutputRowSemanticFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalDayArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalYearArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.InvalidUpdatingSemanticsFunction;
@@ -78,6 +80,7 @@ import static
org.apache.flink.table.annotation.ArgumentTrait.SUPPORT_UPDATES;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.row;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the type inference and planning part of {@link
ProcessTableFunction}. */
@@ -199,6 +202,28 @@ class ProcessTableFunctionTest extends TableTestBase {
util.verifyRelPlan("SELECT * FROM f(uid => 'my-ptf')");
}
+ @Test
+ void testPartitionOnlyColumnOutput() {
+ util.addTemporarySystemFunction("f", EmptyOutputFunction.class);
+ assertThat(
+ util.tableEnv()
+ .sqlQuery("SELECT * FROM f(r => TABLE t
PARTITION BY name)")
+ .getResolvedSchema()
+ .getColumnNames())
+ .containsExactly("name");
+ }
+
+ @Test
+ void testZeroColumnOutput() {
+ util.addTemporarySystemFunction("f",
EmptyOutputRowSemanticFunction.class);
+ assertThat(
+ util.tableEnv()
+ .sqlQuery("SELECT * FROM f(r => TABLE t)")
+ .getResolvedSchema()
+ .getColumnNames())
+ .isEmpty();
+ }
+
@Test
void testIntervalDayArgs() {
util.addTemporarySystemFunction("f", IntervalDayArgFunction.class);
