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-stash-2 in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
commit 1917fad29aed01bd1a8bf4de28004822edb4745a Merge: b8ebfc5 c27c22f Author: Jacob Quinn <[email protected]> AuthorDate: Wed Jan 14 01:26:10 2026 -0700 WIP on main: b8ebfc5 fix test failures on 1.12, avoid race condition in multithreaded partitioned writes (#582) src/arraytypes/dictencoding.jl | 6 ++- src/table.jl | 94 +++++++++++++++++++++++++++++++++++++++++- src/write.jl | 29 +++++++++++-- 3 files changed, 122 insertions(+), 7 deletions(-) diff --cc src/arraytypes/dictencoding.jl index 84d5105,84d5105..1250bdf --- a/src/arraytypes/dictencoding.jl +++ b/src/arraytypes/dictencoding.jl @@@ -233,10 -233,10 +233,12 @@@ function arrowvector end # horrible hack? yes. better than taking CategoricalArrays dependency? also yes. if typeof(pool).name.name == :CategoricalRefPool ++ # In CategoricalArrays 1.0+, levels() returns a CategoricalArray, ++ # so we need to collect() to get a plain array if eltype(x) >: Missing -- pool = vcat(missing, DataAPI.levels(x)) ++ pool = vcat(missing, collect(DataAPI.levels(x))) else -- pool = DataAPI.levels(x) ++ pool = collect(DataAPI.levels(x)) for i = 1:length(inds) @inbounds inds[i] -= 1 end diff --cc src/table.jl index de8bfc3,de8bfc3..201e9d0 --- a/src/table.jl +++ b/src/table.jl @@@ -23,11 -23,11 +23,58 @@@ en ArrowBlob(bytes::Vector{UInt8}, pos::Int, len::Nothing) = ArrowBlob(bytes, pos, length(bytes)) ++# Registry to track mmap'd byte arrays and their source file paths ++# Used to detect when user tries to write to a file they're currently reading from ++const MMAP_FILE_REGISTRY = WeakKeyDict{Vector{UInt8},String}() ++ tobytes(bytes::Vector{UInt8}) = bytes tobytes(io::IO) = Base.read(io) --tobytes(io::IOStream) = Mmap.mmap(io) ++ ++function tobytes(io::IOStream) ++ # Try to use mmap for seekable streams (regular files) ++ # Fall back to read() for non-seekable streams (FIFOs, pipes, etc.) ++ # where mmap would return empty bytes or fail ++ try ++ # Check if stream is seekable by testing position/seek ++ pos = position(io) ++ seek(io, pos) ++ # Also check filesize - FIFOs report size 0 ++ if filesize(io) > 0 ++ return Mmap.mmap(io) ++ end ++ catch ++ # Not seekable, fall through to read ++ end ++ # Non-seekable or zero-size: read all bytes ++ return Base.read(io) ++end ++ ++function tobytes(file_path::AbstractString) ++ bytes = open(tobytes, file_path, "r") ++ # Register the mmap'd bytes with the absolute file path for later detection ++ MMAP_FILE_REGISTRY[bytes] = abspath(file_path) ++ return bytes ++end ++# Fallback for other path-like types (e.g., FilePathsBase.PosixPath, AWSS3.S3Path) ++# These types implement `open(file_path, mode)` so use the same pattern as AbstractString tobytes(file_path) = open(tobytes, file_path, "r") ++""" ++ Arrow.getsourcefile(bytes::Vector{UInt8}) ++ ++Return the source file path if `bytes` was mmap'd from a file, otherwise `nothing`. ++""" ++getsourcefile(bytes::Vector{UInt8}) = get(MMAP_FILE_REGISTRY, bytes, nothing) ++ ++""" ++ Arrow.sourcefile(x) ++ ++Return the source file path if `x` is an Arrow array backed by mmap'd data, otherwise `nothing`. ++For Arrow.Table, returns a vector of unique source files. ++""" ++sourcefile(x) = nothing ++sourcefile(x::ArrowVector) = getfield(x, :arrow) isa Vector{UInt8} ? getsourcefile(getfield(x, :arrow)) : nothing ++ struct BatchIterator bytes::Vector{UInt8} startpos::Int @@@ -349,6 -349,6 +396,51 @@@ lookup(t::Table) = getfield(t, :lookup schema(t::Table) = getfield(t, :schema) metadata(t::Table) = getfield(t, :metadata) ++""" ++ Arrow.sourcefiles(t::Table) ++ ++Return a vector of unique source file paths that the table's columns are backed by. ++Returns an empty vector if no columns are backed by mmap'd files. ++""" ++function sourcefiles(t::Table) ++ files = String[] ++ for col in columns(t) ++ f = sourcefile(col) ++ if f !== nothing && f ∉ files ++ push!(files, f) ++ end ++ end ++ return files ++end ++ ++""" ++ Arrow.ismmappedfrom(t::Table, file_path::AbstractString) ++ Arrow.ismmappedfrom(t::Table, file_path) ++ ++Return `true` if the table contains columns backed by mmap'd data from the specified file. ++""" ++function ismmappedfrom(t::Table, file_path::AbstractString) ++ target = abspath(file_path) ++ for col in columns(t) ++ f = sourcefile(col) ++ if f !== nothing && f == target ++ return true ++ end ++ end ++ return false ++end ++# Fallback for path-like types (e.g., FilePathsBase.PosixPath) ++# Note: We only track string paths in MMAP_FILE_REGISTRY, so non-string paths ++# won't match anyway. This fallback tries string conversion, but returns false ++# if conversion fails (some path types have broken string() implementations). ++function ismmappedfrom(t::Table, file_path) ++ try ++ return ismmappedfrom(t, string(file_path)) ++ catch ++ return false # Can't convert to string, can't match registry ++ end ++end ++ """ Arrow.getmetadata(x) diff --cc src/write.jl index 4c3800f,4c3800f..1aef397 --- a/src/write.jl +++ b/src/write.jl @@@ -54,6 -54,6 +54,17 @@@ function write en write(io_or_file; kw...) = x -> write(io_or_file, x; kw...) function write(file_path, tbl; kwargs...) ++ # Check if tbl is an Arrow.Table that was mmap'd from the target file ++ # Writing to the same file would cause a bus error (SIGBUS) because ++ # opening for write truncates the file, invalidating the mmap'd memory ++ if tbl isa Table && ismmappedfrom(tbl, file_path) ++ path_str = file_path isa AbstractString ? abspath(file_path) : string(file_path) ++ throw(ArgumentError( ++ "Cannot write to '$(path_str)': the source table contains data " * ++ "mmap'd from this file. Writing would invalidate the memory and cause a crash. " * ++ "To overwrite the file, first copy the data: Arrow.write(path, Arrow.Table(copy.(eachcol(tbl))))" ++ )) ++ end open(Writer, file_path; file=true, kwargs...) do writer write(writer, tbl) end @@@ -567,10 -567,10 +578,20 @@@ function Base.write(io::IO, msg::Messag metalen = padding(length(msg.msgflatbuf), alignment) @debug "writing message: metalen = $metalen, bodylen = $(msg.bodylen), isrecordbatch = $(msg.isrecordbatch), headerType = $(msg.headerType)" if msg.blockmsg -- push!( -- blocks[msg.isrecordbatch ? 1 : 2], -- Block(position(io), metalen + 8, msg.bodylen), -- ) ++ # Track block positions for file format footer ++ # Skip for non-seekable streams (streaming to FIFOs, sockets, etc.) ++ # where position() would fail - block tracking is only needed for file format ++ pos = try ++ position(io) ++ catch ++ nothing # Non-seekable stream, skip block tracking ++ end ++ if pos !== nothing ++ push!( ++ blocks[msg.isrecordbatch ? 1 : 2], ++ Block(pos, metalen + 8, msg.bodylen), ++ ) ++ end end # now write the final message spec out # continuation byte
