diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
index 1020a39705c..02691968c27 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -55,8 +55,10 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlOperatorTable;
 import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlSyntax;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.parser.SqlParserPos;
@@ -66,9 +68,11 @@
 import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.util.SqlOperatorTables;
+import org.apache.calcite.sql.util.SqlShuttle;
 import org.apache.calcite.sql.validate.SqlConformance;
 import org.apache.calcite.sql.validate.SqlConformanceEnum;
 import org.apache.calcite.sql.validate.SqlDelegatingConformance;
+import org.apache.calcite.sql.validate.SqlNameMatcher;
 import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.sql.validate.SqlValidatorUtil;
 import org.apache.calcite.sql2rel.SqlToRelConverter;
@@ -214,18 +218,22 @@ private static RelNode sqlToRel(
         TransformSqlOperatorTable transformSqlOperatorTable = TransformSqlOperatorTable.instance();
         SqlOperatorTable udfOperatorTable = SqlOperatorTables.of(udfFunctions);
         SqlOperatorTable aiFunctionOperatorTable = AiFunctionSqlOperatorTable.create();
+        // Calcite looks up function candidates again when deriving a call's type. Rebind
+        // same-name calls and expose only UDF candidates to keep validation consistent with
+        // UDF-first code generation.
         SqlValidator validator =
                 SqlValidatorUtil.newValidator(
-                        SqlOperatorTables.chain(
-                                transformSqlOperatorTable,
+                        new UdfFirstSqlOperatorTable(
                                 udfOperatorTable,
-                                aiFunctionOperatorTable),
+                                SqlOperatorTables.chain(
+                                        transformSqlOperatorTable, aiFunctionOperatorTable)),
                         calciteCatalogReader,
                         factory,
                         SqlValidator.Config.DEFAULT
                                 .withIdentifierExpansion(true)
                                 .withConformance(SqlConformanceEnum.MYSQL_5));
-        SqlNode validateSqlNode = validator.validate(sqlNode);
+        SqlNode validateSqlNode =
+                validator.validate(resolveUserDefinedFunctions(sqlNode, udfFunctions));
         SqlToRelConverter sqlToRelConverter =
                 new SqlToRelConverter(
                         null,
@@ -240,6 +248,70 @@ private static RelNode sqlToRel(
         return relRoot.rel;
     }
 
+    private static SqlNode resolveUserDefinedFunctions(
+            SqlNode sqlNode, List<SqlFunction> udfFunctions) {
+        return sqlNode.accept(
+                new SqlShuttle() {
+                    @Override
+                    public SqlNode visit(SqlCall call) {
+                        SqlNode visited = super.visit(call);
+                        if (visited instanceof SqlBasicCall) {
+                            SqlBasicCall basicCall = (SqlBasicCall) visited;
+                            if (basicCall.getOperator().getSyntax().family != SqlSyntax.FUNCTION) {
+                                return visited;
+                            }
+                            udfFunctions.stream()
+                                    .filter(
+                                            udf ->
+                                                    udf.getName()
+                                                            .equalsIgnoreCase(
+                                                                    basicCall
+                                                                            .getOperator()
+                                                                            .getName()))
+                                    .findFirst()
+                                    .ifPresent(basicCall::setOperator);
+                        }
+                        return visited;
+                    }
+                });
+    }
+
+    private static final class UdfFirstSqlOperatorTable implements SqlOperatorTable {
+        private final SqlOperatorTable udfOperatorTable;
+        private final SqlOperatorTable fallbackOperatorTable;
+
+        private UdfFirstSqlOperatorTable(
+                SqlOperatorTable udfOperatorTable, SqlOperatorTable fallbackOperatorTable) {
+            this.udfOperatorTable = udfOperatorTable;
+            this.fallbackOperatorTable = fallbackOperatorTable;
+        }
+
+        @Override
+        public void lookupOperatorOverloads(
+                SqlIdentifier opName,
+                @Nullable SqlFunctionCategory category,
+                SqlSyntax syntax,
+                List<SqlOperator> operatorList,
+                SqlNameMatcher nameMatcher) {
+            List<SqlOperator> udfOperators = new ArrayList<>();
+            udfOperatorTable.lookupOperatorOverloads(
+                    opName, category, syntax, udfOperators, nameMatcher);
+            if (udfOperators.isEmpty()) {
+                fallbackOperatorTable.lookupOperatorOverloads(
+                        opName, category, syntax, operatorList, nameMatcher);
+            } else {
+                operatorList.addAll(udfOperators);
+            }
+        }
+
+        @Override
+        public List<SqlOperator> getOperatorList() {
+            List<SqlOperator> operators = new ArrayList<>(udfOperatorTable.getOperatorList());
+            operators.addAll(fallbackOperatorTable.getOperatorList());
+            return operators;
+        }
+    }
+
     public static SqlSelect parseSelect(String statement) {
         SqlNode sqlNode;
         try {
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
index 8bfa4a03ed9..2e02b05a3b9 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
@@ -488,7 +488,7 @@ void testUdfTakesPrecedenceOverBuiltInFunction() throws Exception {
                 PostTransformOperator.newBuilder()
                         .addTransform(
                                 CUSTOMERS_TABLEID.identifier(),
-                                "*, CAST(IFNULL(1, 0) AS VARCHAR) AS udf_ifnull, "
+                                "*, IFNULL(1, 0) AS udf_ifnull, "
                                         + "TRY_CAST(col1) AS udf_try_cast, "
                                         + "NULLIF('%s', col1) AS udf_nullif",
                                 null)
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 50fcc1e4556..06db8f9f472 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -1230,6 +1230,28 @@ void testUdfTakesPrecedenceOverBuiltInFunction() {
                 udfDescriptors);
     }
 
+    @Test
+    void testUdfDoesNotReplaceStructuralOperator() {
+        List<UserDefinedFunctionDescriptor> udfDescriptors =
+                Collections.singletonList(
+                        new UserDefinedFunctionDescriptor(
+                                "as",
+                                "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass"));
+
+        List<ProjectionColumn> projectionColumns =
+                TransformParser.generateProjectionColumns(
+                        "id AS alias",
+                        DUMMY_COLUMNS,
+                        udfDescriptors,
+                        new SupportedMetadataColumn[0]);
+
+        Assertions.assertThat(projectionColumns).hasSize(1);
+        ProjectionColumn projectionColumn = projectionColumns.get(0);
+        Assertions.assertThat(projectionColumn.getColumnName()).isEqualTo("alias");
+        Assertions.assertThat(projectionColumn.getDataType()).isEqualTo(DataTypes.INT());
+        Assertions.assertThat(projectionColumn.getScriptExpression()).isEqualTo("$0");
+    }
+
     @Test
     public void testTranslateUdfFilterToJaninoExpressionWithColumnNameMap() {
         List<Column> columns =