Fail when Delta protocol is needed but not supported

server/app/src/main/java/io/whitefox/api/deltasharing/DeltaMappers.java

@@ -11,6 +11,7 @@
 import io.whitefox.core.*;
 import io.whitefox.core.Schema;
 import io.whitefox.core.Share;
+import io.whitefox.core.services.capabilities.ClientCapabilities;
 import io.whitefox.core.services.capabilities.ResponseFormat;
 import java.util.*;
 import java.util.stream.Collectors;
@@ -29,7 +30,8 @@ public static io.whitefox.api.deltasharing.model.v1.generated.Schema schema2api(
         .share(schema.share());
   }
 
-  public static ReadTableRequest api2ReadTableRequest(QueryRequest request) {
+  public static ReadTableRequest api2ReadTableRequest(
+      QueryRequest request, ClientCapabilities clientCapabilities) {
     if (request.getStartingVersion() != null && request.getEndingVersion() != null) {
       throw new IllegalArgumentException("The startingVersion and endingVersion are not supported");
     } else if (request.getStartingVersion() != null) {
@@ -43,18 +45,21 @@ public static ReadTableRequest api2ReadTableRequest(QueryRequest request) {
           Optional.ofNullable(request.getPredicateHints()),
           Optional.ofNullable(request.getJsonPredicateHints()),
           Optional.ofNullable(request.getLimitHint()),
-          request.getVersion());
+          request.getVersion(),
+          clientCapabilities);
     } else if (request.getVersion() == null && request.getTimestamp() != null) {
       return new ReadTableRequest.ReadTableAsOfTimestamp(
           Optional.ofNullable(request.getPredicateHints()),
-          Optional.ofNullable(request.getJsonPredicateHints()),
           Optional.ofNullable(request.getLimitHint()),
-          CommonMappers.parseTimestamp(request.getTimestamp()));
+          Optional.ofNullable(request.getJsonPredicateHints()),
+          CommonMappers.parseTimestamp(request.getTimestamp()),
+          clientCapabilities);
     } else if (request.getVersion() == null && request.getTimestamp() == null) {
       return new ReadTableRequest.ReadTableCurrentVersion(
           Optional.ofNullable(request.getPredicateHints()),
           Optional.ofNullable(request.getJsonPredicateHints()),
-          Optional.ofNullable(request.getLimitHint()));
+          Optional.ofNullable(request.getLimitHint()),
+          clientCapabilities);
     } else {
       throw new IllegalArgumentException("Cannot specify both version and timestamp");
     }
@@ -83,11 +88,9 @@ private static ParquetMetadata metadata2Api(Metadata metadata) {
                 .numFiles(metadata.numFiles())
                 .build())
             .build();
-      case delta:
-        throw new IllegalArgumentException("Delta response format is not supported");
       default:
         throw new IllegalArgumentException(
-            String.format("%s response format is not supported", metadata.format()));
+            String.format("%s file format is not supported", metadata.format()));
     }
   }
 
@@ -110,10 +113,12 @@ private static ParquetFile file2Api(TableFile f) {
         .build();
   }
 
-  public static TableReferenceAndReadRequest api2TableReferenceAndReadRequest(
-      QueryRequest request, String share, String schema, String table) {
-    return new TableReferenceAndReadRequest(share, schema, table, api2ReadTableRequest(request));
-  }
+  //  public static TableReferenceAndReadRequest api2TableReferenceAndReadRequest(
+  //      QueryRequest request, String share, String schema, String table, ClientCapabilities
+  // clientCapabilities) {
+  //    return new TableReferenceAndReadRequest(share, schema, table, api2ReadTableRequest(request,
+  // clientCapabilities));
+  //  }
 
   public static io.whitefox.api.deltasharing.model.v1.generated.Table table2api(
       SharedTable sharedTable) {

server/app/src/main/java/io/whitefox/api/deltasharing/server/DeltaSharesApiImpl.java

@@ -90,12 +90,12 @@ public Response getTableMetadata(
                   deltaSharesService.getTableVersion(share, schema, table, startingTimestamp),
                   v -> Response.ok(
                           tableResponseSerializer.serialize(
-                              DeltaMappers.toTableResponseMetadata(m)),
+                              DeltaMappers.toTableResponseMetadata(m.metadata())),
                           ndjsonMediaType)
                       .header(DELTA_TABLE_VERSION_HEADER, String.valueOf(v))
                       .header(
                           DELTA_SHARE_CAPABILITIES_HEADER,
-                          DeltaMappers.toResponseFormatHeader(m.format()))
+                          DeltaMappers.toResponseFormatHeader(m.responseFormat()))
                       .build()));
         },
         exceptionToResponse);
@@ -196,12 +196,13 @@ public Response queryTable(
       String deltaSharingCapabilities) {
     return wrapExceptions(
         () -> {
+          var clientCapabilities =
+              clientCapabilitiesMapper.parseDeltaSharingCapabilities(deltaSharingCapabilities);
           var readResult = deltaSharesService.queryTable(
               share,
               schema,
               table,
-              DeltaMappers.api2ReadTableRequest(queryRequest),
-              clientCapabilitiesMapper.parseDeltaSharingCapabilities(deltaSharingCapabilities));
+              DeltaMappers.api2ReadTableRequest(queryRequest, clientCapabilities));
           var serializedReadResult =
               tableQueryResponseSerializer.serialize(DeltaMappers.readTableResult2api(readResult));
           return Response.ok(serializedReadResult, ndjsonMediaType)

server/app/src/test/java/io/whitefox/api/deltasharing/ClientCapabilitiesMapperTest.java

@@ -17,42 +17,42 @@ public class ClientCapabilitiesMapperTest implements DeltaHeaders {
   @Test
   void parseSimpleResponseFormatDelta() {
     Assertions.assertEquals(
-        ResponseFormat.delta,
-        mapper.parseDeltaSharingCapabilities(responseFormatDelta).responseFormat());
+        Set.of(ResponseFormat.delta, ResponseFormat.parquet),
+        mapper.parseDeltaSharingCapabilities(responseFormatDelta).responseFormats());
   }
 
   @Test
   void parseSimpleResponseFormatParquet() {
     Assertions.assertEquals(
-        ResponseFormat.parquet,
-        mapper.parseDeltaSharingCapabilities("responseformat=PaRquEt").responseFormat());
+        Set.of(ResponseFormat.parquet),
+        mapper.parseDeltaSharingCapabilities("responseformat=PaRquEt").responseFormats());
   }
 
   @Test
   void failToParseUnknownResponseFormatAndFail() {
     Assertions.assertThrows(
         UnknownResponseFormat.class,
-        () -> mapper.parseDeltaSharingCapabilities("responseformat=iceberg").responseFormat());
+        () -> mapper.parseDeltaSharingCapabilities("responseformat=iceberg").responseFormats());
   }
 
   @Test
   void failToParseUnknownResponseFormatAndReturnOthers() {
     Assertions.assertEquals(
-        ResponseFormat.delta,
+        Set.of(ResponseFormat.delta, ResponseFormat.parquet),
         mapper
             .parseDeltaSharingCapabilities("responseformat=iceberg,parquet,delta")
-            .responseFormat());
+            .responseFormats());
   }
 
   @Test
   void noCapabilitiesEqualsDefault() {
     Assertions.assertEquals(
-        ResponseFormat.parquet,
-        mapper.parseDeltaSharingCapabilities((String) null).responseFormat());
+        Set.of(ResponseFormat.parquet),
+        mapper.parseDeltaSharingCapabilities((String) null).responseFormats());
     Assertions.assertEquals(
         Set.of(), mapper.parseDeltaSharingCapabilities((String) null).readerFeatures());
     Assertions.assertEquals(
-        ResponseFormat.parquet, mapper.parseDeltaSharingCapabilities("").responseFormat());
+        Set.of(ResponseFormat.parquet), mapper.parseDeltaSharingCapabilities("").responseFormats());
     Assertions.assertEquals(Set.of(), mapper.parseDeltaSharingCapabilities("").readerFeatures());
   }
 
@@ -103,7 +103,8 @@ void kitchenSink() {
     var responseFormat = "responseformat=iceberg,parquet,delta";
     var capabilities = mapper.parseDeltaSharingCapabilities(
         String.format("%s;%s", readerFeatures, responseFormat));
-    Assertions.assertEquals(ResponseFormat.delta, capabilities.responseFormat());
+    Assertions.assertEquals(
+        Set.of(ResponseFormat.delta, ResponseFormat.parquet), capabilities.responseFormats());
     Assertions.assertEquals(
         Set.of(ReaderFeatures.COLUMN_MAPPING, ReaderFeatures.DOMAIN_METADATA),
         capabilities.readerFeatures());

server/core/src/main/java/io/whitefox/core/FileFormat.java

@@ -0,0 +1,5 @@
+package io.whitefox.core;
+
+public enum FileFormat {
+  parquet
+}

server/core/src/main/java/io/whitefox/core/Metadata.java

@@ -1,6 +1,5 @@
 package io.whitefox.core;
 
-import io.whitefox.core.services.capabilities.ResponseFormat;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -11,7 +10,7 @@ public class Metadata {
   String id;
   Optional<String> name;
   Optional<String> description;
-  ResponseFormat format;
+  FileFormat format;
   TableSchema tableSchema;
   List<String> partitionColumns;
   Map<String, String> configuration;