feat: Support passing kwargs to polars.scan_iceberg in Table.to_polars (#3868)

feat: Support passing kwargs to polars.scan_iceberg in Table.to_polars (#3128)

Co-authored-by: Kevin Liu <kevinjqliu@users.noreply.github.com>

Author: hedger9487

pyiceberg/table/__init__.py

@@ -1900,15 +1900,19 @@ def to_bodo(self) -> bd.DataFrame:
 
         return bd.read_iceberg_table(self)
 
-    def to_polars(self) -> pl.LazyFrame:
+    def to_polars(self, **kwargs: Any) -> pl.LazyFrame:
         """Lazily read from this Apache Iceberg table.
 
+        Args:
+            **kwargs: Additional keyword arguments to forward to :func:`polars.scan_iceberg`
+                (e.g. ``storage_options``, ``snapshot_id``, ``reader_override``).
+
         Returns:
-            pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table
+            pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table.
         """
         import polars as pl
 
-        return pl.scan_iceberg(self)
+        return pl.scan_iceberg(self, **kwargs)
 
     def __datafusion_table_provider__(self, session: Any | None = None) -> IcebergDataFusionTable:
         """Return the DataFusion table provider PyCapsule interface.

tests/table/test_init.py

@@ -2067,3 +2067,29 @@ def _spy(*args: Any, **kwargs: Any) -> FileIO:
 
     assert seen_locations, "expected at least one load_file_io call"
     assert all(loc is not None for loc in seen_locations), f"load_file_io called without a location: {seen_locations}"
+
+
+def test_table_to_polars_forwards_kwargs(table_v2: Table, monkeypatch: pytest.MonkeyPatch) -> None:
+    import polars as pl
+
+    recorded_calls: list[tuple[Any, dict[str, Any]]] = []
+
+    def mock_scan_iceberg(source: Any, **kwargs: Any) -> Any:
+        recorded_calls.append((source, kwargs))
+        return "mock_lazy_frame"
+
+    monkeypatch.setattr(pl, "scan_iceberg", mock_scan_iceberg)
+
+    # Test without kwargs
+    res1 = table_v2.to_polars()
+    assert res1 == "mock_lazy_frame"
+    assert recorded_calls[0] == (table_v2, {})
+
+    # Test with kwargs
+    storage_options = {"s3": {"endpoint": "https://s3.custom.endpoint"}}
+    res2 = table_v2.to_polars(storage_options=storage_options, snapshot_id=12345, reader_override="native")
+    assert res2 == "mock_lazy_frame"
+    assert recorded_calls[1] == (
+        table_v2,
+        {"storage_options": storage_options, "snapshot_id": 12345, "reader_override": "native"},
+    )