Feature: dictionary_columns for IncrementalAppendScan

Expose the dictionary_columns Arrow-read option (added to DataScan in
#3461) on IncrementalAppendScan.to_arrow / to_arrow_batch_reader. The
shared _to_arrow*_via_file_scan_tasks helpers already thread it, so this
just forwards the argument. Follow-up to #3512.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: smaheshwar-pltr

pyiceberg/table/__init__.py

@@ -2466,28 +2466,46 @@ def plan_files(self) -> Iterable[FileScanTask]:
             and manifest_entry.status == ManifestEntryStatus.ADDED,
         )
 
-    def to_arrow(self) -> pa.Table:
+    def to_arrow(self, dictionary_columns: tuple[str, ...] = ()) -> pa.Table:
         """Read an Arrow table eagerly from this IncrementalAppendScan.
 
         All rows will be loaded into memory at once.
 
+        Args:
+            dictionary_columns:
+                A tuple of column names that PyArrow should read as
+                dictionary-encoded (``pa.DictionaryArray``).  Dictionary
+                encoding can substantially reduce memory usage for columns
+                with low-cardinality repeated string values.
+                Only applies to Parquet files; silently ignored for ORC.
+
         Returns:
             pa.Table: Materialized Arrow Table from the Iceberg table's IncrementalAppendScan
         """
-        return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files())
+        return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns)
 
-    def to_arrow_batch_reader(self) -> pa.RecordBatchReader:
+    def to_arrow_batch_reader(self, dictionary_columns: tuple[str, ...] = ()) -> pa.RecordBatchReader:
         """Return an Arrow RecordBatchReader from this IncrementalAppendScan.
 
         For large results, using a RecordBatchReader requires less memory than
         loading an Arrow Table for the same IncrementalAppendScan, because a
         RecordBatch is read one at a time.
 
+        Args:
+            dictionary_columns:
+                A tuple of column names that PyArrow should read as
+                dictionary-encoded (``pa.DictionaryArray``).  Dictionary
+                encoding can substantially reduce memory usage for columns
+                with low-cardinality repeated string values.
+                Only applies to Parquet files; silently ignored for ORC.
+
         Returns:
             pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's IncrementalAppendScan
                 which can be used to read a stream of record batches one by one.
         """
-        return _to_arrow_batch_reader_via_file_scan_tasks(self, self.projection(), self.plan_files())
+        return _to_arrow_batch_reader_via_file_scan_tasks(
+            self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns
+        )
 
     def _validate_and_resolve_snapshots(self) -> tuple[int, int]:
         if self.from_snapshot_id_exclusive is None:

tests/integration/test_reads.py

@@ -1295,6 +1295,24 @@ def test_incremental_append_scan_append_only(catalog: Catalog) -> None:
     assert len(scan.to_polars()) == 3
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog_hive"), lf("session_catalog")])
+def test_incremental_append_scan_dictionary_columns(catalog: Catalog) -> None:
+    test_table = catalog.load_table("default.test_incremental_read")
+
+    scan = (
+        test_table.incremental_append_scan()
+        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
+        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    )
+    result = scan.to_arrow(dictionary_columns=("letter",))
+
+    # The requested column is dictionary-encoded; others stay plain.
+    assert pa.types.is_dictionary(result.schema.field("letter").type)
+    assert not pa.types.is_dictionary(result.schema.field("number").type)
+    assert sorted(result["number"].to_pylist()) == [2, 3, 4]
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [lf("session_catalog_hive"), lf("session_catalog")])
 def test_incremental_append_scan_ignores_non_append_snapshots(catalog: Catalog) -> None: