fix: Return 400 for CSV export with array columns

When exporting a ViewDefinition result to CSV via $viewdefinition-export,
if the view contains array columns, Spark throws an AnalysisException with
UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE. This previously surfaced as a 500
Internal Server Error. Now it returns a 400 Bad Request with a clear error
message indicating the unsupported column.

Author: johngrimes

server/src/main/java/au/csiro/pathling/operations/view/ViewDefinitionExportExecutor.java

@@ -40,6 +40,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.AnalysisException;
 import org.apache.spark.sql.Column;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -281,7 +282,12 @@ private List<String> writeNdjson(
         FileSystemPersistence.renamePartitionedFiles(sparkSession, outputPath, outputPath, "json"));
   }
 
-  /** Writes the result as CSV files. */
+  /**
+   * Writes the result as CSV files.
+   *
+   * @throws InvalidRequestException if the dataset contains data types that are not supported by
+   *     the CSV format
+   */
   @Nonnull
   private List<String> writeCsv(
       @Nonnull final Dataset<Row> result,
@@ -290,7 +296,22 @@ private List<String> writeCsv(
       final boolean includeHeader,
       @Nonnull final Path jobDirPath) {
 
-    result.write().mode(SaveMode.Overwrite).option("header", includeHeader).csv(outputPath);
+    try {
+      result.write().mode(SaveMode.Overwrite).option("header", includeHeader).csv(outputPath);
+    } catch (final Exception e) {
+      // Spark throws AnalysisException when it encounters unsupported data types for a datasource.
+      // We convert this to an InvalidRequestException to return a 400 status code.
+      if (e instanceof final AnalysisException ae
+          && "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE".equals(ae.getErrorClass())) {
+        throw new InvalidRequestException(
+            "CSV export failed for view '%s': %s".formatted(viewName, e.getMessage()));
+      }
+      // Re-throw unexpected exceptions as runtime exceptions.
+      if (e instanceof RuntimeException re) {
+        throw re;
+      }
+      throw new RuntimeException(e);
+    }
 
     // Rename partitioned files to follow naming convention.
     return new ArrayList<>(

server/src/test/java/au/csiro/pathling/operations/view/ViewDefinitionExportExecutorTest.java

@@ -29,6 +29,7 @@
 import au.csiro.pathling.test.SpringBootUnitTest;
 import au.csiro.pathling.util.CustomObjectDataSource;
 import au.csiro.pathling.util.FhirServerTestConfiguration;
+import au.csiro.pathling.views.Column;
 import au.csiro.pathling.views.FhirView;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException;
@@ -148,6 +149,41 @@ void csvOutputCreatesFiles() {
     assertThat(outputs.get(0).fileUrls().get(0)).contains(".csv");
   }
 
+  @Test
+  void csvExportWithArrayColumnThrowsInvalidRequestException() {
+    // CSV format does not support array columns. When a view contains a collection column, CSV
+    // export should fail with a 400 error rather than a 500 internal server error.
+    final Patient patient = createPatient("test-1", "Smith");
+    patient.addName().addGiven("John").addGiven("James");
+    executor = createExecutor(patient);
+
+    // Create a view that includes a collection column (given names as an array).
+    final FhirView view =
+        FhirView.ofResource("Patient")
+            .select(
+                FhirView.columns(
+                    FhirView.column("id", "id"),
+                    Column.collection("given_names", "name.first().given")))
+            .build();
+
+    final ViewInput viewInput = new ViewInput("patients", view);
+    final ViewDefinitionExportRequest request =
+        new ViewDefinitionExportRequest(
+            "http://example.org/$viewdefinition-export",
+            "http://example.org/fhir",
+            List.of(viewInput),
+            null,
+            ViewExportFormat.CSV,
+            true,
+            Collections.emptySet(),
+            null);
+
+    assertThatThrownBy(() -> executor.execute(request, UUID.randomUUID().toString()))
+        .isInstanceOf(InvalidRequestException.class)
+        .hasMessageContaining("CSV export failed")
+        .hasMessageContaining("given_names");
+  }
+
   // -------------------------------------------------------------------------
   // Parquet output tests
   // -------------------------------------------------------------------------