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-fca4d7ca40 in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
commit 64ff77fb9ba290e7db08a2afb6f745ae6ef8067a Author: Jacob Quinn <[email protected]> AuthorDate: Mon Sep 7 16:49:37 2026 -0600 fix: discard inherited statistics when rewriting tables --- docs/src/manual.md | 11 +++-- src/write.jl | 28 +++++++----- test/rewrite_regressions.jl | 104 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 15 deletions(-) diff --git a/docs/src/manual.md b/docs/src/manual.md index aac9875..b22f4df 100644 --- a/docs/src/manual.md +++ b/docs/src/manual.md @@ -295,8 +295,12 @@ Batch pruning uses per-batch statistics (row count, null count, min, max per column) carried in the file's footer schema metadata under the key `JuliaArrow:batch_statistics.v1`, in the value layout of the Arrow project's statistics schema — other readers see ordinary metadata. -[`Arrow.write`](@ref) does not embed them; files that carry them prune, -files that do not are simply scanned batch by batch. +[`Arrow.write`](@ref) does not compute them; files that carry them prune, +files that do not are simply scanned batch by batch. `Arrow.write` and +[`Arrow.Writer`](@ref) drop automatically inherited batch statistics when +rewriting an `Arrow.Table` or `Arrow.Stream`. The source statistics may no +longer describe the output values, column order, or batch boundaries. +An explicit `metadata` override is used as supplied, including any statistics. ### Reading remote and partial files @@ -479,7 +483,8 @@ When the source is an `Arrow.Table` or `Arrow.Stream`, the writer retains the compatible Arrow descriptor tree. Temporal units, byte and list widths, Struct, Map, Run-End Encoding, nullability, field metadata, schema metadata, and top-level dictionary index types and category order survive a read/write -round trip. Buffer sharing, overlapping ListView ranges, and exact run +round trip. Automatically inherited batch statistics are dropped as described +above. Buffer sharing, overlapping ListView ranges, and exact run segmentation are rebuilt into a canonical form without changing logical values. A retained Map that declares sorted keys is rewritten only when each row remains sorted. diff --git a/src/write.jl b/src/write.jl index 9176cca..fee1e0c 100644 --- a/src/write.jl +++ b/src/write.jl @@ -49,6 +49,10 @@ Write any Tables.jl source as Arrow IPC. `sink` is a file path or an `IO`. one record batch. `compress` is `nothing`, `:lz4`, or `:zstd`. `metadata`/`colmetadata` attach schema- and per-column key-value pairs (a `Dict`, or pairs; `colmetadata` maps column name `Symbol`s to them). +Schema metadata is inherited from an `Arrow.Table` or `Arrow.Stream` when +`metadata` is `nothing`. Inherited batch statistics are dropped because the +values, columns, or batch boundaries may have changed. Explicit `metadata` +is used as supplied. Returns `sink`: the path for the path method, the `io` for the `IO` method. The one-argument form curries for pipelines: `table |> Arrow.write(sink)`. @@ -101,6 +105,7 @@ so producing tables one at a time never holds more than the current table in memory. `sink` is a file path (the writer opens and owns the handle) or an `IO` (borrowed; `close` finishes the IPC output but leaves the `IO` open). The function form runs `f(writer)` and always closes the writer. +Metadata follows the inheritance rules of [`Arrow.write`](@ref). w = Arrow.Writer(path) for tbl in tables @@ -227,12 +232,7 @@ function write(w::Writer, tbl) _retainedfieldfn(retained, names), w.colmetadata, ) - schmeta = - w.metadata !== nothing ? _metapairs(w.metadata) : - ( - retained === nothing || retained.metadata === nothing ? nothing : - collect(Pair{String,String}, retained.metadata) - ) + schmeta = _writermetadata(w.metadata, retained) sch = AC.Schema(fields; metadata=schmeta) st = beginwrite!( w.io, @@ -375,6 +375,15 @@ _retainedschema(t::Table) = getfield(t, :schema) _retainedschema(s::Stream) = _tableschema(getfield(s, :src)) _retainedschema(::Any) = nothing +function _writermetadata(metadata, retained) + metadata !== nothing && return _metapairs(metadata) + (retained === nothing || retained.metadata === nothing) && return nothing + # Statistics describe the source's values, field indexes, and batch + # boundaries. Retaining its descriptors does not preserve that identity. + # Keep user metadata in order, including duplicate keys. + return Pair{String,String}[kv for kv in retained.metadata if first(kv) != STATS_KEY] +end + "Dictionary pools retained by one facade partition, in column order." _partitiondictpools(t::Table) = getfield(t, :retainedpools) _partitiondictpools(::Any) = nothing @@ -424,12 +433,7 @@ function _writebytes( _retainedfieldfn(retained, names), colmetadata, ) - schmeta = - metadata !== nothing ? _metapairs(metadata) : - ( - retained === nothing || retained.metadata === nothing ? nothing : - collect(Pair{String,String}, retained.metadata) - ) + schmeta = _writermetadata(metadata, retained) sch = AC.Schema(fields; metadata=schmeta) batches = _tablebatches(sch, coldata, rowcounts) codec = compress === nothing ? :none : compress diff --git a/test/rewrite_regressions.jl b/test/rewrite_regressions.jl index b6b9da6..8394e04 100644 --- a/test/rewrite_regressions.jl +++ b/test/rewrite_regressions.jl @@ -400,6 +400,110 @@ end @test count(kv -> first(kv) == Arrow.STATS_KEY, statsschema.metadata) == 1 end + @testset "rewrites discard inherited batch statistics" begin + xfield, xdata = AC.fromjulia("x", Union{Missing,Int64}[1, 2]) + yfield, ydata = AC.fromjulia("y", Int64[100, 200]) + colmetadata = ["unit" => "first", "unit" => "second"] + xfield = AC.Field("x", xfield.type; nullable=true, metadata=colmetadata) + metadata = ["source" => "first", "note" => "middle", "source" => "second"] + schema = AC.Schema([xfield, yfield]; metadata=metadata) + batches = [AC.RecordBatch(schema, [xdata, ydata], 2)] + statsschema = Arrow.withstatistics(schema, batches) + statspair = last(statsschema.metadata) + # Remove every copy of the derived key, without deduplicating or + # reordering the surrounding user metadata. + statsschema = AC.Schema( + [xfield, yfield]; + metadata=[metadata[1], statspair, metadata[2], statspair, metadata[3]], + ) + bytes = Arrow.writefile(statsschema, batches) + _, xdata2 = AC.fromjulia("x", Union{Missing,Int64}[100, 200]) + _, ydata2 = AC.fromjulia("y", Int64[300, 400]) + groupedbytes = Arrow.statsfile( + schema, + [only(batches), AC.RecordBatch(schema, [xdata2, ydata2], 2)], + ) + + for incremental in (false, true), file in (false, true) + rewrite = function (table; metadata=nothing) + io = IOBuffer() + if incremental + Arrow.Writer(io; file=file, metadata=metadata) do writer + Arrow.write(writer, table) + end + else + Arrow.write(io, table; file=file, metadata=metadata) + end + return take!(io) + end + + table = Arrow.Table(bytes) + table.x[1] = 100 + table.x[2] = missing + rewritten = rewrite(table) + result = Arrow.Table(rewritten) + @test isequal(result.x, [100, missing]) + @test collect(getfield(result, :schema).metadata) == metadata + @test collect(getfield(result, :schema).fields[1].metadata) == colmetadata + for source in (rewritten, _BytesSource(rewritten)) + filtered = Arrow.Table(source; scan=Tables.Scan(filter=Tables.col(:x) > 50)) + @test filtered.x == [100] + nulls = Arrow.Table( + source; + scan=Tables.Scan(filter=Tables.isnull(Tables.col(:x))), + ) + @test isequal(nulls.x, [missing]) + end + # Reading or rewriting a table must not change its source metadata. + @test collect(getfield(table, :schema).metadata) == + collect(statsschema.metadata) + + # Projection changes the flattened field indexes used by statistics. + projected = Arrow.Table(bytes; scan=Tables.Scan(select=(:y => :picked,))) + selected = rewrite(projected) + @test Arrow.Table( + selected; + scan=Tables.Scan(filter=Tables.col(:picked) > 50), + ).picked == [100, 200] + + # Stream inputs also inherit a source schema, even without mutation. + streamed = Arrow.Table(rewrite(Arrow.Stream(bytes))) + @test streamed.x == [1, 2] + @test collect(getfield(streamed, :schema).metadata) == metadata + + # Change [1, 2] / [100, 200] into one batch for the eager writer, + # or [1, 2, 100] / [200] for the incremental writer. Retaining two + # statistics records in the latter case would wrongly prune 100. + io = IOBuffer() + if incremental + Arrow.Writer(io; file=file) do writer + for scan in (Tables.Scan(limit=3), Tables.Scan(offset=3)) + Arrow.write(writer, Arrow.Table(groupedbytes; scan=scan)) + end + end + else + Arrow.write(io, Arrow.Table(groupedbytes); file=file) + end + regrouped = take!(io) + @test Arrow.Table(regrouped).x == [1, 2, 100, 200] + @test length(Arrow.Stream(regrouped)) == (incremental ? 2 : 1) + for source in (regrouped, _BytesSource(regrouped)) + @test Arrow.Table(source; scan=Tables.Scan(filter=Tables.col(:x) > 50)).x == + [100, 200] + end + + # An explicit metadata override is caller-owned and keeps its + # existing semantics, including duplicate keys and order. + override = ["source" => "override", "source" => "last"] + explicit = Arrow.Table(rewrite(table; metadata=override)) + @test collect(getfield(explicit, :schema).metadata) == override + explicitstats = + Arrow.Table(rewrite(Arrow.Table(bytes); metadata=statsschema.metadata)) + @test collect(getfield(explicitstats, :schema).metadata) == + collect(statsschema.metadata) + end + end + @testset "NUL field names stay Core-only and fail cleanly at the facade" begin field, data = AC.fromjulia("a\0b", Int64[1]) schema = AC.Schema([field])
