You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "tom-at-rewbi (via GitHub)" <gi...@apache.org> on 2025/10/31 23:21:32 UTC
[PR] do not materialize entire files in `to_record_batches` [iceberg-python]
tom-at-rewbi opened a new pull request, #2676:
URL: https://github.com/apache/iceberg-python/pull/2676
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace ${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
# Rationale for this change
My expectation for `to_record_batches` was that it would yield batches and not materialize an entire parquet file in memory, but it looks like the current implementation explicitly does this.
This change makes `to_record_batches` iterate batches lazily.
## Are these changes tested?
I ran `make test` and all tests completed successfully except those that import kerberos (2 of them), which I can't seem to build at the moment.
As for testing that this reduces memory usage, this change made my data pipeline stop OOM'ing.
## Are there any user-facing changes?
There should not be any user-facing changes.
<!-- In the case of user-facing changes, please add the changelog label. -->
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#issuecomment-4022375236
Current main:
<img width="2240" height="800" alt="image" src="https://github.com/user-attachments/assets/d531d338-16c5-4603-a787-9b67403b03ec" />
This branch:
<img width="2240" height="800" alt="image" src="https://github.com/user-attachments/assets/6ea89e86-0c3c-4dd9-b6cf-59440628a7f8" />
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "tom-at-rewbi (via GitHub)" <gi...@apache.org>.
tom-at-rewbi commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487391312
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
If I'm not mistaken, reading is still parallel after this change. Every time a scan task is processed, `pyarrow.dataset.Scanner.from_fragment` is called with the default arguments for `use_threads`, which has a default of maximum parallelism. [(link to source)](https://github.com/tom-at-rewbi/iceberg-python/blob/c8b76875bad3b81dc8e9b4ee9ec9c459b3babd8a/pyiceberg/io/pyarrow.py#L1557)
Are there any situations where we would expect stacking these two forms of parallelism to be beneficial?
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #2676: do not materialize entire files in `to_record_batches`
URL: https://github.com/apache/iceberg-python/pull/2676
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#issuecomment-4256587528
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.
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#issuecomment-4022252823
I've fixed the conflicts because of the renewed interest by https://github.com/apache/iceberg-python/discussions/3122
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "tom-at-rewbi (via GitHub)" <gi...@apache.org>.
tom-at-rewbi commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487391312
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
If I'm not mistaken, reading is still parallel after this change, albeit in a different way. Every time a scan task is processed, `pyarrow.dataset.Scanner.from_fragment` is called with the default arguments for `use_threads`, which has a default of maximum parallelism. [(link to source)](https://github.com/tom-at-rewbi/iceberg-python/blob/c8b76875bad3b81dc8e9b4ee9ec9c459b3babd8a/pyiceberg/io/pyarrow.py#L1557)
pyarrow.dataset.Scanner also has configurable options for readahead (also enabled by default), which I _assume_ is parallelized, but I am not sure.
Are there any situations where we would expect stacking these two forms of parallelism to be beneficial?
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "tom-at-rewbi (via GitHub)" <gi...@apache.org>.
tom-at-rewbi commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487391312
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
If I'm not mistaken, reading is still parallel after this change, albeit in a different way. Every time a scan task is processed, `pyarrow.dataset.Scanner.from_fragment` is called with the default arguments for `use_threads`, which has a default of maximum parallelism. [(link to source)](https://github.com/tom-at-rewbi/iceberg-python/blob/c8b76875bad3b81dc8e9b4ee9ec9c459b3babd8a/pyiceberg/io/pyarrow.py#L1557)
pyarrow.dataset.Scanner also has configurable options for readahead (also enabled by default), which I _assume_ are parallelized, but I am not sure.
Are there any situations where we would expect stacking these two forms of parallelism to be beneficial?
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "tom-at-rewbi (via GitHub)" <gi...@apache.org>.
tom-at-rewbi commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487391312
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
If I'm not mistaken, reading is still parallel after this change, albeit in a different way. Every time a scan task is processed, `pyarrow.dataset.Scanner.from_fragment` is called with the default arguments for `use_threads`, which has a default of maximum parallelism. [(link to source)](https://github.com/tom-at-rewbi/iceberg-python/blob/c8b76875bad3b81dc8e9b4ee9ec9c459b3babd8a/pyiceberg/io/pyarrow.py#L1557)
pyarrow.dataset.Scanner also has configurable options for readahead, which I _assume_ is parallelized, but I am not sure.
Are there any situations where we would expect stacking these two forms of parallelism to be beneficial?
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "tom-at-rewbi (via GitHub)" <gi...@apache.org>.
tom-at-rewbi commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487391312
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
If I'm not mistaken, reading is still parallel after this change, albeit in a different way. Every time a scan task is processed, `pyarrow.dataset.Scanner.from_fragment` is called with the default arguments for `use_threads`, which has a default of maximum parallelism. [(link to source)](https://github.com/tom-at-rewbi/iceberg-python/blob/c8b76875bad3b81dc8e9b4ee9ec9c459b3babd8a/pyiceberg/io/pyarrow.py#L1557)
Are there any situations where we would expect stacking these two forms of parallelism to be beneficial?
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#discussion_r2487206100
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1726,16 +1726,10 @@ def to_record_batches(self, tasks: Iterable[FileScanTask]) -> Iterator[pa.Record
deletes_per_file = _read_all_delete_files(self._io, tasks)
total_row_count = 0
- executor = ExecutorFactory.get_or_create()
-
- def batches_for_task(task: FileScanTask) -> List[pa.RecordBatch]:
- # Materialize the iterator here to ensure execution happens within the executor.
- # Otherwise, the iterator would be lazily consumed later (in the main thread),
- # defeating the purpose of using executor.map.
- return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
Review Comment:
this logic was added in #1995 to parallelize the work using the executor.
> My expectation for to_record_batches was that it would yield batches and not materialize an entire parquet file in memory, but it looks like the current implementation explicitly does this.
that would be my expectation as well. The current implementation reads an entire parquet file.
BUT I do want to still keep the ability to parallelize reading...
lets see if there's a better way to refactor this
--
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
Re: [PR] do not materialize entire files in `to_record_batches` [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2676:
URL: https://github.com/apache/iceberg-python/pull/2676#issuecomment-4210598667
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.
--
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