Autoresolve query format and support delta format RPC call

python/delta_sharing/reader.py

@@ -86,6 +86,11 @@ def limit(self, limit: Optional[int]) -> "DeltaSharingReader":
         )
 
     def to_pandas(self) -> pd.DataFrame:
+        response_format = self._rest_client.autoresolve_query_format(self._table)
+
+        if (response_format == DataSharingRestClient.DELTA_FORMAT):
+            raise Exception("Delta format not supported in query yet.")
+
         response = self._rest_client.list_files_in_table(
             self._table,
             predicateHints=self._predicateHints,

python/delta_sharing/rest_client.py

@@ -138,6 +138,11 @@ def _client_user_agent() -> str:
 
 class DataSharingRestClient:
     USER_AGENT: ClassVar[str] = _client_user_agent()
+    DELTA_RESPONSE_FORMAT = "responseformat=delta,parquet;"
+    DELTA_READER_FEATURES = "readerfeatures=deletionvectors,columnmapping"
+    DELTA_SHARING_CAPABILITIES_HEADER = "delta-sharing-capabilities"
+    DELTA_FORMAT = "delta"
+    PARQUET_FORMAT = "parquet"
 
     def __init__(self, profile: DeltaSharingProfile, num_retries=10):
         self._profile = profile
@@ -203,6 +208,24 @@ def __auth_basic(self, profile):
             }
         )
 
+    def set_delta_format_header(self):
+        delta_sharing_capabilities = (
+            DataSharingRestClient.DELTA_RESPONSE_FORMAT +
+            DataSharingRestClient.DELTA_READER_FEATURES
+        )
+        self._session.headers.update(
+            {
+                DataSharingRestClient.DELTA_SHARING_CAPABILITIES_HEADER: delta_sharing_capabilities,
+            }
+        )
+
+    def remove_delta_format_header(self):
+        self._session.headers.update(
+            {
+                DataSharingRestClient.DELTA_SHARING_CAPABILITIES_HEADER: "",
+            }
+        )
+
     @retry_with_exponential_backoff
     def list_shares(
         self, *, max_results: Optional[int] = None, page_token: Optional[str] = None
@@ -294,6 +317,39 @@ def query_table_metadata(self, table: Table) -> QueryTableMetadataResponse:
                 metadata=Metadata.from_json(metadata_json["metaData"]),
             )
 
+    @retry_with_exponential_backoff
+    def autoresolve_query_format(self, table: Table):
+        """
+        This function determines the query format for the table (parquet or delta).
+        It sends a query table metadata request with capabilities set to parquet and delta
+        and uses what the server responds to use in the header.
+        """
+        self.set_delta_format_header()
+
+        with self._get_internal(
+            f"/shares/{table.share}/schemas/{table.schema}/tables/{table.name}/metadata",
+            return_headers=True
+        ) as values:
+            headers = values[0]
+            # it's a bug in the server if it doesn't return delta-table-version in the header
+            if "delta-table-version" not in headers:
+                raise LookupError("Missing delta-table-version header")
+            if DataSharingRestClient.DELTA_SHARING_CAPABILITIES_HEADER not in headers:
+                return DataSharingRestClient.PARQUET_FORMAT
+
+            # the response_format will either be responseformat=delta or responseformat=parquet
+            response_format = headers[DataSharingRestClient.DELTA_SHARING_CAPABILITIES_HEADER]
+
+            # we now parse it to get either "delta" or "parquet"
+            if (DataSharingRestClient.DELTA_FORMAT in response_format):
+                return DataSharingRestClient.DELTA_FORMAT
+            else:
+                return DataSharingRestClient.PARQUET_FORMAT
+
+            # removing the client-reader-features that were set to avoid diverging standard codepath
+            self.remove_delta_format_header()
+            return response_format
+
     @retry_with_exponential_backoff
     def query_table_version(
         self,

python/delta_sharing/tests/test_delta_sharing.py

@@ -647,6 +647,47 @@ def test_load_as_pandas_exception(
         assert error in str(e)
 
 
+@pytest.mark.skipif(not ENABLE_INTEGRATION, reason=SKIP_MESSAGE)
+@pytest.mark.parametrize(
+    "fragments,version,timestamp,error",
+    [
+        pytest.param(
+            "share8.default.deletion_vectors_with_dvs_dv_property_on",
+            None,
+            None,
+            "Delta format not supported in query yet.",
+            id="deletion vector not supported",
+        ),
+        pytest.param(
+            "share8.default.table_with_cm_id",
+            None,
+            None,
+            "Delta format not supported in query yet.",
+            id="column mapping id not supported",
+        ),
+        pytest.param(
+            "share8.default.table_with_cm_name",
+            None,
+            None,
+            "Delta format not supported in query yet.",
+            id="column mapping name not supported",
+        ),
+    ],
+)
+def test_load_as_pandas_exception_dv_cm(
+    profile_path: str,
+    fragments: str,
+    version: Optional[int],
+    timestamp: Optional[str],
+    error: Optional[str]
+):
+    try:
+        load_as_pandas(f"{profile_path}#{fragments}", None, version, timestamp)
+        assert False
+    except Exception as e:
+        assert error in str(e)
+
+
 @pytest.mark.skipif(not ENABLE_INTEGRATION, reason=SKIP_MESSAGE)
 @pytest.mark.parametrize(
     "fragments,starting_version,ending_version,starting_timestamp,ending_timestamp,error,expected",

python/delta_sharing/tests/test_reader.py

@@ -84,6 +84,9 @@ def list_files_in_table(
                 delta_table_version=1, protocol=None, metadata=metadata, add_files=add_files
             )
 
+        def autoresolve_query_format(self, table: Table):
+            return "parquet"
+
     reader = DeltaSharingReader(Table("table_name", "share_name", "schema_name"), RestClientMock())
     pdf = reader.to_pandas()
     expected = pd.concat([pdf1, pdf2]).reset_index(drop=True)
@@ -156,6 +159,9 @@ def list_files_in_table(
                 delta_table_version=1, protocol=None, metadata=metadata, add_files=add_files
             )
 
+        def autoresolve_query_format(self, table: Table):
+            return "parquet"
+
     reader = DeltaSharingReader(Table("table_name", "share_name", "schema_name"), RestClientMock())
     pdf = reader.to_pandas()
 
@@ -217,6 +223,9 @@ def list_files_in_table(
                 delta_table_version=1, protocol=None, metadata=metadata, add_files=add_files
             )
 
+        def autoresolve_query_format(self, table: Table):
+            return "parquet"
+
     reader = DeltaSharingReader(Table("table_name", "share_name", "schema_name"), RestClientMock())
     pdf = reader.to_pandas()
 
@@ -274,6 +283,9 @@ def list_files_in_table(
                 delta_table_version=1, protocol=None, metadata=metadata, add_files=add_files
             )
 
+        def autoresolve_query_format(self, table: Table):
+            return "parquet"
+
     reader = DeltaSharingReader(
         Table("table_name", "share_name", "schema_name"), RestClientMock()  # type: ignore
     )
@@ -378,6 +390,9 @@ def list_table_changes(
             ]
             return ListTableChangesResponse(protocol=None, metadata=metadata, actions=actions)
 
+        def autoresolve_query_format(self, table: Table):
+            return "parquet"
+
     reader = DeltaSharingReader(Table("table_name", "share_name", "schema_name"), RestClientMock())
     pdf = reader.table_changes_to_pandas(CdfOptions())
 

python/delta_sharing/tests/test_rest_client.py

@@ -217,6 +217,26 @@ def test_query_table_metadata_partitioned_different_schemas(
     )
 
 
+@pytest.mark.skipif(not ENABLE_INTEGRATION, reason=SKIP_MESSAGE)
+def test_autoresolve_query_format(
+    rest_client: DataSharingRestClient,
+):
+    tables = [
+        ("table3", "share1", "parquet"),
+        ('deletion_vectors_with_dvs_dv_property_on', "share8", "delta"),
+        ('table_with_cm_name', "share8", "delta")]
+
+    for table in tables:
+        table_name = table[0]
+        table_share = table[1]
+        expected_format = table[2]
+
+        resolved_format = rest_client.autoresolve_query_format(
+            Table(name=table_name, share=table_share, schema="default")
+        )
+        assert resolved_format == expected_format
+
+
 @pytest.mark.skipif(not ENABLE_INTEGRATION, reason=SKIP_MESSAGE)
 def test_query_existed_table_version(rest_client: DataSharingRestClient):
     response = rest_client.query_table_version(