rambleraptor commented on code in PR #3447:
URL: https://github.com/apache/iceberg-python/pull/3447#discussion_r3927562342
##########
pyiceberg/io/pyarrow.py:
##########
@@ -3043,3 +3060,23 @@ def _get_field_from_arrow_table(arrow_table: pa.Table,
field_path: str) -> pa.Ar
field_array = arrow_table[path_parts[0]]
# Navigate into the struct using the remaining path parts
return pc.struct_field(field_array, path_parts[1:])
+
+
+def _coerce_arrow_input(df: pa.Table | pa.RecordBatchReader |
ArrowStreamExportable) -> pa.Table | pa.RecordBatchReader:
Review Comment:
We're importing this all over the place. Might as well make it public.
##########
pyiceberg/io/pyarrow.py:
##########
@@ -2690,30 +2690,45 @@ def bin_pack_arrow_table(tbl: pa.Table,
target_file_size: int) -> Iterator[list[
"""Bin-pack ``tbl`` into groups of RecordBatches, each
~``target_file_size``.
Note:
- ``target_file_size`` is measured in **uncompressed in-memory** Arrow
bytes
- (``Table.nbytes`` / ``RecordBatch.nbytes``), not compressed on-disk
Parquet
- bytes. The resulting Parquet file after compression (zstd by default,
- plus dictionary/RLE encoding) is typically 3-10× smaller than
- ``target_file_size``. This is a coarse proxy for the spec-defined
+ ``target_file_size`` is measured in **uncompressed in-memory** Arrow
+ bytes, not compressed on-disk Parquet bytes. The size estimate uses
+ ``nbytes`` when available and falls back to referenced buffer size for
+ Arrow view types that do not support ``nbytes``. The resulting Parquet
+ file after compression (zstd by default, plus dictionary/RLE encoding)
+ is typically 3-10× smaller than ``target_file_size``. This is a coarse
+ proxy for the spec-defined
``write.target-file-size-bytes`` and will be tightened to true on-disk
bytes once the writer is switched to a rolling-``ParquetWriter`` with
``OutputStream.tell()`` (#2998).
"""
from pyiceberg.utils.bin_packing import PackingIterator
- avg_row_size_bytes = tbl.nbytes / tbl.num_rows
+ avg_row_size_bytes = _arrow_data_size(tbl) / tbl.num_rows
target_rows_per_file = max(1, int(target_file_size / avg_row_size_bytes))
batches = tbl.to_batches(max_chunksize=target_rows_per_file)
bin_packed_record_batches = PackingIterator(
items=batches,
target_weight=target_file_size,
lookback=len(batches), # ignore lookback
- weight_func=lambda x: x.nbytes,
+ weight_func=_arrow_data_size,
largest_bin_first=False,
)
return bin_packed_record_batches
+def _arrow_data_size(data: pa.Table | pa.RecordBatch) -> int:
+ """Estimate Arrow data size for writer bin-packing.
+
+ ``nbytes`` is the better logical-size estimate, but PyArrow can raise for
+ view types such as ``string_view`` exported by libraries like Polars. Fall
+ back to total referenced buffer size so those streams can still be written.
+ """
+ try:
+ return data.nbytes
+ except pyarrow.lib.ArrowTypeError:
+ return data.get_total_buffer_size()
Review Comment:
This is an overestimation. Is there a way for us to get more precise?
(I recognize this is just the case for the stream case, so an overestimation
is probably fine. It's not changing the existing code paths.)
##########
pyiceberg/table/__init__.py:
##########
@@ -1778,6 +1777,10 @@ def __datafusion_table_provider__(self, session: Any |
None = None) -> IcebergDa
).__datafusion_table_provider__
return provider(session)
+ def __arrow_c_stream__(self, requested_schema: object | None = None) ->
object:
+ """Export this Table as an Arrow C stream (PyCapsule interface)."""
+ return
self.scan().to_arrow_batch_reader().__arrow_c_stream__(requested_schema)
Review Comment:
This is the part I'm most concerned about and something I'd like the
community's input on.
By turning a Table into a PyCapsule interface, we can now `append` or
`overwrite` a Table with another Table. We explicitly blocked this earlier. Is
this something we're okay enabling?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
