This is an automated email from the ASF dual-hosted git repository. quinnj pushed a commit to branch core-rewrite in repository https://gitbox.apache.org/repos/asf/arrow-julia.git
commit 415d0c798d492b866d0c23a7a12dba8fe473a253 Author: Jacob Quinn <[email protected]> AuthorDate: Wed Aug 19 02:28:46 2026 -0600 fix: resolve round 65 findings — Integer length guard, policy-compliant worker pool, no GCS claim - `SourceFile` requires `sourcelength` to return an `Integer` (a Float, String, or `nothing` is a ValidationError, pinned through `Arrow.Table`) - the range-read worker pool follows the repository guidelines: the shared request counter is an `@atomic` field on a mutable `_SpanQueue` (no `Threads.Atomic`), each worker is `errormonitor(Threads.@spawn …)`, and the per-worker loop is its own function - the CloudStore extension claims S3 and Azure Blob Storage only: the registered CloudStore releases (1.6–1.8) have no GCS `Object` path Co-Authored-By: Claude Fable 5 <[email protected]> --- README.md | 2 +- docs/src/manual.md | 2 +- ext/ArrowCloudStoreExt.jl | 2 +- src/scan.jl | 32 ++++++++++++++++++++++++-------- src/source.jl | 2 +- test/facade_tests.jl | 12 ++++++++++-- 6 files changed, 38 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 03f393c..36de18b 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ This is a pure Julia implementation of the - `src/table.jl`, `src/write.jl` — the public facade: `Arrow.Table`, `Arrow.Stream`, `Arrow.write`, `close!`. - `ext/ArrowCloudStoreExt.jl` — CloudStore.jl objects as sources (S3, - Azure Blob Storage, GCS) with concurrent range reads. + Azure Blob Storage) with concurrent range reads. - `src/FlatBuffers/` — the vendored FlatBuffers runtime the generated bindings run over. - `src/ArrowStrings/` — ArrowStrings.jl, a separate package (to be diff --git a/docs/src/manual.md b/docs/src/manual.md index a96f7df..f1559eb 100644 --- a/docs/src/manual.md +++ b/docs/src/manual.md @@ -249,7 +249,7 @@ selected (and filter-referenced) columns, coalesced into a few range reads: three rounds of requests, however many columns and batches the file holds. With [CloudStore.jl](https://github.com/JuliaServices/CloudStore.jl) -loaded, a `CloudStore.Object` (S3, Azure Blob Storage, GCS) is such a source +loaded, a `CloudStore.Object` (S3 or Azure Blob Storage) is such a source directly, and its planned ranges are requested concurrently: ```julia diff --git a/ext/ArrowCloudStoreExt.jl b/ext/ArrowCloudStoreExt.jl index 6ba9c60..5d1bfb1 100644 --- a/ext/ArrowCloudStoreExt.jl +++ b/ext/ArrowCloudStoreExt.jl @@ -16,7 +16,7 @@ # CloudStore.jl objects as Arrow byte-range sources: `Arrow.Table(obj; # scan=…)` reads only the selected columns' bytes from S3, Azure Blob -# Storage, or GCS through HTTP `Range` requests. +# Storage through HTTP `Range` requests. module ArrowCloudStoreExt using Arrow diff --git a/src/scan.jl b/src/scan.jl index 01424e6..3614245 100644 --- a/src/scan.jl +++ b/src/scan.jl @@ -782,8 +782,8 @@ function SourceFile( gap = Int64(coalesce_gap) gap >= 0 || throw(ArgumentError("negative coalesce gap")) reported = sourcelength(src) - (reported >= 0 && reported <= typemax(Int64)) || - throw(ValidationError("source reports an invalid length $reported")) + (reported isa Integer && reported >= 0 && reported <= typemax(Int64)) || + throw(ValidationError("source reports an invalid length $(repr(reported))")) return SourceFile( src, Int64(reported), @@ -916,14 +916,10 @@ function _readspans(sf::SourceFile, spans::Vector{NTuple{2,Int64}}) end return results end - next = Threads.Atomic{Int}(1) + queue = _SpanQueue(0) try @sync for _ = 1:k - Threads.@spawn while true - i = Threads.atomic_add!(next, 1) - i > n && break - results[i] = _fetchexact(sf, spans[i][1], spans[i][2]) - end + errormonitor(Threads.@spawn _readworker!(results, sf, spans, queue)) end catch e rethrow(_firstcause(e)) @@ -931,6 +927,26 @@ function _readspans(sf::SourceFile, spans::Vector{NTuple{2,Int64}}) return results end +# The shared request counter of one round's worker pool. +mutable struct _SpanQueue + @atomic next::Int +end + +# One worker: claim the next request index, read it into its slot, repeat. +function _readworker!( + results::Vector{Vector{UInt8}}, + sf::SourceFile, + spans::Vector{NTuple{2,Int64}}, + queue::_SpanQueue, +) + n = length(spans) + while true + i = @atomic queue.next += 1 + i > n && return nothing + results[i] = _fetchexact(sf, spans[i][1], spans[i][2]) + end +end + # The underlying exception of a failed worker task (`@sync` wraps it). function _firstcause(e) e isa CompositeException && !isempty(e) && return _firstcause(first(e)) diff --git a/src/source.jl b/src/source.jl index 8310512..d0f7840 100644 --- a/src/source.jl +++ b/src/source.jl @@ -60,7 +60,7 @@ tbl = Arrow.Table(BytesSource(bytes); scan=Tables.Scan(select=(:a, :b))) The `CloudStore.jl` extension makes a `CloudStore.Object` a source, so `Arrow.Table(CloudStore.Object(bucket, key); scan=…)` reads just the needed -column bytes from S3, Azure Blob Storage, or GCS. Stream-format objects have +column bytes from S3 or Azure Blob Storage. Stream-format objects have no footer and are read whole. """ abstract type AbstractArrowSource end diff --git a/test/facade_tests.jl b/test/facade_tests.jl index eb91a6a..fd21cd3 100644 --- a/test/facade_tests.jl +++ b/test/facade_tests.jl @@ -93,7 +93,7 @@ end Arrow.sourcelength(s::_WrongTypeSource) = length(s.data) Arrow.readrange(s::_WrongTypeSource, off, len) = String(s.data[(off + 1):(off + len)]) struct _BadLengthSource <: Arrow.AbstractArrowSource - reported::Integer + reported::Any end Arrow.sourcelength(s::_BadLengthSource) = s.reported Arrow.readrange(s::_BadLengthSource, off, len) = zeros(UInt8, len) @@ -282,7 +282,15 @@ end scan=Tables.Scan(select=(:a,)), ) end - for reported in (-1, Int128(typemax(Int64)) + 1, Int128(typemin(Int64)) - 1) + for reported in ( + -1, + Int128(typemax(Int64)) + 1, + Int128(typemin(Int64)) - 1, + 3402.0, + 1.5, + "3402", + nothing, + ) @test_throws Arrow.AC.ValidationError Arrow.Table(_BadLengthSource(reported)) end end
