Add integration test for SqlJob with inline UDF files

Tests that CREATE FUNCTION with CODE option produces a SqlJob
containing the files map. The inline Python code is extracted
from the CODE option and mapped to a filename derived from the
AS clause (e.g., 'my_udf.transform' -> 'my_udf.py').

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ryannedolan

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/TestSqlScripts.java

@@ -29,6 +29,11 @@ public void k8sDdlScriptUdfDemo() throws Exception {
     run("k8s-ddl-udf-demo.id");
   }
 
+  @Test
+  public void k8sDdlScriptUdfFiles() throws Exception {
+    run("k8s-ddl-udf-files.id");
+  }
+
   @Test
   public void k8sValidationScript() throws Exception {
     run("k8s-validation.id");

hoptimator-k8s/src/test/resources/k8s-ddl-udf-files.id

@@ -0,0 +1,47 @@
+!set outputformat mysql
+!use k8s
+
+# Register a Python UDF with inline CODE that should appear in SqlJob files
+create function my_py_udf as 'my_udf.transform' language python with (CODE='x = 1');
+(0 rows modified)
+
+!update
+
+# UDF validates against real data (opaque placeholder returns null)
+select my_py_udf("FIRST_NAME") as val from profile.members;
++-----+
+| VAL |
++-----+
+|     |
+|     |
+|     |
++-----+
+(3 rows)
+
+!ok
+
+# Pipeline generates SqlJob with files containing the inline CODE
+insert into ads.page_views select my_py_udf("FIRST_NAME") as page_urn, "MEMBER_URN" as member_urn from profile.members;
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: SqlJob
+metadata:
+  name: ads-database-pageviews
+spec:
+  dialect: Flink
+  executionMode: Streaming
+  sql:
+  - CREATE FUNCTION IF NOT EXISTS `MY_PY_UDF` AS 'my_udf.transform' LANGUAGE PYTHON
+  - CREATE DATABASE IF NOT EXISTS `PROFILE` WITH ()
+  - CREATE TABLE IF NOT EXISTS `PROFILE`.`MEMBERS` (`FIRST_NAME` VARCHAR, `LAST_NAME` VARCHAR, `MEMBER_URN` VARCHAR, `COMPANY_URN` VARCHAR) WITH ('connector'='datagen', 'number-of-rows'='10')
+  - CREATE DATABASE IF NOT EXISTS `ADS` WITH ()
+  - CREATE TABLE IF NOT EXISTS `ADS`.`PAGE_VIEWS` (`PAGE_URN` VARCHAR, `MEMBER_URN` VARCHAR) WITH ('connector'='blackhole')
+  - INSERT INTO `ADS`.`PAGE_VIEWS` (`PAGE_URN`, `MEMBER_URN`) SELECT `MY_PY_UDF`(`FIRST_NAME`) AS `PAGE_URN`, `MEMBER_URN` FROM `PROFILE`.`MEMBERS`
+  files:
+    {my_udf.py: x = 1}
+!specify PAGE_VIEWS
+
+# Clean up
+drop function my_py_udf;
+(0 rows modified)
+
+!update