[FLINK-39273][table-common] Fix argument name conflict when UDF method contains lambda expression

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java

@@ -961,17 +961,21 @@ private static class ParameterExtractor extends ClassVisitor {
 
         private static final int OPCODE = Opcodes.ASM9;
 
+        private final String methodName;
+
         private final String methodDescriptor;
 
         private final Map<Integer, String> parameterNamesWithIndex = new TreeMap<>();
 
         ParameterExtractor(Constructor<?> constructor) {
             super(OPCODE);
+            methodName = "<init>";
             methodDescriptor = getConstructorDescriptor(constructor);
         }
 
         ParameterExtractor(Method method) {
             super(OPCODE);
+            methodName = method.getName();
             methodDescriptor = getMethodDescriptor(method);
         }
 
@@ -986,7 +990,7 @@ List<String> getParameterNames() {
         @Override
         public MethodVisitor visitMethod(
                 int access, String name, String descriptor, String signature, String[] exceptions) {
-            if (descriptor.equals(methodDescriptor)) {
+            if (name.equals(methodName) && descriptor.equals(methodDescriptor)) {
                 return new MethodVisitor(OPCODE) {
                     @Override
                     public void visitLocalVariable(

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/ExtractionUtilsTest.java

@@ -30,6 +30,7 @@
 import java.lang.reflect.Type;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
@@ -241,4 +242,26 @@ public void method(
                 List<CompletableFuture<Long>> listOfGenericFuture,
                 Long[] array) {}
     }
+
+    /**
+     * Verifies that {@link ExtractionUtils#extractExecutableNames} returns correct parameter names
+     * when a method contains a lambda that captures its own parameters, producing a synthetic
+     * method with the same bytecode descriptor.
+     */
+    @Test
+    void testExtractExecutableNamesWithLambdaCapture() {
+        Method method = ExtractionUtils.collectMethods(LambdaCaptureClass.class, "eval").get(0);
+        assertThat(ExtractionUtils.extractExecutableNames(method))
+                .isEqualTo(ImmutableList.of("id", "field"));
+    }
+
+    /** A single-method class where the lambda captures all parameters of the enclosing method. */
+    public static class LambdaCaptureClass {
+
+        @SuppressWarnings("unused")
+        public String eval(Long id, String field) {
+            Supplier<String> supplier = () -> String.valueOf(id) + field;
+            return supplier.get();
+        }
+    }
 }

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java

@@ -230,6 +230,17 @@ private static Stream<TestSpec> functionSpecs() {
                                 TypeStrategies.explicit(
                                         DataTypes.BIGINT().notNull().bridgedTo(long.class))),
                 // ---
+                // test eval method with lambda capture whose synthetic method shares the
+                // same bytecode descriptor, previously causing "Argument name conflict"
+                TestSpec.forScalarFunction(
+                                "ScalarFunctionWithLambdaCapture",
+                                ScalarFunctionWithLambdaCapture.class)
+                        .expectStaticArgument(
+                                StaticArgument.scalar("id", DataTypes.BIGINT(), false))
+                        .expectStaticArgument(
+                                StaticArgument.scalar("field", DataTypes.STRING(), false))
+                        .expectOutput(TypeStrategies.explicit(DataTypes.STRING())),
+                // ---
                 // test overloaded arguments extraction async
                 TestSpec.forAsyncScalarFunction(OverloadedFunctionAsync.class)
                         .expectOutputMapping(
@@ -1544,6 +1555,13 @@ public long eval(String s) {
         }
     }
 
+    private static class ScalarFunctionWithLambdaCapture extends ScalarFunction {
+        public String eval(Long id, String field) {
+            Supplier<String> supplier = () -> String.valueOf(id) + field;
+            return supplier.get();
+        }
+    }
+
     private static class VarArgFunction extends ScalarFunction {
         public String eval(int i, int... more) {
             return null;