fix(operator): guard against NoSuchElementException in ParallelCSVScanSourceOpDesc.getPhysicalOp (#4375)

### What changes were proposed in this PR?

`ParallelCSVScanSourceOpDesc.getPhysicalOp` called `customDelimiter.get`
without
first checking whether the `Option` is defined. When `customDelimiter`
is `None`
(the field's default), this throws a `NoSuchElementException` before the
fallback
comma delimiter can be applied.

**Before:**
```scala
if (customDelimiter.get.isEmpty) {   // throws NoSuchElementException when None
  customDelimiter = Option(",")
}
```

**After:**
```scala
if (customDelimiter.isEmpty || customDelimiter.get.isEmpty) {
  customDelimiter = Option(",")
}
```

This brings the parallel variant in line with `CSVScanSourceOpDesc`,
which has
always used the correct two-part guard.

### Any related issues, documentation, discussions?

Closes #4374

### How was this PR tested?

Two new test cases were added to `CSVScanSourceOpDescSpec`:

1. `"use comma as the default delimiter when customDelimiter is not set
for parallel CSV"` — verifies that `getPhysicalOp` does not throw when
`customDelimiter` is `None` and that the default `,` is applied.
2. `"use comma as the default delimiter when customDelimiter is empty
string for parallel CSV"` — same verification for `Some("")`.

The existing parallel-CSV schema-inference tests continue to pass
unchanged.

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

No.

---------

Signed-off-by: Asish Kumar <officialasishkumar@gmail.com>

Author: officialasishkumar

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala

@@ -55,7 +55,7 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
       executionId: ExecutionIdentity
   ): PhysicalOp = {
     // fill in default values
-    if (customDelimiter.isEmpty || customDelimiter.get.isEmpty) {
+    if (customDelimiter.forall(_.isEmpty)) {
       customDelimiter = Option(",")
     }
 

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala

@@ -56,7 +56,7 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
       executionId: ExecutionIdentity
   ): PhysicalOp = {
     // fill in default values
-    if (customDelimiter.get.isEmpty) {
+    if (customDelimiter.forall(_.isEmpty)) {
       customDelimiter = Option(",")
     }
 

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala

@@ -128,4 +128,36 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     )
   }
 
+  it should "use comma as the default delimiter when customDelimiter is not set for parallel CSV" in {
+    parallelCsvScanSourceOpDesc.customDelimiter = None
+
+    parallelCsvScanSourceOpDesc.getPhysicalOp(DEFAULT_WORKFLOW_ID, DEFAULT_EXECUTION_ID)
+
+    assert(parallelCsvScanSourceOpDesc.customDelimiter.contains(","))
+  }
+
+  it should "use comma as the default delimiter when customDelimiter is empty string for parallel CSV" in {
+    parallelCsvScanSourceOpDesc.customDelimiter = Some("")
+
+    parallelCsvScanSourceOpDesc.getPhysicalOp(DEFAULT_WORKFLOW_ID, DEFAULT_EXECUTION_ID)
+
+    assert(parallelCsvScanSourceOpDesc.customDelimiter.contains(","))
+  }
+
+  it should "use comma as the default delimiter when customDelimiter is not set for CSV" in {
+    csvScanSourceOpDesc.customDelimiter = None
+
+    csvScanSourceOpDesc.getPhysicalOp(DEFAULT_WORKFLOW_ID, DEFAULT_EXECUTION_ID)
+
+    assert(csvScanSourceOpDesc.customDelimiter.contains(","))
+  }
+
+  it should "use comma as the default delimiter when customDelimiter is empty string for CSV" in {
+    csvScanSourceOpDesc.customDelimiter = Some("")
+
+    csvScanSourceOpDesc.getPhysicalOp(DEFAULT_WORKFLOW_ID, DEFAULT_EXECUTION_ID)
+
+    assert(csvScanSourceOpDesc.customDelimiter.contains(","))
+  }
+
 }