Merge branch 'apache:main' into fix-3008-sigv4-retries

Author: mrutunjay-kinagi

.github/workflows/python-ci-docs.yml

@@ -41,8 +41,6 @@ jobs:
           python-version: 3.12
       - name: Install UV
         uses: astral-sh/setup-uv@v7
-      - name: Install
-        run: make docs-install
       - name: Build docs
         run: make docs-build
       - name: Run linters

.github/workflows/python-release-docs.yml

@@ -36,8 +36,6 @@ jobs:
           python-version: ${{ matrix.python }}
       - name: Install UV
         uses: astral-sh/setup-uv@v7
-      - name: Install docs
-        run: make docs-install
       - name: Build docs
         run: make docs-build
       - name: Copy

Makefile

@@ -14,6 +14,13 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+.PHONY: help install install-uv check-license lint \
+        test test-integration test-integration-setup test-integration-exec test-integration-cleanup test-integration-rebuild \
+        test-s3 test-adls test-gcs test-coverage coverage-report \
+        docs-serve docs-build notebook notebook-infra \
+        clean uv-lock uv-lock-check
+
+.DEFAULT_GOAL := help
 # ========================
 # Configuration Variables
 # ========================
@@ -149,29 +156,23 @@ coverage-report: ## Combine and report coverage
 
 ##@ Documentation
 
-docs-install: ## Install docs dependencies (included in default groups)
-	uv sync $(PYTHON_ARG) --group docs
-
 docs-serve: ## Serve local docs preview (hot reload)
-	uv run $(PYTHON_ARG) mkdocs serve -f mkdocs/mkdocs.yml --livereload
+	uv run $(PYTHON_ARG) --group docs mkdocs serve -f mkdocs/mkdocs.yml --livereload
 
 docs-build: ## Build the static documentation site
-	uv run $(PYTHON_ARG) mkdocs build -f mkdocs/mkdocs.yml --strict
+	uv run $(PYTHON_ARG) --group docs mkdocs build -f mkdocs/mkdocs.yml --strict
 
 # ========================
 # Experimentation
 # ========================
 
 ##@ Experimentation
 
-notebook-install: ## Install notebook dependencies
-	uv sync $(PYTHON_ARG) --all-extras --group notebook
-
-notebook: notebook-install ## Launch notebook for experimentation
-	uv run jupyter lab --notebook-dir=notebooks
+notebook: ## Launch notebook for experimentation
+	uv run $(PYTHON_ARG) --all-extras --group notebook jupyter lab --notebook-dir=notebooks
 
-notebook-infra: notebook-install test-integration-setup ## Launch notebook with integration test infra (Spark, Iceberg Rest Catalog, object storage, etc.)
-	uv run jupyter lab --notebook-dir=notebooks
+notebook-infra: test-integration-setup ## Launch notebook with integration test infra (Spark, Iceberg Rest Catalog, object storage, etc.)
+	uv run $(PYTHON_ARG) --all-extras --group notebook jupyter lab --notebook-dir=notebooks
 
 # ===================
 # Project Maintenance
@@ -189,6 +190,8 @@ clean: ## Remove build artifacts and caches
 	@find . -name "*.pyo" -exec echo Deleting {} \; -delete
 	@echo "Cleaning up Jupyter notebook checkpoints..."
 	@find . -name ".ipynb_checkpoints" -exec echo Deleting {} \; -exec rm -rf {} +
+	@echo "Cleaning up coverage files..."
+	@rm -rf .coverage .coverage.* htmlcov/ coverage.xml
 	@echo "Cleanup complete."
 
 uv-lock: ## Regenerate uv.lock file from pyproject.toml

mkdocs/README.md

@@ -22,6 +22,5 @@ The pyiceberg docs are stored in `docs/`.
 ## Running docs locally
 
 ```sh
-make docs-install
 make docs-serve
 ```

pyiceberg/io/fsspec.py

@@ -29,7 +29,7 @@
     TYPE_CHECKING,
     Any,
 )
-from urllib.parse import urlparse
+from urllib.parse import ParseResult, urlparse
 
 import requests
 from fsspec import AbstractFileSystem
@@ -244,7 +244,7 @@ def _gs(properties: Properties) -> AbstractFileSystem:
     )
 
 
-def _adls(properties: Properties) -> AbstractFileSystem:
+def _adls(properties: Properties, hostname: str | None = None) -> AbstractFileSystem:
     # https://fsspec.github.io/adlfs/api/
 
     from adlfs import AzureBlobFileSystem
@@ -259,6 +259,10 @@ def _adls(properties: Properties) -> AbstractFileSystem:
         if ADLS_SAS_TOKEN not in properties:
             properties[ADLS_SAS_TOKEN] = sas_token
 
+    # Fallback: extract account_name from URI hostname (e.g. "account.dfs.core.windows.net" -> "account")
+    if hostname and ADLS_ACCOUNT_NAME not in properties:
+        properties[ADLS_ACCOUNT_NAME] = hostname.split(".")[0]
+
     class StaticTokenCredential(AsyncTokenCredential):
         _DEFAULT_EXPIRY_SECONDS = 3600
 
@@ -300,7 +304,7 @@ def _hf(properties: Properties) -> AbstractFileSystem:
     )
 
 
-SCHEME_TO_FS = {
+SCHEME_TO_FS: dict[str, Callable[..., AbstractFileSystem]] = {
     "": _file,
     "file": _file,
     "s3": _s3,
@@ -313,6 +317,8 @@ def _hf(properties: Properties) -> AbstractFileSystem:
     "hf": _hf,
 }
 
+_ADLS_SCHEMES = frozenset({"abfs", "abfss", "wasb", "wasbs"})
+
 
 class FsspecInputFile(InputFile):
     """An input file implementation for the FsspecFileIO.
@@ -414,8 +420,7 @@ class FsspecFileIO(FileIO):
     """A FileIO implementation that uses fsspec."""
 
     def __init__(self, properties: Properties):
-        self._scheme_to_fs = {}
-        self._scheme_to_fs.update(SCHEME_TO_FS)
+        self._scheme_to_fs: dict[str, Callable[..., AbstractFileSystem]] = dict(SCHEME_TO_FS)
         self._thread_locals = threading.local()
         super().__init__(properties=properties)
 
@@ -429,7 +434,7 @@ def new_input(self, location: str) -> FsspecInputFile:
             FsspecInputFile: An FsspecInputFile instance for the given location.
         """
         uri = urlparse(location)
-        fs = self.get_fs(uri.scheme)
+        fs = self._get_fs_from_uri(uri)
         return FsspecInputFile(location=location, fs=fs)
 
     def new_output(self, location: str) -> FsspecOutputFile:
@@ -442,7 +447,7 @@ def new_output(self, location: str) -> FsspecOutputFile:
             FsspecOutputFile: An FsspecOutputFile instance for the given location.
         """
         uri = urlparse(location)
-        fs = self.get_fs(uri.scheme)
+        fs = self._get_fs_from_uri(uri)
         return FsspecOutputFile(location=location, fs=fs)
 
     def delete(self, location: str | InputFile | OutputFile) -> None:
@@ -459,20 +464,30 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
             str_location = location
 
         uri = urlparse(str_location)
-        fs = self.get_fs(uri.scheme)
+        fs = self._get_fs_from_uri(uri)
         fs.rm(str_location)
 
-    def get_fs(self, scheme: str) -> AbstractFileSystem:
+    def _get_fs_from_uri(self, uri: "ParseResult") -> AbstractFileSystem:
+        """Get a filesystem from a parsed URI, using hostname for ADLS account resolution."""
+        if uri.scheme in _ADLS_SCHEMES:
+            return self.get_fs(uri.scheme, uri.hostname)
+        return self.get_fs(uri.scheme)
+
+    def get_fs(self, scheme: str, hostname: str | None = None) -> AbstractFileSystem:
         """Get a filesystem for a specific scheme, cached per thread."""
         if not hasattr(self._thread_locals, "get_fs_cached"):
             self._thread_locals.get_fs_cached = lru_cache(self._get_fs)
 
-        return self._thread_locals.get_fs_cached(scheme)
+        return self._thread_locals.get_fs_cached(scheme, hostname)
 
-    def _get_fs(self, scheme: str) -> AbstractFileSystem:
+    def _get_fs(self, scheme: str, hostname: str | None = None) -> AbstractFileSystem:
         """Get a filesystem for a specific scheme."""
         if scheme not in self._scheme_to_fs:
             raise ValueError(f"No registered filesystem for scheme: {scheme}")
+
+        if scheme in _ADLS_SCHEMES:
+            return _adls(self.properties, hostname)
+
         return self._scheme_to_fs[scheme](self.properties)
 
     def __getstate__(self) -> dict[str, Any]:

pyiceberg/table/__init__.py

@@ -26,25 +26,12 @@
 from functools import cached_property
 from itertools import chain
 from types import TracebackType
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    TypeVar,
-)
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from pydantic import Field
 
 import pyiceberg.expressions.parser as parser
-from pyiceberg.expressions import (
-    AlwaysFalse,
-    AlwaysTrue,
-    And,
-    BooleanExpression,
-    EqualTo,
-    IsNull,
-    Or,
-    Reference,
-)
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference
 from pyiceberg.expressions.visitors import (
     ResidualEvaluator,
     _InclusiveMetricsEvaluator,
@@ -54,36 +41,17 @@
     manifest_evaluator,
 )
 from pyiceberg.io import FileIO, load_file_io
-from pyiceberg.manifest import (
-    DataFile,
-    DataFileContent,
-    ManifestContent,
-    ManifestEntry,
-    ManifestFile,
-)
-from pyiceberg.partitioning import (
-    PARTITION_FIELD_ID_START,
-    UNPARTITIONED_PARTITION_SPEC,
-    PartitionKey,
-    PartitionSpec,
-)
+from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestFile
+from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.inspect import InspectTable
 from pyiceberg.table.locations import LocationProvider, load_location_provider
 from pyiceberg.table.maintenance import MaintenanceTable
-from pyiceberg.table.metadata import (
-    INITIAL_SEQUENCE_NUMBER,
-    TableMetadata,
-)
-from pyiceberg.table.name_mapping import (
-    NameMapping,
-)
+from pyiceberg.table.metadata import INITIAL_SEQUENCE_NUMBER, TableMetadata
+from pyiceberg.table.name_mapping import NameMapping
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef
-from pyiceberg.table.snapshots import (
-    Snapshot,
-    SnapshotLogEntry,
-)
+from pyiceberg.table.snapshots import Snapshot, SnapshotLogEntry
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.table.update import (
     AddPartitionSpecUpdate,
@@ -107,11 +75,7 @@
     update_table_metadata,
 )
 from pyiceberg.table.update.schema import UpdateSchema
-from pyiceberg.table.update.snapshot import (
-    ManageSnapshots,
-    UpdateSnapshot,
-    _FastAppendFiles,
-)
+from pyiceberg.table.update.snapshot import ManageSnapshots, UpdateSnapshot, _FastAppendFiles
 from pyiceberg.table.update.sorting import UpdateSortOrder
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
@@ -126,9 +90,7 @@
     Record,
     TableVersion,
 )
-from pyiceberg.types import (
-    strtobool,
-)
+from pyiceberg.types import strtobool
 from pyiceberg.utils.concurrent import ExecutorFactory
 from pyiceberg.utils.config import Config
 from pyiceberg.utils.properties import property_as_bool
@@ -144,11 +106,7 @@
     from pyiceberg_core.datafusion import IcebergDataFusionTable
 
     from pyiceberg.catalog import Catalog
-    from pyiceberg.catalog.rest.scan_planning import (
-        RESTContentFile,
-        RESTDeleteFile,
-        RESTFileScanTask,
-    )
+    from pyiceberg.catalog.rest.scan_planning import RESTContentFile, RESTDeleteFile, RESTFileScanTask
 
 ALWAYS_TRUE = AlwaysTrue()
 DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE = "downcast-ns-timestamp-to-us-on-write"
@@ -396,17 +354,19 @@ def _set_ref_snapshot(
 
         return updates, requirements
 
-    def _build_partition_predicate(self, partition_records: set[Record]) -> BooleanExpression:
+    def _build_partition_predicate(
+        self, partition_records: set[Record], spec: PartitionSpec, schema: Schema
+    ) -> BooleanExpression:
         """Build a filter predicate matching any of the input partition records.
 
         Args:
             partition_records: A set of partition records to match
+            spec: An optional partition spec, if none then defaults to current
+            schema: An optional schema, if none then defaults to current
         Returns:
             A predicate matching any of the input partition records.
         """
-        partition_spec = self.table_metadata.spec()
-        schema = self.table_metadata.schema()
-        partition_fields = [schema.find_field(field.source_id).name for field in partition_spec.fields]
+        partition_fields = [schema.find_field(field.source_id).name for field in spec.fields]
 
         expr: BooleanExpression = AlwaysFalse()
         for partition_record in partition_records:
@@ -583,7 +543,9 @@ def dynamic_partition_overwrite(
         )
 
         partitions_to_overwrite = {data_file.partition for data_file in data_files}
-        delete_filter = self._build_partition_predicate(partition_records=partitions_to_overwrite)
+        delete_filter = self._build_partition_predicate(
+            partition_records=partitions_to_overwrite, spec=self.table_metadata.spec(), schema=self.table_metadata.schema()
+        )
         self.delete(delete_filter=delete_filter, snapshot_properties=snapshot_properties, branch=branch)
 
         with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
@@ -673,11 +635,7 @@ def delete(
             case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
             branch: Branch Reference to run the delete operation
         """
-        from pyiceberg.io.pyarrow import (
-            ArrowScan,
-            _dataframe_to_data_files,
-            _expression_to_complementary_pyarrow,
-        )
+        from pyiceberg.io.pyarrow import ArrowScan, _dataframe_to_data_files, _expression_to_complementary_pyarrow
 
         if (
             self.table_metadata.properties.get(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_DEFAULT)

pyiceberg/table/delete_file_index.py

@@ -54,6 +54,10 @@ def filter_by_seq(self, seq: int) -> list[DataFile]:
         start_idx = bisect_left(self._seqs, seq)
         return [delete_file for delete_file, _ in self._files[start_idx:]]
 
+    def referenced_delete_files(self) -> list[DataFile]:
+        self._ensure_indexed()
+        return [data_file for data_file, _ in self._files]
+
 
 def _has_path_bounds(delete_file: DataFile) -> bool:
     lower = delete_file.lower_bounds
@@ -140,3 +144,14 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
             deletes.update(path_deletes.filter_by_seq(seq_num))
 
         return deletes
+
+    def referenced_delete_files(self) -> list[DataFile]:
+        data_files: list[DataFile] = []
+
+        for deletes in self._by_partition.values():
+            data_files.extend(deletes.referenced_delete_files())
+
+        for deletes in self._by_path.values():
+            data_files.extend(deletes.referenced_delete_files())
+
+        return data_files