Release Python Connector 1.3.3 (#740)

* fix python lint and reformat scripts (#668)

* Reduce memory usage in parquet format code path (#737)

* refactor: Resolve lint errors in python release script. (#660)

* Reduce memory usage in delta format code paths (#723)

* Update Python connector version to 1.3.3

---------

Co-authored-by: Kyle Chui <48545987+kylechui@users.noreply.github.com>

Author: PatrickJin-db

client/src/test/scala/io/delta/sharing/client/DeltaSharingRestClientSuite.scala

@@ -179,7 +179,8 @@ class DeltaSharingRestClientSuite extends DeltaSharingIntegrationTest {
         Table(name = "table_with_cm_id", schema = "default", share = "share8"),
         Table(name = "deletion_vectors_with_dvs_dv_property_on", schema = "default", share = "share8"),
         Table(name = "dv_and_cm_table", schema = "default", share = "share8"),
-        Table(name = "timestampntz_cdf_table", schema = "default", share = "share8")
+        Table(name = "timestampntz_cdf_table", schema = "default", share = "share8"),
+        Table(name = "12k_rows", schema = "default", share = "share8")
       )
       assert(expected == client.listAllTables().toSet)
     } finally {

dev/python_release.sh

@@ -1,31 +1,32 @@
-#!/bin/bash -e -pipe
+#!/usr/bin/env bash
+set -euo pipefail
 
 # Switch to the project root directory
-cd $( dirname $0 )
-cd ..
+pushd "$(dirname "$0")/.."
 
 # Clean existing artifacts
-cd python
+pushd python
 python3 setup.py clean --all
 rm -rf delta_sharing.egg-info dist
-cd ..
+popd
 
 printf "Please type the python release version: "
-read VERSION
-echo $VERSION
+read -r VERSION
+echo "$VERSION"
 
 # Update the Python connector version
-sed -i '' "s/__version__ = \".*\"/__version__ = \"$VERSION\"/g"  python/delta_sharing/version.py
-git add python/delta_sharing/version.py
+sed -i'' "s/^__version__ = \".*\"$/__version__ = \"$VERSION\"/" ./python/delta_sharing/version.py
+git add ./python/delta_sharing/version.py
 # Use --allow-empty so that we can re-run this script even if the Python connector version has been updated
 git commit -m "Update Python connector version to $VERSION" --allow-empty
 
 # This creates a lightweight tag that points to the current commit.
-git tag py-v$VERSION
+git tag "py-v$VERSION"
 
 # Generate Python artifacts
-cd python/
+pushd python
 python3 setup.py sdist bdist_wheel
-cd ..
+popd
 
 echo "=== Generated all release artifacts ==="
+popd

python/delta_sharing/_internal_auth.py

@@ -35,12 +35,16 @@
 
 
 class AuthConfig:
-    def __init__(self, token_exchange_max_retries=5,
-                 token_exchange_max_retry_duration_in_seconds=60,
-                 token_renewal_threshold_in_seconds=600):
+    def __init__(
+        self,
+        token_exchange_max_retries=5,
+        token_exchange_max_retry_duration_in_seconds=60,
+        token_renewal_threshold_in_seconds=600,
+    ):
         self.token_exchange_max_retries = token_exchange_max_retries
         self.token_exchange_max_retry_duration_in_seconds = (
-            token_exchange_max_retry_duration_in_seconds)
+            token_exchange_max_retry_duration_in_seconds
+        )
         self.token_renewal_threshold_in_seconds = token_renewal_threshold_in_seconds
 
 
@@ -90,7 +94,10 @@ def __init__(self, endpoint: str, username: str, password: str):
 
     def add_auth_header(self, session: requests.Session) -> None:
         session.auth = (self.username, self.password)
-        session.post(self.endpoint, data={"grant_type": "client_credentials"},)
+        session.post(
+            self.endpoint,
+            data={"grant_type": "client_credentials"},
+        )
 
     def is_expired(self) -> bool:
         return False
@@ -107,23 +114,22 @@ def __init__(self, access_token: str, expires_in: int, creation_timestamp: int):
 
 
 class OAuthClient:
-    def __init__(self,
-                 token_endpoint: str,
-                 client_id: str,
-                 client_secret: str,
-                 scope: Optional[str] = None):
+    def __init__(
+        self, token_endpoint: str, client_id: str, client_secret: str, scope: Optional[str] = None
+    ):
         self.token_endpoint = token_endpoint
         self.client_id = client_id
         self.client_secret = client_secret
         self.scope = scope
 
     def client_credentials(self) -> OAuthClientCredentials:
         credentials = base64.b64encode(
-            f"{self.client_id}:{self.client_secret}".encode('utf-8')).decode('utf-8')
+            f"{self.client_id}:{self.client_secret}".encode("utf-8")
+        ).decode("utf-8")
         headers = {
-            'accept': 'application/json',
-            'authorization': f'Basic {credentials}',
-            'content-type': 'application/x-www-form-urlencoded'
+            "accept": "application/json",
+            "authorization": f"Basic {credentials}",
+            "content-type": "application/x-www-form-urlencoded",
         }
         body = f"grant_type=client_credentials{f'&scope={self.scope}' if self.scope else ''}"
         response = requests.post(self.token_endpoint, headers=headers, data=body)
@@ -136,9 +142,9 @@ def parse_oauth_token_response(self, response: str) -> OAuthClientCredentials:
         # Parsing the response per oauth spec
         # https://datatracker.ietf.org/doc/html/rfc6749#section-5.1
         json_node = json.loads(response)
-        if 'access_token' not in json_node or not isinstance(json_node['access_token'], str):
+        if "access_token" not in json_node or not isinstance(json_node["access_token"], str):
             raise RuntimeError("Missing 'access_token' field in OAuth token response")
-        if 'expires_in' not in json_node:
+        if "expires_in" not in json_node:
             raise RuntimeError("Missing 'expires_in' field in OAuth token response")
         try:
             # OAuth spec requires 'expires_in' to be an integer, e.g., 3600.
@@ -153,15 +159,13 @@ def parse_oauth_token_response(self, response: str) -> OAuthClientCredentials:
             #   -d "client_id=$CLIENT_ID" \
             #   -d "client_secret=$CLIENT_SECRET" \
             #   -d "scope=https://graph.microsoft.com/.default"
-            expires_in = int(json_node['expires_in'])  # Convert to int if it's a string
+            expires_in = int(json_node["expires_in"])  # Convert to int if it's a string
         except ValueError:
             raise RuntimeError(
                 "'expires_in' field must be an integer or a string convertible to integer"
             )
         return OAuthClientCredentials(
-            json_node['access_token'],
-            expires_in,
-            int(datetime.now().timestamp())
+            json_node["access_token"], expires_in, int(datetime.now().timestamp())
         )
 
 
@@ -172,7 +176,7 @@ def __init__(self, oauth_client: OAuthClient, auth_config: AuthConfig = AuthConf
         self.current_token: Optional[OAuthClientCredentials] = None
         self.lock = threading.RLock()
 
-    def add_auth_header(self,session: requests.Session) -> None:
+    def add_auth_header(self, session: requests.Session) -> None:
         token = self.maybe_refresh_token()
         with self.lock:
             session.headers.update(
@@ -199,9 +203,7 @@ def get_expiration_time(self) -> Optional[str]:
 
 
 class AuthCredentialProviderFactory:
-    __oauth_auth_provider_cache : Dict[
-        DeltaSharingProfile,
-        OAuthClientCredentialsAuthProvider] = {}
+    __oauth_auth_provider_cache: Dict[DeltaSharingProfile, OAuthClientCredentialsAuthProvider] = {}
 
     @staticmethod
     def create_auth_credential_provider(profile: DeltaSharingProfile):
@@ -210,14 +212,17 @@ def create_auth_credential_provider(profile: DeltaSharingProfile):
                 return AuthCredentialProviderFactory.__oauth_client_credentials(profile)
             elif profile.type == "basic":
                 return AuthCredentialProviderFactory.__auth_basic(profile)
-        elif (profile.share_credentials_version == 1 and
-              (profile.type is None or profile.type == "bearer_token")):
+        elif profile.share_credentials_version == 1 and (
+            profile.type is None or profile.type == "bearer_token"
+        ):
             return AuthCredentialProviderFactory.__auth_bearer_token(profile)
 
         # any other scenario is unsupported
-        raise RuntimeError(f"unsupported profile.type: {profile.type}"
-                           f" profile.share_credentials_version"
-                           f" {profile.share_credentials_version}")
+        raise RuntimeError(
+            f"unsupported profile.type: {profile.type}"
+            f" profile.share_credentials_version"
+            f" {profile.share_credentials_version}"
+        )
 
     @staticmethod
     def __oauth_client_credentials(profile):
@@ -235,11 +240,10 @@ def __oauth_client_credentials(profile):
             token_endpoint=profile.token_endpoint,
             client_id=profile.client_id,
             client_secret=profile.client_secret,
-            scope=profile.scope
+            scope=profile.scope,
         )
         provider = OAuthClientCredentialsAuthProvider(
-            oauth_client=oauth_client,
-            auth_config=AuthConfig()
+            oauth_client=oauth_client, auth_config=AuthConfig()
         )
         AuthCredentialProviderFactory.__oauth_auth_provider_cache[profile] = provider
         return provider

python/delta_sharing/delta_sharing.py

@@ -51,10 +51,7 @@ def _parse_url(url: str) -> Tuple[str, str, str, str]:
     return (profile, share, schema, table)
 
 
-def get_table_version(
-    url: str,
-    starting_timestamp: Optional[str] = None
-) -> int:
+def get_table_version(url: str, starting_timestamp: Optional[str] = None) -> int:
     """
     Get the shared table version using the given url.
 
@@ -66,8 +63,7 @@ def get_table_version(
     profile = DeltaSharingProfile.read_from_file(profile_json)
     rest_client = DataSharingRestClient(profile)
     response = rest_client.query_table_version(
-        Table(name=table, share=share, schema=schema),
-        starting_timestamp
+        Table(name=table, share=share, schema=schema), starting_timestamp
     )
     return response.delta_table_version
 
@@ -117,6 +113,7 @@ def load_as_pandas(
     timestamp: Optional[str] = None,
     jsonPredicateHints: Optional[str] = None,
     use_delta_format: Optional[bool] = None,
+    convert_in_batches: bool = False,
 ) -> pd.DataFrame:
     """
     Load the shared table using the given url as a pandas DataFrame.
@@ -128,6 +125,12 @@ def load_as_pandas(
     :param version: an optional non-negative int. Load the snapshot of table at version
     :param jsonPredicateHints: Predicate hints to be applied to the table. For more details see:
       https://github.com/delta-io/delta-sharing/blob/main/PROTOCOL.md#json-predicates-for-filtering
+    :param use_delta_format: Whether to use delta format or parquet format. If not set, table
+      metadata will be used to determine whether to use delta or parquet format.
+    :param convert_in_batches: Whether to convert the parquet files to pandas one batch at a time
+      rather than one file at a time. This may reduce memory consumption at the cost of taking
+      longer or downloading more data, with parquet format queries being more likely to see
+      improvements.
     :return: A pandas DataFrame representing the shared table.
     """
     profile_json, share, schema, table = _parse_url(url)
@@ -139,14 +142,13 @@ def load_as_pandas(
         limit=limit,
         version=version,
         timestamp=timestamp,
-        use_delta_format=use_delta_format
+        use_delta_format=use_delta_format,
+        convert_in_batches=convert_in_batches,
     ).to_pandas()
 
 
 def load_as_spark(
-    url: str,
-    version: Optional[int] = None,
-    timestamp: Optional[str] = None
+    url: str, version: Optional[int] = None, timestamp: Optional[str] = None
 ) -> "PySparkDataFrame":  # noqa: F821
     """
     Load the shared table using the given url as a Spark DataFrame. `PySpark` must be installed,
@@ -182,7 +184,7 @@ def load_table_changes_as_spark(
     starting_version: Optional[int] = None,
     ending_version: Optional[int] = None,
     starting_timestamp: Optional[str] = None,
-    ending_timestamp: Optional[str] = None
+    ending_timestamp: Optional[str] = None,
 ) -> "PySparkDataFrame":  # noqa: F821
     """
     Load the table changes of a shared table as a Spark DataFrame using the given url.
@@ -203,7 +205,8 @@ def load_table_changes_as_spark(
         from pyspark.sql import SparkSession
     except ImportError:
         raise ImportError(
-            "Unable to import pyspark. `load_table_changes_as_spark` requires PySpark.")
+            "Unable to import pyspark. `load_table_changes_as_spark` requires PySpark."
+        )
 
     spark = SparkSession.getActiveSession()
     assert spark is not None, (
@@ -228,7 +231,8 @@ def load_table_changes_as_pandas(
     ending_version: Optional[int] = None,
     starting_timestamp: Optional[str] = None,
     ending_timestamp: Optional[str] = None,
-    use_delta_format: Optional[bool] = None
+    use_delta_format: Optional[bool] = None,
+    convert_in_batches: bool = False,
 ) -> pd.DataFrame:
     """
     Load the table changes of shared table as a pandas DataFrame using the given url.
@@ -241,23 +245,32 @@ def load_table_changes_as_pandas(
     :param ending_version: The ending version of table changes.
     :param starting_timestamp: The starting timestamp of table changes.
     :param ending_timestamp: The ending timestamp of table changes.
+    :param use_delta_format: Whether to use delta format or parquet format. Default is parquet
+      format.
+    :param convert_in_batches: Whether to convert the parquet files to pandas one batch at a time
+      rather than one file at a time. This may reduce memory consumption at the cost of taking
+      longer or downloading more data, with parquet format queries being more likely to see
+      improvements.
     :return: A pandas DataFrame representing the shared table.
     """
     profile_json, share, schema, table = _parse_url(url)
     profile = DeltaSharingProfile.read_from_file(profile_json)
     return DeltaSharingReader(
         table=Table(name=table, share=share, schema=schema),
         rest_client=DataSharingRestClient(profile),
-        use_delta_format=use_delta_format
-    ).table_changes_to_pandas(CdfOptions(
-        starting_version=starting_version,
-        ending_version=ending_version,
-        starting_timestamp=starting_timestamp,
-        ending_timestamp=ending_timestamp,
-        # when using delta format, we need to get metadata changes and
-        # handle them properly when replaying the delta log
-        include_historical_metadata=use_delta_format
-    ))
+        use_delta_format=use_delta_format,
+        convert_in_batches=convert_in_batches,
+    ).table_changes_to_pandas(
+        CdfOptions(
+            starting_version=starting_version,
+            ending_version=ending_version,
+            starting_timestamp=starting_timestamp,
+            ending_timestamp=ending_timestamp,
+            # when using delta format, we need to get metadata changes and
+            # handle them properly when replaying the delta log
+            include_historical_metadata=use_delta_format,
+        )
+    )
 
 
 class SharingClient:

python/delta_sharing/protocol.py

@@ -97,7 +97,7 @@ def from_json(json) -> "DeltaSharingProfile":
                     type=type,
                     endpoint=endpoint,
                     bearer_token=json["bearerToken"],
-                    expiration_time=json.get("expirationTime")
+                    expiration_time=json.get("expirationTime"),
                 )
             elif type == "basic":
                 return DeltaSharingProfile(
@@ -109,8 +109,8 @@ def from_json(json) -> "DeltaSharingProfile":
                 )
             else:
                 raise ValueError(
-                    f"The current release does not supports {type} type. "
-                    "Please check type.")
+                    f"The current release does not supports {type} type. " "Please check type."
+                )
         else:
             raise ValueError(
                 "'shareCredentialsVersion' in the profile is "
@@ -153,8 +153,7 @@ class Table:
     def from_json(json) -> "Table":
         if isinstance(json, (str, bytes, bytearray)):
             json = loads(json)
-        return Table(name=json["name"], share=json["share"],
-                     schema=json["schema"])
+        return Table(name=json["name"], share=json["share"], schema=json["schema"])
 
 
 @dataclass(frozen=True)
@@ -234,7 +233,7 @@ def from_json(json) -> "Metadata":
                 version=json.get("version", None),
                 size=json.get("size", None),
                 num_files=json.get("numFiles", None),
-                created_time=delta_metadata.get("createdTime", None)
+                created_time=delta_metadata.get("createdTime", None),
             )
         else:
             configuration = json.get("configuration", {})
@@ -248,7 +247,7 @@ def from_json(json) -> "Metadata":
                 partition_columns=json["partitionColumns"],
                 version=json.get("version", None),
                 size=json.get("size", None),
-                num_files=json.get("numFiles", None)
+                num_files=json.get("numFiles", None),
             )