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-1 in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
commit 1e5460de2a483900f3b1e8b21b96fe119d6dc3ae Merge: 7d33fc2 f144211 Author: Jacob Quinn <[email protected]> AuthorDate: Wed Jan 14 01:52:24 2026 -0700 WIP on pr-583: 7d33fc2 Use local ArrowTypes in CI tests and release verification src/table.jl | 93 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/write.jl | 29 ++++++++++++++++--- 2 files changed, 117 insertions(+), 5 deletions(-) diff --cc src/table.jl index de8bfc3,de8bfc3..7652855 --- a/src/table.jl +++ b/src/table.jl @@@ -23,11 -23,11 +23,57 @@@ 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 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`. ++""" ++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 +395,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
