(flink) branch master updated: [FLINK-39994][table] Allow the use of primitive scalar arguments in ProcessTableFunctionTestHarness (#28545)

[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 b7d0869ea26 [FLINK-39994][table] Allow the use of primitive scalar 
arguments in ProcessTableFunctionTestHarness (#28545)
b7d0869ea26 is described below

commit b7d0869ea26e5efc7a9362cb7e019ec85ab89d1b
Author: Mika Naylor <mnay...@confluent.io>
AuthorDate: Fri Jun 26 08:42:13 2026 +0200

    [FLINK-39994][table] Allow the use of primitive scalar arguments in 
ProcessTableFunctionTestHarness (#28545)
---
 .../functions/ProcessTableFunctionTestHarness.java |  5 ++++-
 .../ProcessTableFunctionTestHarnessTest.java       | 24 ++++++++++++++++++++++
 2 files changed, 28 insertions(+), 1 deletion(-)

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 5a6eecb50f8..fc4e50618e1 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
@@ -47,6 +47,8 @@ import org.apache.flink.types.Row;
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Collector;
 
+import org.apache.commons.lang3.ClassUtils;
+
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Parameter;
@@ -871,7 +873,8 @@ public class ProcessTableFunctionTestHarness<OUT> 
implements AutoCloseable {
 
                 if (arg instanceof ScalarArgumentInfo) {
                     Object value = ((ScalarArgumentInfo) arg).value;
-                    if (value != null && 
!paramType.isAssignableFrom(value.getClass())) {
+                    Class<?> expectedType = 
ClassUtils.primitiveToWrapper(paramType);
+                    if (value != null && 
!expectedType.isAssignableFrom(value.getClass())) {
                         throw new IllegalStateException(
                                 String.format(
                                         "Type mismatch for scalar argument 
'%s' at position %d: "
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 88e1c8b32d8..fe0fac2390a 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
@@ -299,6 +299,14 @@ class ProcessTableFunctionTestHarnessTest {
         }
     }
 
+    /** PTF with primitive scalar arguments. */
+    @DataTypeHint("ROW<sum INT>")
+    public static class PrimitiveScalarPTF extends ProcessTableFunction<Row> {
+        public void eval(int a, int b) {
+            collect(Row.of(a + b));
+        }
+    }
+
     /** PTF with Context parameter - should be rejected by test harness. */
     @DataTypeHint("ROW<value INT>")
     public static class PTFWithContext extends ProcessTableFunction<Row> {
@@ -528,6 +536,22 @@ class ProcessTableFunctionTestHarnessTest {
         }
     }
 
+    @Test
+    void testPrimitiveScalarArguments() throws Exception {
+        try (ProcessTableFunctionTestHarness<Row> harness =
+                
ProcessTableFunctionTestHarness.ofClass(PrimitiveScalarPTF.class)
+                        .withScalarArgument("a", 10)
+                        .withScalarArgument("b", 20)
+                        .build()) {
+
+            harness.process();
+
+            List<Row> output = harness.getOutput();
+            assertThat(output).hasSize(1);
+            assertThat(output.get(0).getField("sum")).isEqualTo(30);
+        }
+    }
+
     @Test
     void testScalarOnlyPTFWithWrongArgumentTypes() {
         Exception exception =