Skip to content

Commit 5953815

Browse files
committed
Import Arrow UTF-8 columns directly from buffers
1 parent c3047a4 commit 5953815

4 files changed

Lines changed: 138 additions & 1 deletion

File tree

‎src/blosc2/_utf8_array.py‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -678,6 +678,59 @@ def extend(self, values: Iterable[Any]) -> None:
678678
self._pending_chars += len(v)
679679
self._flush_if_needed()
680680

681+
def extend_arrow(self, array) -> None:
682+
"""Append an Arrow string array directly from offsets and UTF-8 buffers.
683+
684+
No Python strings are materialized. Null slots use the same physical
685+
fill as ``extend``; CTable remains responsible for validity sidecars.
686+
Sliced arrays are rebased, and view-based strings are cast within Arrow.
687+
"""
688+
import pyarrow as pa
689+
import pyarrow.compute as pc
690+
691+
self._check_writable()
692+
if not isinstance(array, (pa.Array, pa.ChunkedArray)):
693+
raise TypeError("extend_arrow requires an Arrow string array")
694+
is_view = getattr(pa.types, "is_string_view", lambda _: False)(array.type)
695+
if not (pa.types.is_string(array.type) or pa.types.is_large_string(array.type) or is_view):
696+
raise TypeError("extend_arrow requires an Arrow string array")
697+
if isinstance(array, pa.ChunkedArray):
698+
for chunk in array.chunks:
699+
self.extend_arrow(chunk)
700+
return
701+
if is_view:
702+
array = array.cast(pa.large_string())
703+
array.validate(full=True)
704+
if not len(array):
705+
return
706+
if array.null_count:
707+
array = pc.fill_null(array, self._coerce(None))
708+
_, offsets_buffer, data_buffer = array.buffers()
709+
dtype = np.dtype("<i8" if pa.types.is_large_string(array.type) else "<i4")
710+
offsets = np.frombuffer(
711+
offsets_buffer, dtype=dtype, count=len(array) + 1, offset=array.offset * dtype.itemsize
712+
)
713+
first, last = int(offsets[0]), int(offsets[-1])
714+
data = (
715+
np.frombuffer(data_buffer, dtype=np.uint8, count=last - first, offset=first)
716+
if last > first
717+
else np.empty(0, dtype=np.uint8)
718+
)
719+
# Preserve the ordering of any earlier Python-string appends.
720+
self.flush()
721+
start = self._bytes_used
722+
rows = self._persisted_rows + len(array)
723+
used = start + len(data)
724+
self._data.resize((max(used, 1),))
725+
if len(data):
726+
self._data[start:used] = data
727+
self._offsets.resize((rows + 1,))
728+
self._offsets[self._persisted_rows + 1 : rows + 1] = (
729+
offsets[1:].astype(np.int64, copy=False) - first + start
730+
)
731+
self._persisted_rows = rows
732+
self._bytes_used_cache = used
733+
681734
def flush(self) -> None:
682735
"""Write pending rows to the backing offsets/data NDArrays."""
683736
self._check_open()

‎src/blosc2/ctable.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9392,7 +9392,10 @@ def _write_arrow_batch(
93929392
obj._ensure_null_mask(col.name)[pos : pos + m] = arrow_col.is_valid().to_numpy(
93939393
zero_copy_only=False
93949394
)
9395-
new_cols[col.name].extend(arrow_col.to_pylist())
9395+
if cls._is_utf8_column(col):
9396+
new_cols[col.name].extend_arrow(arrow_col)
9397+
else:
9398+
new_cols[col.name].extend(arrow_col.to_pylist())
93969399
elif cls._is_dictionary_column(col):
93979400
import pyarrow as _pa
93989401

‎tests/ctable/test_arrow_interop.py‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,32 @@
1919
pa = pytest.importorskip("pyarrow")
2020

2121

22+
@pytest.mark.parametrize("arrow_kind", ["string", "large_string", "string_view"])
23+
@pytest.mark.parametrize("null_storage", ["mask", "sentinel"])
24+
@pytest.mark.parametrize("persistent", [False, True])
25+
def test_arrow_utf8_import_avoids_python_strings(
26+
arrow_kind, null_storage, persistent, tmp_path, monkeypatch
27+
):
28+
from blosc2._utf8_array import UTF8Array
29+
30+
if not hasattr(pa, arrow_kind):
31+
pytest.skip(f"Installed PyArrow has no {arrow_kind}")
32+
arrow_type = getattr(pa, arrow_kind)()
33+
values = ["ignored", "café", "", None, "日本語", "emoji 🎉", "a\0b", None, "ignored"]
34+
array = pa.array(values, type=arrow_type).slice(1, 7)
35+
source = pa.table({"name": array})
36+
monkeypatch.setattr(UTF8Array, "extend", lambda *_: pytest.fail("Python-string ingestion used"))
37+
options = {"urlpath": str(tmp_path / "utf8.b2z")} if persistent else {}
38+
with CTable.from_arrow(
39+
source.schema, iter(source.to_batches(max_chunksize=3)), null_storage=null_storage, **options
40+
) as table:
41+
assert table.to_arrow().column("name").to_pylist() == values[1:8]
42+
assert table["name"].null_count() == 2
43+
if persistent:
44+
with CTable.open(options["urlpath"]) as table:
45+
assert table.to_arrow().column("name").to_pylist() == values[1:8]
46+
47+
2248
@pytest.mark.parametrize("first_rows", [0, 1, 7])
2349
@pytest.mark.parametrize("producer", ["schema", "reader"])
2450
def test_unknown_arrow_stream_grid_is_independent_of_first_batch(first_rows, producer, monkeypatch):

‎tests/ctable/test_utf8.py‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,61 @@ def make_table(values=None, **kwargs):
5454
return CTable(Row, new_data={"name": list(values), "x": list(range(len(values)))}, **kwargs)
5555

5656

57+
@pytest.mark.parametrize("large", [False, True])
58+
@pytest.mark.parametrize("null_storage", ["mask", "sentinel"])
59+
def test_utf8_extend_arrow_buffers_and_pending_order(large, null_storage):
60+
pa = pytest.importorskip("pyarrow")
61+
from blosc2._utf8_array import UTF8Array
62+
63+
options = {"null_value": "<NULL>"} if null_storage == "sentinel" else {}
64+
backend = UTF8Array(blosc2.utf8(nullable=True, null_storage=null_storage, **options))
65+
backend.extend(["pending"])
66+
values = ["ignored", "café", "", None, "日本語", "a\0b", "ignored"]
67+
array = pa.array(values, type=pa.large_string() if large else pa.string()).slice(1, 5)
68+
backend.extend_arrow(array)
69+
backend.extend_arrow(pa.chunked_array([array.slice(0, 2), array.slice(2)]))
70+
backend.append("last")
71+
backend.flush()
72+
expected = ["pending"] + [backend._coerce(value) for value in values[1:6]] * 2 + ["last"]
73+
assert backend[:].tolist() == expected
74+
assert len(backend) == len(expected)
75+
empty = pa.array([], type=array.type)
76+
backend.extend_arrow(empty)
77+
assert backend[:].tolist() == expected
78+
79+
80+
@pytest.mark.parametrize("values", [[], ["", ""], [None, None]])
81+
@pytest.mark.parametrize("large", [False, True])
82+
def test_utf8_extend_arrow_empty_byte_spans(values, large):
83+
pa = pytest.importorskip("pyarrow")
84+
from blosc2._utf8_array import UTF8Array
85+
86+
backend = UTF8Array(blosc2.utf8(nullable=True, null_storage="mask"))
87+
arrow_type = pa.large_string() if large else pa.string()
88+
array = pa.array(["prefix", *values, "suffix"], type=arrow_type).slice(1, len(values))
89+
backend.extend_arrow(array)
90+
backend.extend_arrow(pa.array(["tail"], type=arrow_type))
91+
assert backend[:].tolist() == [""] * len(values) + ["tail"]
92+
93+
94+
def test_utf8_extend_arrow_rejects_wrong_types_and_invalid_utf8():
95+
pa = pytest.importorskip("pyarrow")
96+
from blosc2._utf8_array import UTF8Array
97+
98+
backend = UTF8Array(blosc2.utf8())
99+
for wrong in (["hello"], pa.array([1]), pa.chunked_array([], type=pa.int64())):
100+
with pytest.raises(TypeError, match="Arrow string array"):
101+
backend.extend_arrow(wrong)
102+
with pytest.raises(TypeError, match="not nullable"):
103+
backend.extend_arrow(pa.array([None], type=pa.string()))
104+
invalid = pa.Array.from_buffers(
105+
pa.string(), 1, [None, pa.py_buffer(np.array([0, 1], dtype="<i4")), pa.py_buffer(b"\xff")]
106+
)
107+
with pytest.raises(pa.ArrowInvalid):
108+
backend.extend_arrow(invalid)
109+
assert len(backend) == 0
110+
111+
57112
# ---------------------------------------------------------------------------
58113
# Schema spec
59114
# ---------------------------------------------------------------------------

0 commit comments

Comments
 (0)