[SPARK-53869] Support multiple files in pyFiles field

### What changes were proposed in this pull request?

This PR aims to support multiple files in `SparkApplication`'s `pyFiles` field.

### Why are the changes needed?

Currently, `pyFiles` is mapped to the main resource directly because it assumes a single Python file.

https://github.com/apache/spark-kubernetes-operator/blob/75515c752086853b1676cff39c7fc84b99163dc0/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java#L143

However, it's supposed to be a comma-separated string. If users provide multiple files, t causes a failure like the following.

```
python3: can't open file '/opt/spark/examples/src/main/python/pi.py,local:///opt/spark/examples/src/main/python/sort.py': [Errno 2] │
```

This PR proposes a mitigation to handle the first file of `pyFiles` as the primary resource and the rest of files as the real `pyFiles`. Note that the previous logic works without any change and new logic is going to be applied only when `mainClass` is `org.apache.spark.deploy.PythonRunner` specified additionally.

**BEFORE**

```yaml
spec:
  pyFiles: "local:///opt/spark/examples/src/main/python/pi.py"
```

**AFTER**

```yaml
spec:
  mainClass: "org.apache.spark.deploy.PythonRunner"
  pyFiles: "local:///opt/spark/examples/src/main/python/pi.py,local:///opt/spark/examples/src/main/python/lib.py"
```

### Does this PR introduce _any_ user-facing change?

No behavior change because new logic works only when `mainClass` is `org.apache.spark.deploy.PythonRunner`.

### How was this patch tested?

Pass the CIs with newly added test case.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#382 from dongjoon-hyun/SPARK-53869.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java

@@ -139,6 +139,12 @@ protected SparkAppDriverConf buildDriverConf(
     if (StringUtils.isNotEmpty(applicationSpec.getJars())) {
       primaryResource = new JavaMainAppResource(Option.apply(applicationSpec.getJars()));
       effectiveSparkConf.setIfMissing("spark.jars", applicationSpec.getJars());
+    } else if ("org.apache.spark.deploy.PythonRunner".equals(applicationSpec.getMainClass())) {
+      String[] files = applicationSpec.getPyFiles().split(",", 2);
+      primaryResource = new PythonMainAppResource(files[0]);
+      if (files.length > 1 && !files[1].isBlank()) {
+        effectiveSparkConf.setIfMissing("spark.submit.pyFiles", files[1]);
+      }
     } else if (StringUtils.isNotEmpty(applicationSpec.getPyFiles())) {
       primaryResource = new PythonMainAppResource(applicationSpec.getPyFiles());
       effectiveSparkConf.setIfMissing("spark.submit.pyFiles", applicationSpec.getPyFiles());

spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorkerTest.java

@@ -129,6 +129,34 @@ void buildDriverConfForPythonApp() {
     }
   }
 
+  @Test
+  void handlePyFiles() {
+    Map<SparkAppDriverConf, List<Object>> constructorArgs = new HashMap<>();
+    try (MockedConstruction<SparkAppDriverConf> mocked =
+        mockConstruction(
+            SparkAppDriverConf.class,
+            (mock, context) -> constructorArgs.put(mock, new ArrayList<>(context.arguments())))) {
+      SparkApplication mockApp = mock(SparkApplication.class);
+      ApplicationSpec mockSpec = mock(ApplicationSpec.class);
+      ObjectMeta appMeta = new ObjectMetaBuilder().withName("app1").withNamespace("ns1").build();
+      when(mockApp.getSpec()).thenReturn(mockSpec);
+      when(mockApp.getMetadata()).thenReturn(appMeta);
+      when(mockSpec.getMainClass()).thenReturn("org.apache.spark.deploy.PythonRunner");
+      when(mockSpec.getPyFiles()).thenReturn("main.py,lib.py");
+
+      SparkAppSubmissionWorker submissionWorker = new SparkAppSubmissionWorker();
+      SparkAppDriverConf conf = submissionWorker.buildDriverConf(mockApp, Collections.emptyMap());
+      assertEquals(6, constructorArgs.get(conf).size());
+      assertEquals(
+          "lib.py", ((SparkConf) constructorArgs.get(conf).get(0)).get("spark.submit.pyFiles"));
+
+      // validate main resources
+      assertInstanceOf(PythonMainAppResource.class, constructorArgs.get(conf).get(2));
+      PythonMainAppResource mainResource = (PythonMainAppResource) constructorArgs.get(conf).get(2);
+      assertEquals("main.py", mainResource.primaryResource());
+    }
+  }
+
   @Test
   void buildDriverConfForRApp() {
     Map<SparkAppDriverConf, List<Object>> constructorArgs = new HashMap<>();