(flink) branch master updated: [FLINK-40012][table] Correctly inject Context into scalar-only PTFs in ProcessTableFunctionTestHarness (#28579)

[fhueske]

This is an automated email from the ASF dual-hosted git repository.

fhueske 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 ce4cec57bc6 [FLINK-40012][table] Correctly inject Context into 
scalar-only PTFs in ProcessTableFunctionTestHarness (#28579)
ce4cec57bc6 is described below

commit ce4cec57bc654433a8a23f116ca6fc7f0bf0ec15
Author: Mika Naylor <mnay...@confluent.io>
AuthorDate: Tue Jun 30 19:01:26 2026 +0200

    [FLINK-40012][table] Correctly inject Context into scalar-only PTFs in 
ProcessTableFunctionTestHarness (#28579)
---
 .../table/runtime/functions/InvocationContext.java |  4 +++
 .../functions/ProcessTableFunctionTestHarness.java |  5 +++-
 .../ProcessTableFunctionTestHarnessTest.java       | 33 ++++++++++++++++++++++
 3 files changed, 41 insertions(+), 1 deletion(-)

diff --git 
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/InvocationContext.java
 
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/InvocationContext.java
index 0ff59b00d87..4c1acd3a5b9 100644
--- 
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/InvocationContext.java
+++ 
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/InvocationContext.java
@@ -51,6 +51,10 @@ class InvocationContext {
         return new InvocationContext(timer.partitionKey, null, null, timer);
     }
 
+    static InvocationContext forScalarOnlyEval() {
+        return new InvocationContext(Row.of(), null, null, null);
+    }
+
     boolean isTimerInvocation() {
         return firingTimer != null;
     }
diff --git 
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
 
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
index c5baddc1fcd..fa0297bd488 100644
--- 
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
+++ 
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
@@ -287,10 +287,13 @@ public class ProcessTableFunctionTestHarness<OUT> 
implements AutoCloseable {
         Object[] args = arguments.stream().map(arg -> ((ScalarArgumentInfo) 
arg).value).toArray();
 
         try {
-            eval.method.invoke(function, args);
+            currentInvocation = InvocationContext.forScalarOnlyEval();
+            eval.invoke(function, new TestContext(new HashMap<>()), args);
         } catch (InvocationTargetException e) {
             handleEvalInvocationException(
                     "Exception occurred during scalar-only PTF eval() 
invocation.\n", args, e);
+        } finally {
+            currentInvocation = null;
         }
     }
 
diff --git 
a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
 
b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
index b2ab499ca2c..f774a8fc422 100644
--- 
a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
+++ 
b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
@@ -328,6 +328,21 @@ class ProcessTableFunctionTestHarnessTest {
         }
     }
 
+    /**
+     * Scalar-only PTF whose eval() takes a {@link 
ProcessTableFunction.Context}. With no table
+     * arguments, we expect there to be no time/watermark, so both resolve to 
null.
+     */
+    @DataTypeHint("ROW<ts INT, watermark INT, value INT>")
+    public static class ScalarOnlyContextPTF extends ProcessTableFunction<Row> 
{
+        public void eval(
+                Context ctx,
+                @ArgumentHint(ArgumentTrait.SCALAR) Integer a,
+                @ArgumentHint(ArgumentTrait.SCALAR) Integer b) {
+            final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+            collect(Row.of(timeCtx.time(), timeCtx.currentWatermark(), a + b));
+        }
+    }
+
     /** PTF with simple value state - counts rows per partition. */
     @DataTypeHint("ROW<count BIGINT>")
     public static class PTFWithValueState extends ProcessTableFunction<Row> {
@@ -565,6 +580,24 @@ class ProcessTableFunctionTestHarnessTest {
         }
     }
 
+    @Test
+    void testScalarOnlyPTFInjectsContext() throws Exception {
+        try (ProcessTableFunctionTestHarness<Row> harness =
+                
ProcessTableFunctionTestHarness.ofClass(ScalarOnlyContextPTF.class)
+                        .withScalarArgument("a", 10)
+                        .withScalarArgument("b", 20)
+                        .build()) {
+
+            harness.process();
+
+            List<Row> output = harness.getOutput();
+            assertThat(output).hasSize(1);
+            assertThat(output.get(0).getField("ts")).isNull();
+            assertThat(output.get(0).getField("watermark")).isNull();
+            assertThat(output.get(0).getField("value")).isEqualTo(30);
+        }
+    }
+
     @Test
     void testScalarOnlyPTFWithWrongArgumentTypes() {
         Exception exception =