Fix after merging upstream/main.

Author: mbutrovich

.github/workflows/stale.yml

@@ -17,28 +17,28 @@
 # under the License.
 #
 
-name: "Close Stale Issues"
+name: "Close Stale Issues and PRs"
 on:
   schedule:
     - cron: '0 0 * * *'
 
 permissions:
   # All other permissions are set to none
   issues: write
+  pull-requests: write
 
 jobs:
   stale:
     if: github.repository_owner == 'apache'
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-24.04
     steps:
       - uses: actions/stale@v10.2.0
         with:
-          stale-issue-label: 'stale'
+          # stale issues
+          stale-issue-label: 'stale,security'
           exempt-issue-labels: 'not-stale'
           days-before-issue-stale: 180
           days-before-issue-close: 14
-          # Only close stale issues, leave PRs alone
-          days-before-pr-stale: -1
           stale-issue-message: >
             This issue has been automatically marked as stale because it has been open for 180 days
             with no activity. It will be closed in next 14 days if no further activity occurs. To
@@ -47,3 +47,12 @@ jobs:
           close-issue-message: >
             This issue has been closed because it has not received any activity in the last 14 days
             since being marked as 'stale'
+          # stale PRs
+          stale-pr-label: 'stale'
+          exempt-pr-labels: 'not-stale,security'
+          stale-pr-message: 'This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.'
+          close-pr-message: 'This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.'
+          days-before-pr-stale: 30
+          days-before-pr-close: 7
+          ascending: true
+          operations-per-run: 200

bindings/python/Makefile

@@ -15,12 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
-
-install:
-	uv sync --group dev --no-install-project
+.PHONY: build install test
 
 build:
 	uv run maturin develop
 
+# uv sync may remove local editable pyiceberg-core, so install rebuilds it afterwards.
+install:
+	uv sync --group dev --no-install-project
+	$(MAKE) build
+
 test:
 	uv run --no-sync pytest

bindings/python/pyproject.toml

@@ -51,6 +51,11 @@ include = [
 [tool.ruff.lint]
 ignore = ["F403", "F405"]
 
+[tool.pytest.ini_options]
+filterwarnings = [
+  "error",
+]
+
 [dependency-groups]
 dev = [
     "maturin>=1.0,<2.0",

bindings/python/tests/conftest.py

@@ -25,6 +25,7 @@
 
 import pytest
 from pydantic_core import to_json
+from pyiceberg.utils.config import Config
 
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
@@ -36,6 +37,20 @@
 )
 
 
+@pytest.fixture(scope="session", autouse=True)
+def isolate_pyiceberg_config() -> Generator[None, None, None]:
+    monkeypatch = pytest.MonkeyPatch()
+    with TemporaryDirectory() as empty_home_dir:
+        monkeypatch.setenv("HOME", empty_home_dir)
+        monkeypatch.setenv("PYICEBERG_HOME", empty_home_dir)
+
+        import pyiceberg.catalog as catalog
+
+        monkeypatch.setattr(catalog, "_ENV_CONFIG", Config())
+        yield
+    monkeypatch.undo()
+
+
 @pytest.fixture(scope="session")
 def avro_schema_manifest_entry() -> Dict[str, Any]:
     return {

bindings/python/tests/test_datafusion_table_provider.py

@@ -106,7 +106,7 @@ def test_register_iceberg_table_provider(
     )
 
     ctx = SessionContext()
-    ctx.register_table_provider("test", iceberg_table_provider)
+    ctx.register_table("test", iceberg_table_provider)
 
     datafusion_table = ctx.table("test")
     assert datafusion_table is not None
@@ -154,7 +154,7 @@ def __datafusion_table_provider__(self):
     )
 
     ctx = SessionContext()
-    ctx.register_table_provider("test", iceberg_table)
+    ctx.register_table("test", iceberg_table)
 
     datafusion_table = ctx.table("test")
     assert datafusion_table is not None

crates/iceberg/src/arrow/caching_delete_file_loader.rs

@@ -235,7 +235,7 @@ impl CachingDeleteFileLoader {
                     PosDelLoadAction::Load => Ok(DeleteFileContext::PosDels {
                         file_path: task.file_path.clone(),
                         stream: basic_delete_file_loader
-                            .parquet_to_batch_stream(&task.file_path)
+                            .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
                             .await?,
                     }),
                 }
@@ -254,7 +254,7 @@ impl CachingDeleteFileLoader {
                 let equality_ids_vec = task.equality_ids.clone().unwrap();
                 let evolved_stream = BasicDeleteFileLoader::evolve_schema(
                     basic_delete_file_loader
-                        .parquet_to_batch_stream(&task.file_path)
+                        .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
                         .await?,
                     schema,
                     &equality_ids_vec,
@@ -614,7 +614,10 @@ mod tests {
 
         let basic_delete_file_loader = BasicDeleteFileLoader::new(file_io.clone());
         let record_batch_stream = basic_delete_file_loader
-            .parquet_to_batch_stream(&eq_delete_file_path)
+            .parquet_to_batch_stream(
+                &eq_delete_file_path,
+                std::fs::metadata(&eq_delete_file_path).unwrap().len(),
+            )
             .await
             .expect("could not get batch stream");
 
@@ -811,7 +814,10 @@ mod tests {
         let basic_delete_file_loader = BasicDeleteFileLoader::new(file_io.clone());
 
         let batch_stream = basic_delete_file_loader
-            .parquet_to_batch_stream(&delete_file_path)
+            .parquet_to_batch_stream(
+                &delete_file_path,
+                std::fs::metadata(&delete_file_path).unwrap().len(),
+            )
             .await
             .unwrap();
 
@@ -913,14 +919,16 @@ mod tests {
 
         // Create FileScanTask with BOTH positional and equality deletes
         let pos_del = FileScanTaskDeleteFile {
-            file_path: pos_del_path,
+            file_path: pos_del_path.clone(),
+            file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(),
             file_type: DataContentType::PositionDeletes,
             partition_spec_id: 0,
             equality_ids: None,
         };
 
         let eq_del = FileScanTaskDeleteFile {
             file_path: eq_delete_path.clone(),
+            file_size_in_bytes: std::fs::metadata(&eq_delete_path).unwrap().len(),
             file_type: DataContentType::EqualityDeletes,
             partition_spec_id: 0,
             equality_ids: Some(vec![2, 3]), // Only use field IDs that exist in both schemas
@@ -994,7 +1002,7 @@ mod tests {
 
         let basic_delete_file_loader = BasicDeleteFileLoader::new(file_io.clone());
         let record_batch_stream = basic_delete_file_loader
-            .parquet_to_batch_stream(&path)
+            .parquet_to_batch_stream(&path, std::fs::metadata(&path).unwrap().len())
             .await
             .expect("could not get batch stream");
 

crates/iceberg/src/arrow/delete_file_loader.rs

@@ -55,20 +55,21 @@ impl BasicDeleteFileLoader {
     pub(crate) async fn parquet_to_batch_stream(
         &self,
         data_file_path: &str,
+        file_size_in_bytes: u64,
     ) -> Result<ArrowRecordBatchStream> {
         /*
            Essentially a super-cut-down ArrowReader. We can't use ArrowReader directly
            as that introduces a circular dependency.
         */
         let parquet_metadata =
-            ArrowReader::load_parquet_metadata(data_file_path, &self.file_io, false, None).await?;
+            ArrowReader::load_parquet_metadata(data_file_path, &self.file_io, false, file_size_in_bytes, None).await?;
 
         let record_batch_stream = ArrowReader::create_parquet_record_batch_stream_builder(
             data_file_path,
             self.file_io.clone(),
             false,
             ArrowReaderOptions::default(),
-            None,
+            file_size_in_bytes,
             parquet_metadata,
         )
         .await?
@@ -107,7 +108,9 @@ impl DeleteFileLoader for BasicDeleteFileLoader {
         task: &FileScanTaskDeleteFile,
         schema: SchemaRef,
     ) -> Result<ArrowRecordBatchStream> {
-        let raw_batch_stream = self.parquet_to_batch_stream(&task.file_path).await?;
+        let raw_batch_stream = self
+            .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
+            .await?;
 
         // For equality deletes, only evolve the equality_ids columns.
         // For positional deletes (equality_ids is None), use all field IDs.

crates/iceberg/src/arrow/delete_filter.rs

@@ -378,20 +378,38 @@ pub(crate) mod tests {
 
         let pos_del_1 = FileScanTaskDeleteFile {
             file_path: format!("{}/pos-del-1.parquet", table_location.to_str().unwrap()),
+            file_size_in_bytes: std::fs::metadata(format!(
+                "{}/pos-del-1.parquet",
+                table_location.to_str().unwrap()
+            ))
+            .unwrap()
+            .len(),
             file_type: DataContentType::PositionDeletes,
             partition_spec_id: 0,
             equality_ids: None,
         };
 
         let pos_del_2 = FileScanTaskDeleteFile {
             file_path: format!("{}/pos-del-2.parquet", table_location.to_str().unwrap()),
+            file_size_in_bytes: std::fs::metadata(format!(
+                "{}/pos-del-2.parquet",
+                table_location.to_str().unwrap()
+            ))
+            .unwrap()
+            .len(),
             file_type: DataContentType::PositionDeletes,
             partition_spec_id: 0,
             equality_ids: None,
         };
 
         let pos_del_3 = FileScanTaskDeleteFile {
             file_path: format!("{}/pos-del-3.parquet", table_location.to_str().unwrap()),
+            file_size_in_bytes: std::fs::metadata(format!(
+                "{}/pos-del-3.parquet",
+                table_location.to_str().unwrap()
+            ))
+            .unwrap()
+            .len(),
             file_type: DataContentType::PositionDeletes,
             partition_spec_id: 0,
             equality_ids: None,
@@ -477,6 +495,7 @@ pub(crate) mod tests {
             predicate: None,
             deletes: vec![FileScanTaskDeleteFile {
                 file_path: "eq-del.parquet".to_string(),
+                file_size_in_bytes: 1, // never read; this test fails before opening the file
                 file_type: DataContentType::EqualityDeletes,
                 partition_spec_id: 0,
                 equality_ids: None,