Skip to content

GH-610: Preserve dictionary replacements when reading tables #611

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
quinnj merged 1 commit into from
Oct 27, 2026
Merged

GH-610: Preserve dictionary replacements when reading tables #611

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 6 additions & 15 deletions src/table.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)"
Expand All @@ -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
Expand All @@ -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)
Expand Down
24 changes: 24 additions & 0 deletions test/runtests.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down
Loading