This is an automated email from the ASF dual-hosted git repository. quinnj pushed a commit to branch preservation/storage-20261008/julia-dev-Arrow-b25a15d2-saved-b95ccc2bf5 in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
commit 8491d44e40e1884608dec33f5887405478aac6dd Author: Jacob Quinn <[email protected]> AuthorDate: Mon Sep 7 18:14:39 2026 -0600 perf: fast-path covered view buffers in IPC output Prove that valid entries cover each copied buffer prefix, then copy the prefix once. Keep entry and validity sanitation and fall back to range compaction for gaps or unused buffers. Cache validity lookup and load each entry once. --- src/ipc_write.jl | 31 +++++++-- test/ipc_view_output_tests.jl | 143 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 5 deletions(-) diff --git a/src/ipc_write.jl b/src/ipc_write.jl index 0d27942..1d6f69b 100644 --- a/src/ipc_write.jl +++ b/src/ipc_write.jl @@ -405,26 +405,47 @@ function _encodeviewbuffers!(c::EncodeCursor, d::ArrayData) # In-memory views may borrow a whole parser input or a larger column. # IPC must not persist those unrelated bytes, null entries, or padding. views = AC.rolebuffer(d, AC.VIEWS) + inputvalidity = AC.validitybuffer(d) entries = zeros(UInt8, AC.checked_mul(d.len, Int64(16))) payloads = reinterpret(UInt128, entries) words = reinterpret(Int32, entries) validity = AC.nullcount(d) == 0 ? UInt8[] : zeros(UInt8, cld(d.len, 8)) - refs = Int64[] # rows with non-inline values + # Prove that valid long entries cover a prefix of every source buffer. + # Such columns need neither row sorting nor per-entry data copies. + coverage = zeros(Int64, length(d.buffers) - 2) + compact = true for i = 1:d.len - AC.isvalid_at(d, i) || continue + (AC.isempty_buffer(inputvalidity) || AC.getbit(inputvalidity, i - 1)) || continue isempty(validity) || (validity[(i - 1) ÷ 8 + 1] |= UInt8(1) << ((i - 1) % 8)) base = 16 * (i - 1) - len = AC.loadat(views, Int32, base) payload = AC.loadat(views, UInt128, base) + len = reinterpret(Int32, payload % UInt32) # The writer requires little-endian storage. Mask inline padding; # long entries keep their source coordinates until remapped below. payloads[i] = len <= AC.VIEW_INLINE_MAX ? payload & (typemax(UInt128) >> (8 * (12 - len))) : payload - if len > AC.VIEW_INLINE_MAX - push!(refs, i) + if compact && len > AC.VIEW_INLINE_MAX + bufidx, off = words[4 * i - 1], words[4 * i] + extent = coverage[bufidx + 1] + if off > extent + compact = false + else + coverage[bufidx + 1] = max(extent, Int64(off) + len) + end + end + end + if compact && all(>(0), coverage) + encodebuffer!(c, validity) + encodebuffer!(c, entries) + push!(c.variadics, Int64(length(coverage))) + for (i, extent) in enumerate(coverage) + # Only the proven prefix is copied; trailing bytes are unused. + encodebuffer!(c, AC.slicebytes(AC.subslice(d.buffers[i + 2], 0, extent))) end + return nothing end + refs = Int64[i for i = 1:d.len if words[4 * i - 3] > AC.VIEW_INLINE_MAX] # Sort by source position so duplicate and overlapping views share the # same output bytes. Already ordered producer columns need no sort. sourcepos(i) = (words[4 * i - 1], words[4 * i]) diff --git a/test/ipc_view_output_tests.jl b/test/ipc_view_output_tests.jl index 586062f..2ea12ba 100644 --- a/test/ipc_view_output_tests.jl +++ b/test/ipc_view_output_tests.jl @@ -42,6 +42,24 @@ function fixture() return StringVector{Union{Missing,ArrowString}}(payloads, buffers) end +function packedfixture() + first, second = "first selected long string", "second selected long string" + raw = collect(codeunits(first * second * UNUSED)) + p = ArrowStrings.view_payload(raw, 1, ncodeunits(first), 0, 0) + q = ArrowStrings.view_payload( + raw, + ncodeunits(first) + 1, + ncodeunits(second), + 0, + ncodeunits(first), + ) + inline = ArrowStrings.inline_payload(codeunits("ok"), 1, 2) + return StringVector{Union{Missing,ArrowString}}( + [p, q, p, ArrowStrings.PAYLOAD_MISSING, inline], + [raw], + ) +end + function checkoutput(bytes, expected, ndata; file, compress) @test isequal(Arrow.Table(bytes).s, expected) decoded = file ? Arrow.readfile(bytes) : Arrow.readstream(bytes) @@ -207,6 +225,131 @@ end end end + @testset "covered prefixes and gaps preserve the byte boundary" begin + # Include interleaved buffers, contained/extended overlaps, a late + # gap, a gap filled by a later row, and a wholly unused buffer. + cases = [ + [(0, 0, 20), (1, 0, 18), (0, 8, 24), (1, 10, 22), (0, 0, 13), (0, 32, 32)], + [(0, 0, 20), (1, 0, 18), (0, 32, 32)], + [(0, 0, 20), (1, 0, 18), (0, 32, 32), (0, 20, 20)], + [(1, 0, 18), (1, 10, 22)], + ] + for ranges in cases, utf8 in (true, false) + buffers = [ + vcat(repeat(collect(codeunits("abcdefgh")), 8), codeunits(UNUSED)), + vcat(repeat(collect(codeunits("ijklmnop")), 8), codeunits(UNUSED)), + ] + utf8 || (buffers[1][1] = 0xff) + used = [falses(length(b)) for b in buffers] + payloads = ArrowStringPayload[] + expected = Any[] + for (bufidx, off, len) in ranges + raw = buffers[bufidx + 1] + push!(payloads, ArrowStrings.view_payload(raw, off + 1, len, bufidx, off)) + used[bufidx + 1][(off + 1):(off + len)] .= true + value = raw[(off + 1):(off + len)] + push!(expected, utf8 ? String(value) : value) + end + # A null entry contains a valid-looking reference to unused + # bytes. It must neither extend coverage nor survive as bytes. + push!( + payloads, + ArrowStrings.view_payload(buffers[1], 65, ncodeunits(UNUSED), 0, 64), + ) + push!(expected, missing) + push!(payloads, ArrowStrings.inline_payload(codeunits("ok"), 1, 2)) + push!(expected, utf8 ? "ok" : collect(codeunits("ok"))) + n = length(payloads) + entries = collect(reinterpret(UInt8, payloads)) + entries[(16 * (n - 1) + 7):(16 * n)] .= 0x5a + append!(entries, codeunits("TRAILING_UNUSED_ENTRIES")) + # Exercise an unaligned entry buffer and a nonzero slice base + # for data. BufferSlice bounds, not owner allocation size, apply. + borrowed = vcat(UInt8[0xaa], entries) + data = [ + AC.subslice(AC._databuffer(vcat(UInt8[0xaa], b)), 1, length(b)) for + b in buffers + ] + validity = fill(0xff, cld(n, 8) + 1) + validity[(n - 2) ÷ 8 + 1] &= ~(UInt8(1) << ((n - 2) % 8)) + t = AC.ViewType(utf8) + f = AC.Field("s", t; nullable=true) + d = AC.ArrayData( + t, + n, + [ + AC._databuffer(validity), + AC.subslice(AC._databuffer(borrowed), 1, length(entries)), + data..., + ]; + nullcount=1, + ) + sch = AC.Schema([f]) + batch = AC.RecordBatch(sch, [d]) + original = [AC.slicebytes(b) for b in d.buffers] + for file in (true, false), compress in (:none, :lz4, :zstd) + bytes = + file ? Arrow.writefile(sch, [batch]; compress=compress) : + Arrow.writestream(sch, [batch]; compress=compress) + decoded = file ? Arrow.readfile(bytes) : Arrow.readstream(bytes) + out = (file ? decoded[1] : decoded.batches[1]).columns[1] + @test AC.validate_full(f, out) === out + @test isequal(Arrow.Table(bytes).s, expected) + @test [AC.slicebytes(b) for b in out.buffers[3:end]] == [buffers[i][used[i]] for i = 1:2 if any(used[i])] + outentries = AC.slicebytes(out.buffers[2]) + @test length(outentries) == 16 * n + @test all(iszero, outentries[(16 * (n - 2) + 1):(16 * (n - 1))]) + @test all(iszero, outentries[(16 * (n - 1) + 7):(16 * n)]) + @test all( + b -> findfirst(codeunits(UNUSED), AC.slicebytes(b)) === nothing, + out.buffers, + ) + @test isequal([AC.isvalid_at(out, i) for i = 1:n], [i != n - 1 for i = 1:n]) + end + @test [AC.slicebytes(b) for b in d.buffers] == original + end + end + + @testset "covered dictionary pools retain unreferenced values" begin + col = packedfixture() + f, parts = Arrow._constructcolumn(:s, AbstractVector[col]) + pool = only(parts) + t = AC.DictionaryType(AC.IntType(8, true), f.type, true) + df = AC.Field("s", t; nullable=false) + indices = Int8[0, 0, 4] + d = AC.ArrayData( + t, + 3, + [AC.BufferSlice(), AC._databuffer(indices)]; + dictionary=pool, + nullcount=0, + ) + sch = AC.Schema([df]) + expectedpool = [ + "first selected long string", + "second selected long string", + "first selected long string", + missing, + "ok", + ] + for file in (true, false), compress in (:none, :lz4, :zstd) + bytes = + file ? Arrow.writefile(sch, [AC.RecordBatch(sch, [d])]; compress=compress) : + Arrow.writestream(sch, [AC.RecordBatch(sch, [d])]; compress=compress) + decoded = file ? Arrow.readfile(bytes) : Arrow.readstream(bytes) + out = (file ? decoded[1] : decoded.batches[1]).columns[1] + @test Arrow.Table(bytes).s == + ["first selected long string", "first selected long string", "ok"] + @test decoded.schema.fields[1].type == t + @test AC.slicebytes(out.buffers[2]) == reinterpret(UInt8, indices) + @test out.dictionary.len == 5 + @test isequal(AC.materialize(f, out.dictionary), expectedpool) + @test length(out.dictionary.buffers) == 3 + @test AC.slicebytes(out.dictionary.buffers[3]) == + codeunits(expectedpool[1] * expectedpool[2]) + end + end + @testset "dictionary pools preserve values and indices" begin col = fixture() f, parts = Arrow._constructcolumn(:s, AbstractVector[col])
