This is an automated email from the ASF dual-hosted git repository.
quinnj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
The following commit(s) were added to refs/heads/main by this push:
new 3799e71 GH-610: Preserve dictionary replacements when reading tables
(#611)
3799e71 is described below
commit 3799e717cac12e6891d55d41fb004cf4167de216
Author: Jacob Quinn <[email protected]>
AuthorDate: Tue Oct 27 10:50:22 2026 -0600
GH-610: Preserve dictionary replacements when reading tables (#611)
Addresses #610. A dictionary replacement could overtake queued
record-batch tasks, causing earlier rows to use later dictionary values.
Wait for outstanding batch construction before each dictionary update,
outside the dictionary lock, and clear completed tasks. Use the same
chained dictionary storage as Arrow.Stream so later deltas keep column
types stable.
Dictionary updates become synchronization points; consecutive record
batches with unchanged dictionaries still run in parallel. Dictionary
payloads are not copied. This fixes the current 2.8 reader; the separate
3.0 rewrite in #609 already handles replacement through its own decoder.
Validation: the new regression failed six checks on unchanged main and
passes all 12 after the fix on one and four threads. It covers
replacement followed by delta, compressed input, and multiple input
buffers. Frozen PyArrow V4/V5 fixtures also return correct values. A
channel-gated probe verifies four unchanged-dictionary batch tasks start
before any completes. Full suites passed 66,207 assertions each on Julia
1.9.4 and 1.12.6 using CI's local ArrowTypes setup. Formatting checks
passed.
Co-authored by Codex
---
src/table.jl | 21 ++++++---------------
test/runtests.jl | 24 ++++++++++++++++++++++++
2 files changed, 30 insertions(+), 15 deletions(-)
diff --git a/src/table.jl b/src/table.jl
index de8bfc3..606f2ab 100644
--- a/src/table.jl
+++ b/src/table.jl
@@ -502,6 +502,9 @@ function Table(blobs::Vector{ArrowBlob}; convert::Bool=true)
)
end
elseif header isa Meta.DictionaryBatch
+ # Earlier record batches must capture their dictionaries
before an update.
+ _waitall(tasks)
+ empty!(tasks)
id = header.id
recordbatch = header.data
@debug "parsing dictionary batch message: id = $id,
compression = $(recordbatch.compression)"
@@ -522,20 +525,7 @@ function Table(blobs::Vector{ArrowBlob};
convert::Bool=true)
convert,
)
dictencoding = dictencodings[id]
- if typeof(dictencoding.data) <: ChainedVector
- append!(dictencoding.data, values)
- else
- A = ChainedVector([dictencoding.data, values])
- S =
- field.dictionary.indexType === nothing ? Int32
:
- juliaeltype(field, field.dictionary.indexType,
false)
- dictencodings[id] =
DictEncoding{eltype(A),S,typeof(A)}(
- id,
- A,
- field.dictionary.isOrdered,
- values.metadata,
- )
- end
+ append!(dictencoding.data, values)
continue
end
# new dictencoding or replace
@@ -551,7 +541,8 @@ function Table(blobs::Vector{ArrowBlob}; convert::Bool=true)
Int64(1),
convert,
)
- A = values
+ # Keep the dictionary storage type stable when later
deltas arrive.
+ A = ChainedVector([values])
S =
field.dictionary.indexType === nothing ? Int32 :
juliaeltype(field, field.dictionary.indexType, false)
diff --git a/test/runtests.jl b/test/runtests.jl
index 315d1b6..4f7d28b 100644
--- a/test/runtests.jl
+++ b/test/runtests.jl
@@ -149,6 +149,30 @@ end
@test isequal(collect(str)[2].col1, [missing, 11])
end
+ @testset "# 610: dictionary replacement between record batches" begin
+ for compress in (nothing, :lz4, :zstd)
+ first = (a=Arrow.DictEncode(["alpha", "beta"]),)
+ second = (a=Arrow.DictEncode(["gamma", "delta"]),)
+ third = (a=Arrow.DictEncode(["gamma", "epsilon"]),)
+ b1 = take!(Arrow.tobuffer(Tables.partitioner((first, first));
compress))
+ b2 = take!(Arrow.tobuffer(Tables.partitioner((second, third));
compress))
+
+ # Replace the first stream's EOS with the second stream after
its schema.
+ _, (pos, _) = iterate(Arrow.BatchIterator(Arrow.ArrowBlob(b2,
1, nothing)))
+ bytes = vcat(b1[1:(end - 8)], b2[pos:end])
+ batches = Arrow.BatchIterator(Arrow.ArrowBlob(bytes, 1,
nothing))
+ @test count(batches) do batch
+ header = batch.msg.header
+ header isa Arrow.Meta.DictionaryBatch && header.isDelta
+ end == 1
+
+ expected = [first.a.data, first.a.data, second.a.data,
third.a.data]
+ @test Arrow.Table(bytes).a == vcat(expected...)
+ @test Arrow.Table([b1, b2]).a == vcat(expected...)
+ @test [collect(batch.a) for batch in Arrow.Stream(bytes)] ==
expected
+ end
+ end
+
@testset "# dictionary batch isDelta" begin
t = (
col1=Int64[1, 2, 3, 4],
