fix: update delta.py for deltalake 1.1.3 API compatibility

- Update DeltaTable.create() method signature and instantiation
- Change optimize.compact() to optimize().compact()
- Replace partitions() with get_partitions()
- Replace version() with get_version()
- Replace load_as_version() with load_version()
- Remove deprecated engine=''rust'' parameter from write_deltalake()
- Change ''partitions='' to ''filters='' in to_pyarrow methods
- Update schema().to_pyarrow() to schema.to_pyarrow()
- Fix string concatenation in predicate building for type safety
- Add Optional import for better type annotation support

Fixes type checking issues after deltalake upgrade from 0.25.5 to 1.1.3

Co-authored-by: Yusuf Ali <[email protected]>

Author: claude[bot]

requirements.txt

@@ -5,5 +5,5 @@ simplejson==3.20.1
 flask==3.1.1
 pyyaml==6.0.2
 pyiceberg[sql-sqlite,pyarrow]==0.8.1
-deltalake==0.25.5
+deltalake==1.1.3
 azure-servicebus==7.14.2

servc/svc/com/storage/delta.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 import pyarrow as pa
 from deltalake import DeltaTable, write_deltalake
@@ -57,22 +57,23 @@ def _connect(self):
 
         tablename = self._get_table_name()
         uri = os.path.join(self._location_prefix, tablename)
-        self._conn = DeltaTable.create(
+        DeltaTable.create(
             table_uri=uri,
-            name=tablename,
             schema=self._table["schema"],
             partition_by=self._table["partitions"],
             mode="ignore",
             storage_options=self._storageOptions,
         )
+        # Now load the created table
+        self._conn = DeltaTable(uri, storage_options=self._storageOptions)
 
         return super()._connect()
 
     def optimize(self):
         table = self.getConn()
 
         print("Optimizing", self._get_table_name(), flush=True)
-        table.optimize.compact()
+        table.optimize().compact()
         table.vacuum()
         table.cleanup_metadata()
         table.create_checkpoint()
@@ -81,7 +82,7 @@ def getPartitions(self) -> Dict[str, List[Any]] | None:
         table = self.getConn()
 
         partitions: Dict[str, List[Any]] = {}
-        for obj in table.partitions():
+        for obj in table.get_partitions():
             for key, value in obj.items():
                 if key not in partitions:
                     partitions[key] = []
@@ -92,7 +93,7 @@ def getPartitions(self) -> Dict[str, List[Any]] | None:
 
     def getCurrentVersion(self) -> str | None:
         table = self.getConn()
-        return str(table.version())
+        return str(table.get_version())
 
     def getVersions(self) -> List[str] | None:
         return [str(self.getCurrentVersion())]
@@ -129,15 +130,14 @@ def overwrite(
         predicate: str | None = None
         filter = self._filters(partitions)
         if filter is not None:
-            predicate = " & ".join([" ".join(x) for x in filter])
+            predicate = " & ".join([f"{col} {op} {str(val)}" for col, op, val in filter])
 
         write_deltalake(
             table,
             data=pa.Table.from_pylist(data, self.getSchema()),
             storage_options=self._storageOptions,
             mode="overwrite",
             predicate=predicate,
-            engine="rust",
         )
         return True
 
@@ -150,7 +150,7 @@ def readRaw(
     ) -> Table:
         table = self.getConn()
         if version is not None:
-            table.load_as_version(int(version))
+            table.load_version(int(version))
 
         if options is None or not isinstance(options, dict):
             options = {}
@@ -159,14 +159,14 @@ def readRaw(
 
         if options.get("filter", None) is not None:
             return table.to_pyarrow_dataset(
-                partitions=self._filters(partitions),
+                filters=self._filters(partitions),
             ).to_table(
                 filter=options.get("filter"),
                 columns=rcolumns,
             )
         return table.to_pyarrow_table(
             columns=rcolumns,
-            partitions=self._filters(partitions),
+            filters=self._filters(partitions),
         )
 
     def read(
@@ -181,7 +181,7 @@ def read(
     def getSchema(self) -> Schema | None:
         table = self.getConn()
 
-        return table.schema().to_pyarrow()
+        return table.schema.to_pyarrow()
 
     def _close(self):
         if self._isOpen: