Re: [I] How to load a table with bucket partition [iceberg-python]

["jaimeferj (via GitHub)" <gi...@apache.org>]

jaimeferj commented on issue #548:
URL: https://github.com/apache/iceberg-python/issues/548#issuecomment-3411533044

   Hi there! I'm encountering a similar use case and would like to revive this discussion.
   
   ## Use Case: Efficient Partition-Aware Joins in Polars
   
   I have multiple large Iceberg tables that are all partitioned identically using `bucket(user_id, N)`. I need to perform joins between these tables using Polars, but currently, I cannot load all tables into memory at once.
   
   **What I want to achieve:**
   - Join tables partition-by-partition to avoid loading all data into memory
   - Leverage the fact that both tables share the same bucketing scheme on the join key (`user_id`)
   - Process each bucket independently, which should be much more memory-efficient
   
   **Current workaround:**
   
   Based on @goalzz85's solution above, I can manually filter files by partition:
   
   ```python
   import polars as pl
   from collections import defaultdict
   
   # Group files by partition value
   def get_files_by_partition(table):
       files_by_partition = defaultdict(list)
       for task in table.scan().plan_files():
           partition_key = tuple(task.file.partition) if task.file.partition else ()
           files_by_partition[partition_key].append(task.file.file_path)
       return files_by_partition
   
   # Get partitioned files
   files1_by_partition = get_files_by_partition(table1)
   files2_by_partition = get_files_by_partition(table2)
   
   # Process partition by partition
   results = []
   for partition_key in files1_by_partition.keys():
       if partition_key in files2_by_partition:
           lf1 = pl.scan_parquet(files1_by_partition[partition_key])
           lf2 = pl.scan_parquet(files2_by_partition[partition_key])
           result = lf1.join(lf2, on="user_id").collect()
           results.append(result)
   
   final = pl.concat(results)
   ```
   
   **The problem:**
   This workaround requires manually extracting file paths and bypassing PyIceberg's scan interface entirely. It would be much cleaner to have native support for partition-aware scans.
   
   ## Feature Request
   
   It would be extremely valuable to have one of the following:
   
   1. **Expose partition values in row filters**: Allow filtering by hidden partition columns, e.g.:
      ```python
      table.scan(row_filter=PartitionEquals("user_id_bucket", 5))
      ```
   
   2. **Add a method to scan specific partitions directly**:
      ```python
      table.scan(partition_filter={"user_id_bucket": 5}).to_polars()
      ```
   
   3. **Add an iterator over partitions**:
      ```python
      for partition_value, scan in table.scan_partitions():
          df = scan.to_polars()
          # process partition
      ```
   
   This would enable memory-efficient partition-aware joins and other partition-level processing patterns that are common in distributed systems like Spark, but currently difficult to implement efficiently in PyIceberg + Polars.
   
   Would love to hear thoughts on this! Happy to contribute if there's interest in adding this functionality.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org