apache / apache/arrow-julia

`Arrow.Table` uses last dictionary for all batches in streaming decode, resulting in incorrect reads

Open
#610 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Julia
Stars
312
Forks
78
PR merge metrics
No merged PRs in 30d

Description

This issue was primarily written by AI, but checked by a person. The practical upshot is that data is read back differently than it was written.

Example:

Here's a 2-batch stream where the second `DictionaryBatch` replaces (not deltas) the dictionary for column a: batch 1 uses ["alpha","beta"], batch 2 uses ["gamma","delta"]. Reading it back with `Arrow.Table` gives `["gamma","delta","gamma","delta"]` instead of `["alpha","beta","gamma","delta"]`; `Arrow.Stream` on the same bytes gives the correct answer.

```julia
using Arrow

# Two independent single-batch streams, same schema, different dict pools.
# Each is written fresh, so each DictionaryBatch is "initial" (isDelta=false) —
# exactly what a producer like pyarrow emits when per-batch pools differ.
open(io -> Arrow.write(io, (a=Arrow.DictEncode(["alpha", "beta"]),)), "s1.arrows", "w")
open(io -> Arrow.write(io, (a=Arrow.DictEncode(["gamma", "delta"]),)), "s2.arrows", "w")

# Splice: [schema, dict1, rb1] from s1 + [dict2, rb2] from s2 + EOS.
# This reproduces a foreign stream whose 2nd DictionaryBatch replaces dict id 0.
function messages(bytes)
spans, pos = UnitRange{Int}[], 1
while pos + 3 <= length(bytes) && Arrow.readbuffer(bytes, pos, UInt32) == Arrow.CONTINUATION_INDICATOR_BYTES
pos += 4
msglen = Arrow.readbuffer(bytes, pos, Int32)
msglen == 0 && break
pos += 4
msg = Arrow.FlatBuffers.getrootas(Arrow.Meta.Message, bytes, pos - 1)
start = pos - 8
pos += msglen + msg.bodyLength
push!(spans, start:(pos - 1))
end
spans
end

b1, b2 = read("s1.arrows"), read("s2.arrows")
s1, s2 = messages(b1), messages(b2)
eos = reinterpret(UInt8, [Arrow.CONTINUATION_INDICATOR_BYTES, UInt32(0)])
write("combined.arrows", vcat(b1[s1[1]], b1[s1[2]], b1[s1[3]], b2[s2[2]], b2[s2[3]], eos))

println(collect(Arrow.Table("combined.arrows").a)) # ["gamma","delta","gamma","delta"] -- WRONG
println([collect(b.a) for b in Arrow.Stream("combined.arrows")]) # [["alpha","beta"],["gamma","delta"]] -- correct
```

The `pyarrow` version behaves correctly in this case:

```python
import pyarrow as pa, pyarrow.ipc as ipc
b1 = pa.record_batch({'a': pa.DictionaryArray.from_arrays(
pa.array([0, 1], pa.int32()), pa.array(['alpha', 'beta']))})
b2 = pa.record_batch({'a': pa.DictionaryArray.from_arrays(
pa.array([0, 1], pa.int32()), pa.array(['gamma', 'delta']))})
with ipc.new_stream('r2.arrows', b1.schema) as w:
w.write_batch(b1); w.write_batch(b2)

ipc.open_stream('r2.arrows').read_all().column(0).to_pylist()
# ['alpha', 'beta', 'gamma', 'delta']
```

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by comparing the streaming decode paths used by Arrow.Table and Arrow.Stream with the two-batch reproducer in the issue. Trace how each DictionaryBatch is applied while reading combined.arrows, then verify that the table result preserves ["alpha", "beta", "gamma", "delta"] like Arrow.Stream and pyarrow.

Written by the indexing model from the issue text.

Assessment

Tech stack
julia
Domain
data
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.