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
The following commit(s) were added to refs/heads/core-rewrite by this push:
new d3ea7a7 fix: resolve round 64 findings — Arrow-owned range reads,
ETag pinning, one body round
d3ea7a7 is described below
commit d3ea7a7cf16001985a0237e18c9c873307acbc6c
Author: Jacob Quinn <[email protected]>
AuthorDate: Wed Aug 19 02:01:01 2026 -0600
fix: resolve round 64 findings — Arrow-owned range reads, ETag pinning, one
body round
- (H) `readranges` is gone from the source interface: Arrow reads a round's
planned spans itself, through `readrange`, with a worker pool of
`concurrentreads(src)` tasks (default 1) pulling requests off one counter
and storing every result by request index — a transport's completion
order can never permute payloads, and the reads in flight are bounded
whatever the span count. `readrange` results are type- and
length-checked; the docstring states the trusted-source boundary (Arrow
cannot authenticate the bytes a source returns for a range)
- (M) CloudStore extension: every range GET carries `If-Match` with the
object's ETag, so a handle is pinned to one object version — an
overwritten key fails the next read instead of mixing versions across
request rounds (pinned against Minio); `concurrentreads` is 16
- (M) the planner fetches the selected dictionary bodies and the selected
record buffers in ONE round (dictionaries decode first from the shared
spans); the dead `blockwants` map is gone
- (L) a Footer larger than the tail window is read once and cached on the
handle; ranges the cached tail already covers are served from it without
a request (a file no larger than the window costs exactly one request)
- (L) `sourcelength` is validated as an Integer before conversion; the
coalescer's negative-range guard is a `ValidationError`
- (L) `Arrow.Table(src)` without a scan, with an unpushable scan, over a
zero-field file, or over a stream-format object reads the object whole
(the cached tail plus its prefix: two requests) — what the manual and
docstring say; the DESIGN request-count model matches
- tests: facade pins for request rounds through the public path, tail-once,
tail-cache serving, bounded concurrency with out-of-order completion, and
every contract violation (short/long/wrong-type payloads, invalid
lengths) as ValidationError; the ext test pins ETag rejection; the
battery's coalescing check plans outside a small tail
Co-Authored-By: Claude Fable 5 <[email protected]>
---
docs/dev/DESIGN-scan-ranges-trim.md | 27 +++--
docs/dev/core-README.md | 2 +-
docs/src/manual.md | 10 +-
docs/src/reference.md | 2 +-
ext/ArrowCloudStoreExt.jl | 35 ++++---
src/scan.jl | 194 ++++++++++++++++++++++--------------
src/source.jl | 25 +++--
src/table.jl | 60 ++++++-----
test/cloudstore_tests.jl | 18 +++-
test/facade_tests.jl | 113 +++++++++++++++++++++
test/scan_battery.jl | 10 +-
11 files changed, 341 insertions(+), 155 deletions(-)
diff --git a/docs/dev/DESIGN-scan-ranges-trim.md
b/docs/dev/DESIGN-scan-ranges-trim.md
index 0d888d6..60670a7 100644
--- a/docs/dev/DESIGN-scan-ranges-trim.md
+++ b/docs/dev/DESIGN-scan-ranges-trim.md
@@ -153,7 +153,7 @@ sequential — cloud-native access is a file-format feature,
stated plainly.
`tailbytes` (default 64 KiB). Its trailing magic decides file vs stream
format (a stream object is read whole instead), and it covers
footer-length + magic + the whole Footer in almost every real file; if
- `footerlen + 10 > tailbytes`, one exact follow-up fetch.
+ `footerlen + 10 > tailbytes`, one exact follow-up fetch (also cached).
→ schema, Block indexes, (§3) statistics — everything pruning needs.
The leading magic is not fetched: the Footer is the sole authority.
2. **Statistics prune** from the Footer metadata — zero additional fetches.
@@ -168,9 +168,11 @@ sequential — cloud-native access is a file-format feature,
stated plainly.
ranges → **coalesce** ranges with gaps below `coalesce_gap` (default
~256 KiB — a gap fetch is usually cheaper than a request round-trip;
both knobs are options, not constants).
-5. **Body fetches**: each coalesced range lands in its own owned heap
- region. Decode resolves each declared buffer `(offset, len)` to its
- containing fetched range and subslices — the message-body-authority
+5. **Body fetches** — the selected dictionary bodies and the selected record
+ buffers in ONE round: each coalesced range lands in its own owned heap
+ region; the dictionaries decode first from the shared spans. Decode
+ resolves each declared buffer `(offset, len)` to its containing fetched
+ range and subslices — the message-body-authority
invariant becomes *"every buffer must fall inside a fetched range that
was itself derived from the verified buffer table"*: same trust story,
sparse backing.
@@ -194,23 +196,26 @@ planner; transports live in extensions:
abstract type AbstractArrowSource end
sourcelength(src)::Integer # total object length, known
up front
readrange(src, offset, len)::Vector{UInt8} # one exact range, 0-based
offset
- readranges(src, ranges) -> Vector{Vector{UInt8}}
- # default: serial map over readrange; transports override for
- # concurrent range GETs — concurrency stays in the extension,
- # never in Arrow.
+ concurrentreads(src)::Int # default 1
+ # Arrow reads a round's planned ranges through readrange with a
+ # worker pool of that size, storing every result by request index —
+ # a source's completion order can never permute payloads, and the
+ # reads in flight are bounded whatever the span count.
- The entry point is `Arrow.Table(src; scan=…)`, which builds the internal
`SourceFile(src; limits, tailbytes, coalesce_gap)` handle and runs
`Tables.scan(sf, scan)`; the source's length is read once, at handle
construction, and the tail once per handle. `Arrow.Table(src)` without a
- scan plans every column; `Arrow.Stream(src)` reads the object whole.
+ scan, with a scan that cannot be pushed down, over a zero-field file, or
+ over a stream-format object reads the object whole (one request beyond
+ the cached tail), as does `Arrow.Stream(src)`.
- The abstract type is a dynamic call under `--trim` for a source type the
trimmed app never mentions; an app that names its concrete source type
resolves statically. The planner itself is arithmetic over `Int64`s.
- Extension: `ext/ArrowCloudStoreExt.jl` (loaded with CloudStore.jl) makes
a `CloudStore.Object` a source — its known `size` is the length, one
- range is one HTTP `Range` GET, and `readranges` issues a round's ranges
- concurrently — and adds `Arrow.Table(::CloudStore.Object; …)` and
+ range is one HTTP `Range` GET pinned to the object's ETag with
+ `If-Match`, and `concurrentreads` is 16 — and adds
`Arrow.Table(::CloudStore.Object; …)` and
`Arrow.Stream(::CloudStore.Object; …)`. An HTTP transport is the same two
methods. Zero new hard deps.
- The differential test: sparse fetch ≡ whole-file read, plus
diff --git a/docs/dev/core-README.md b/docs/dev/core-README.md
index e429258..24672f3 100644
--- a/docs/dev/core-README.md
+++ b/docs/dev/core-README.md
@@ -36,7 +36,7 @@ scope of every layer.
| `src/ipc_read.jl` | Checked IPC stream framing, resource limits,
metadata-to-Core mapping, dictionary state, one registry-driven decoder,
per-buffer decompression |
| `src/ipc_write.jl` | The write half over the same registry: Core-to-metadata
mapping, one generic registry-driven encoder, replacement-on-change dictionary
batches, per-buffer compression, the file format (Block index + Footer), and
the lazy random-access `ArrowFile` reader |
| `src/cdata.jl` | C data and C stream interfaces both directions: zero-copy
ownership, move semantics, exactly-once release, field and schema metadata
transport |
-| `src/source.jl` | The `AbstractArrowSource` byte-range source interface
(`sourcelength`, `readrange`, `readranges`) |
+| `src/source.jl` | The `AbstractArrowSource` byte-range source interface
(`sourcelength`, `readrange`, `concurrentreads`) |
| `src/scan.jl` | `Tables.Scan` pushdown over the file format, sparse
byte-range reads over a source (`SourceFile`), embedded per-batch statistics |
| `src/table.jl`, `src/write.jl` | The facade |
| `ext/ArrowCloudStoreExt.jl` | CloudStore.jl objects as sources: HTTP `Range`
reads, concurrent per planned range |
diff --git a/docs/src/manual.md b/docs/src/manual.md
index 2105b2c..a96f7df 100644
--- a/docs/src/manual.md
+++ b/docs/src/manual.md
@@ -271,10 +271,12 @@ Arrow.readrange(s::HTTPSource, offset, len) =
fetchbytes(s.url, offset, len) #
tbl = Arrow.Table(HTTPSource(url, objectsize); scan = Scan(select = (:id,)))
```
-Overriding [`Arrow.readranges`](@ref) lets a transport issue the planned
-ranges concurrently; the default reads them one at a time. Without a scan
-the whole object is read; a stream-format object (no footer) is always read
-whole. Arrow.jl has no HTTP or cloud dependency of its own.
+Overriding [`Arrow.concurrentreads`](@ref) lets Arrow issue a round's
+planned ranges concurrently through `readrange` (up to that many at a
+time, results placed by request); the default reads them one at a time.
+Without a scan the whole object is read, as is a stream-format object (no
+footer) and a scan that cannot be pushed down. Arrow.jl has no HTTP or
+cloud dependency of its own.
## Writing
diff --git a/docs/src/reference.md b/docs/src/reference.md
index a572a0e..5831288 100644
--- a/docs/src/reference.md
+++ b/docs/src/reference.md
@@ -42,7 +42,7 @@ Arrow.DictEncode
Arrow.AbstractArrowSource
Arrow.sourcelength
Arrow.readrange
-Arrow.readranges
+Arrow.concurrentreads
```
## The C data and C stream interfaces
diff --git a/ext/ArrowCloudStoreExt.jl b/ext/ArrowCloudStoreExt.jl
index 9f749ef..6ba9c60 100644
--- a/ext/ArrowCloudStoreExt.jl
+++ b/ext/ArrowCloudStoreExt.jl
@@ -26,41 +26,48 @@ using CloudStore: CloudStore, Object
CloudObjectSource(obj::CloudStore.Object) <: Arrow.AbstractArrowSource
A `CloudStore.Object` as an [`Arrow.AbstractArrowSource`](@ref): the length
-is the object's known size, one range is one HTTP `Range` GET, and the
-planned ranges of a scan are fetched concurrently. `Arrow.Table(obj; …)` and
-`Arrow.Stream(obj; …)` construct one implicitly.
+is the object's known size, one range is one HTTP `Range` GET pinned to the
+object's ETag with `If-Match` (an overwritten key fails the read instead of
+mixing versions across requests), and Arrow issues up to
+`CONCURRENT_RANGE_READS` of a round's ranges at once. `Arrow.Table(obj; …)`
+and `Arrow.Stream(obj; …)` construct one implicitly.
"""
struct CloudObjectSource{O<:Object} <: Arrow.AbstractArrowSource
obj::O
end
+# Independent HTTP range GETs are latency-bound, so a round's requests are
+# worth overlapping; the bound keeps a fragmented object from becoming a
+# request storm.
+const CONCURRENT_RANGE_READS = 16
+
Arrow.sourcelength(s::CloudObjectSource) = Int64(s.obj.size)
+Arrow.concurrentreads(::CloudObjectSource) = CONCURRENT_RANGE_READS
+
+# The object's ETag as an `If-Match` value (the header wants it quoted).
+function _ifmatch(etag::AbstractString)
+ isempty(etag) && return nothing
+ return startswith(etag, '"') ? String(etag) : string('"', etag, '"')
+end
function Arrow.readrange(s::CloudObjectSource, off, len)
len == 0 && return UInt8[]
last = off + len - 1
obj = s.obj
+ headers = ["Range" => "bytes=$(off)-$(last)"]
+ etag = _ifmatch(obj.eTag)
+ etag === nothing || push!(headers, "If-Match" => etag)
bytes = CloudStore.get(
obj.store,
obj.key;
credentials=obj.credentials,
- headers=["Range" => "bytes=$(off)-$(last)"],
+ headers=headers,
allowMultipart=false,
objectMaxSize=Int(len),
)
return bytes isa Vector{UInt8} ? bytes : Vector{UInt8}(bytes)
end
-# One task per planned range: the requests are independent GETs, and the
-# planner has already coalesced neighbours, so the remaining ranges are
-# worth issuing at once.
-function Arrow.readranges(s::CloudObjectSource,
ranges::Vector{NTuple{2,Int64}})
- length(ranges) <= 1 &&
- return Vector{UInt8}[Arrow.readrange(s, off, len) for (off, len) in
ranges]
- tasks = [Threads.@spawn Arrow.readrange(s, off, len) for (off, len) in
ranges]
- return Vector{UInt8}[fetch(t)::Vector{UInt8} for t in tasks]
-end
-
Arrow.Table(obj::Object; kw...) = Arrow.Table(CloudObjectSource(obj); kw...)
Arrow.Stream(obj::Object; kw...) = Arrow.Stream(CloudObjectSource(obj); kw...)
diff --git a/src/scan.jl b/src/scan.jl
index 87bbf85..01424e6 100644
--- a/src/scan.jl
+++ b/src/scan.jl
@@ -769,6 +769,9 @@ struct SourceFile{S<:AbstractArrowSource}
# The last `tailbytes` of the object, read once and reused by every
# footer parse and format probe on this handle: `(bytes, tailstart)`.
tail::Base.RefValue{Union{Nothing,Tuple{Vector{UInt8},Int64}}}
+ # A Footer that did not fit in the tail window, read once as
+ # `(footerstart, bytes)` and reused by every footer parse on this handle.
+ footer::Base.RefValue{Union{Nothing,Tuple{Int64,Vector{UInt8}}}}
end
function SourceFile(
src::AbstractArrowSource;
@@ -778,15 +781,17 @@ function SourceFile(
)
gap = Int64(coalesce_gap)
gap >= 0 || throw(ArgumentError("negative coalesce gap"))
- len = Int64(sourcelength(src))
- len >= 0 || throw(ValidationError("source reports a negative length"))
+ reported = sourcelength(src)
+ (reported >= 0 && reported <= typemax(Int64)) ||
+ throw(ValidationError("source reports an invalid length $reported"))
return SourceFile(
src,
- len,
+ Int64(reported),
limits,
Int64(max(tailbytes, 32)),
gap,
Base.RefValue{Union{Nothing,Tuple{Vector{UInt8},Int64}}}(nothing),
+ Base.RefValue{Union{Nothing,Tuple{Int64,Vector{UInt8}}}}(nothing),
)
end
@@ -801,6 +806,18 @@ function _fetchtail(sf::SourceFile)
return (tail, tailstart)
end
+# A Footer that escapes the tail window: one exact read, cached on the handle
+# so the schema pass and the scan pass share it.
+function _fetchfooter(sf::SourceFile, footerstart::Int64, footerlen::Int64)
+ cached = sf.footer[]
+ if cached !== nothing && cached[1] == footerstart && length(cached[2]) ==
footerlen
+ return cached[2]
+ end
+ bytes = _fetchexact(sf, footerstart, footerlen)
+ sf.footer[] = (footerstart, bytes)
+ return bytes
+end
+
# Whether the object is an IPC FILE (trailing `ARROW1` magic) — a stream
# object has no footer and is read whole instead of range-planned.
function _isfilesource(sf::SourceFile)
@@ -816,11 +833,22 @@ function _wholeobject(sf::SourceFile)
end
# One exact range through the source, bounds-checked against the length
-# read at construction and length-checked on return.
+# read at construction and length-checked on return; a range the cached
+# tail window already covers is served from it without a request.
function _fetchexact(sf::SourceFile, off::Int64, len::Int64)
(off >= 0 && len >= 0 && off <= sf.len - len) ||
throw(ValidationError("range fetch [$off, $len] escapes the object"))
- bytes = readrange(sf.src, off, len)
+ cached = sf.tail[]
+ if cached !== nothing
+ tail, tailstart = cached
+ if off >= tailstart && off + len <= tailstart + length(tail)
+ return tail[(off - tailstart + 1):(off - tailstart + len)]
+ end
+ end
+ got = readrange(sf.src, off, len)
+ got isa AbstractVector{UInt8} ||
+ throw(ValidationError("readrange must return a Vector{UInt8}, got
$(typeof(got))"))
+ bytes = got isa Vector{UInt8} ? got : Vector{UInt8}(got)
length(bytes) == len ||
throw(ValidationError("range fetch returned $(length(bytes)) bytes,
expected $len"))
return bytes
@@ -834,7 +862,7 @@ function _coalesce(ranges::Vector{NTuple{2,Int64}},
gap::Int64)
isempty(ranges) && return NTuple{2,Int64}[]
gap >= 0 || throw(ArgumentError("negative coalesce gap"))
all(r -> r[1] >= 0 && r[2] >= 0, ranges) ||
- throw(ArgumentError("negative range offset or length"))
+ throw(ValidationError("negative range offset or length"))
sorted = sort(ranges)
out = NTuple{2,Int64}[sorted[1]]
for (off, len) in Iterators.drop(sorted, 1)
@@ -868,21 +896,48 @@ function _fetchspans(
all(s -> s[1] >= 0 && s[2] >= 0 && s[1] <= sf.len - s[2], spans) ||
throw(ValidationError("planned range escapes the object"))
budget === nothing || foreach(s -> _charge!(budget, s[2], what), spans)
- payloads = readranges(sf.src, spans)
- length(payloads) == length(spans) || throw(
- ValidationError(
- "range fetch returned $(length(payloads)) payloads, expected
$(length(spans))",
- ),
- )
- for (payload, (_, len)) in zip(payloads, spans)
- length(payload) == len || throw(
- ValidationError("range fetch returned $(length(payload)) bytes,
expected $len"),
- )
- end
+ payloads = _readspans(sf, spans)
slices = BufferSlice[BufferSlice(heapregion(p), 0, length(p)) for p in
payloads]
return FetchedSpans(Int64[s[1] for s in spans], Int64[s[2] for s in
spans], slices)
end
+# Read one round's spans through the source, each length-checked, every
+# result stored by its request index: serially, or through a worker pool of
+# `concurrentreads(src)` tasks pulling requests off one counter — so a
+# source's completion order can never permute payloads, and the number of
+# reads in flight is bounded whatever the span count.
+function _readspans(sf::SourceFile, spans::Vector{NTuple{2,Int64}})
+ n = length(spans)
+ results = Vector{Vector{UInt8}}(undef, n)
+ k = min(n, max(1, Int(concurrentreads(sf.src))))
+ if k <= 1
+ for i = 1:n
+ results[i] = _fetchexact(sf, spans[i][1], spans[i][2])
+ end
+ return results
+ end
+ next = Threads.Atomic{Int}(1)
+ 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
+ end
+ catch e
+ rethrow(_firstcause(e))
+ end
+ return results
+end
+
+# The underlying exception of a failed worker task (`@sync` wraps it).
+function _firstcause(e)
+ e isa CompositeException && !isempty(e) && return _firstcause(first(e))
+ e isa TaskFailedException && return _firstcause(e.task.result)
+ return e
+end
+
function _spanslice(fs::FetchedSpans, off::Int64, len::Int64)
len == 0 && return BufferSlice()
i = searchsortedlast(fs.starts, off)
@@ -992,7 +1047,7 @@ function _rangedfooter(sf::SourceFile,
budget::AllocationBudget)
footerbytes =
footerstart >= tailstart ?
tail[(footerstart - tailstart + 1):(footerstart - tailstart +
footerlen)] :
- _fetchexact(sf, footerstart, footerlen)
+ _fetchfooter(sf, footerstart, footerlen)
version, features, dictblocks, recordblocks, reserve =
verify_footer(footerbytes, limits, budget.left)
_charge!(budget, reserve, "verified footer expansion")
@@ -1266,17 +1321,55 @@ function _applyscan(sf::SourceFile, scan::Tables.Scan)
"record batch references dictionary id $(first(missingids)) before
its dictionary batch",
),
)
+ # One body round: the selected dictionary bodies and the selected record
+ # buffers are planned together and fetched in a single pass; the
+ # dictionaries decode first from the shared spans.
+ bodyranges = NTuple{2,Int64}[
+ (dictblocks[i][1] + dictblocks[i][2], dictblocks[i][3]) for i in
wanted_dict
+ ]
+ for (p, _, _) in window
+ block = recordblocks[recidxs[p]]
+ header = headers[p]
+ buffers = something(header.buffers, Meta.Buffer[])
+ variadics = variadiccounts(header)
+ varidx = Ref(1)
+ wants = NTuple{2,Int64}[]
+ bufidx = 1
+ for (j, fld) in enumerate(fields)
+ span64 = _bufferspan(fld, variadics, varidx)
+ span64 <= typemax(Int) || throw(
+ ValidationError("field buffer span $span64 exceeds the host
index range"),
+ )
+ span = Int(span64)
+ if mask[j]
+ for k = bufidx:(bufidx + span - 1)
+ k <= length(buffers) || throw(
+ ValidationError(
+ "metadata declares fewer buffers than the schema
requires",
+ ),
+ )
+ buf = buffers[k]
+ len = Int64(buf.length)
+ off = Int64(buf.offset)
+ (off >= 0 && len >= 0 && AC.checked_add(off, len) <=
block[3]) || throw(
+ ValidationError(
+ "batch buffer [$off, $len] escapes its message
body",
+ ),
+ )
+ len == 0 && continue
+ push!(wants, (off, len))
+ end
+ end
+ bufidx += span
+ end
+ bodystart = block[1] + block[2]
+ append!(bodyranges, NTuple{2,Int64}[(bodystart + off, len) for (off,
len) in wants])
+ end
+ bodyspans = _fetchspans(sf, bodyranges, sf.coalesce_gap)
+
state = DecodeState(budget)
try
if !isempty(wanted_dict)
- bodyspans = _fetchspans(
- sf,
- NTuple{2,Int64}[
- (dictblocks[i][1] + dictblocks[i][2], dictblocks[i][3]) for
- i in wanted_dict
- ],
- sf.coalesce_gap,
- )
for i in wanted_dict
block = dictblocks[i]
msg, v = blockmeta[i]
@@ -1309,55 +1402,6 @@ function _applyscan(sf::SourceFile, scan::Tables.Scan)
end
end
- bodyranges = NTuple{2,Int64}[]
- blockwants = Dict{Int,Vector{NTuple{2,Int64}}}()
- for (p, _, _) in window
- block = recordblocks[recidxs[p]]
- header = headers[p]
- buffers = something(header.buffers, Meta.Buffer[])
- variadics = variadiccounts(header)
- varidx = Ref(1)
- wants = NTuple{2,Int64}[]
- bufidx = 1
- for (j, fld) in enumerate(fields)
- span64 = _bufferspan(fld, variadics, varidx)
- span64 <= typemax(Int) || throw(
- ValidationError(
- "field buffer span $span64 exceeds the host index
range",
- ),
- )
- span = Int(span64)
- if mask[j]
- for k = bufidx:(bufidx + span - 1)
- k <= length(buffers) || throw(
- ValidationError(
- "metadata declares fewer buffers than the
schema requires",
- ),
- )
- buf = buffers[k]
- len = Int64(buf.length)
- off = Int64(buf.offset)
- (off >= 0 && len >= 0 && AC.checked_add(off, len) <=
block[3]) ||
- throw(
- ValidationError(
- "batch buffer [$off, $len] escapes its
message body",
- ),
- )
- len == 0 && continue
- push!(wants, (off, len))
- end
- end
- bufidx += span
- end
- blockwants[p] = wants
- bodystart = block[1] + block[2]
- append!(
- bodyranges,
- NTuple{2,Int64}[(bodystart + off, len) for (off, len) in
wants],
- )
- end
- bodyspans = _fetchspans(sf, bodyranges, sf.coalesce_gap)
-
parts = Dict{Int,Vector{Any}}(idx => Any[] for idx in decodeidx)
outrows = 0
for (p, skip, take) in window
diff --git a/src/source.jl b/src/source.jl
index 7373728..8310512 100644
--- a/src/source.jl
+++ b/src/source.jl
@@ -38,11 +38,15 @@ An implementation defines two methods:
and may override
- Arrow.readranges(src, ranges::Vector{NTuple{2,Int64}}) ->
Vector{Vector{UInt8}}
+ Arrow.concurrentreads(src)::Int # default 1
-whose default reads the planned `(offset, len)` ranges one at a time through
-`readrange`; a transport that can issue them concurrently should. Every
-returned vector must have exactly the requested length.
+to let Arrow issue the planned ranges of one round through `readrange`
+concurrently, at most that many at a time (Arrow places each result by its
+request, so completion order never matters). Every returned vector must have
+exactly the requested length; Arrow validates lengths and offsets and
+refuses violations with a `ValidationError`, but the source is trusted to
+return the bytes that live at the requested range — Arrow cannot
+authenticate them.
```julia
struct BytesSource <: Arrow.AbstractArrowSource
@@ -77,11 +81,12 @@ elements. Required of every implementation.
function readrange end
"""
- Arrow.readranges(src::AbstractArrowSource,
ranges::Vector{NTuple{2,Int64}}) -> Vector{Vector{UInt8}}
+ Arrow.concurrentreads(src::AbstractArrowSource) -> Int
-One result per requested `(offset, len)`, in order. The default reads them
-serially through [`Arrow.readrange`](@ref); a transport overrides this to
-issue the planned ranges concurrently.
+How many [`Arrow.readrange`](@ref) calls Arrow may have in flight at once
+when it fetches the planned ranges of one round. The default, `1`, reads
+them one at a time; a transport whose requests are independent (HTTP range
+GETs) returns a bound suited to it. Arrow runs a worker pool of that size
+and stores every result by request index.
"""
-readranges(src::AbstractArrowSource, ranges::Vector{NTuple{2,Int64}}) =
- Vector{UInt8}[readrange(src, off, len) for (off, len) in ranges]
+concurrentreads(::AbstractArrowSource) = 1
diff --git a/src/table.jl b/src/table.jl
index bb67f1a..6b501b6 100644
--- a/src/table.jl
+++ b/src/table.jl
@@ -49,9 +49,10 @@ One exception: a scan that cannot run in the storage domain
— a filter
literal with no exact storage representation (a cross-domain or
out-of-range value), or an empty projection (`select=()`), whose row count
only the full read can carry — falls back to reading the whole source and
-evaluating over the converted public values. Over a ranged source that
-fallback fetches the entire object, as does reading a zero-field source;
-plan remote filters in each column's public value domain.
+evaluating over the converted public values. An `AbstractArrowSource` is
+read whole in that fallback, as it is without a scan, for a zero-field
+file, and for a stream-format object; plan remote filters in each column's
+public value domain.
Columns are materialized (plain `Vector`s): the returned table does not
borrow the source bytes, and [`Arrow.close!`](@ref) may be called at any
@@ -462,8 +463,9 @@ const _FILE_MAGIC = b"ARROW1"
_isfilebytes(bytes::Vector{UInt8}) = length(bytes) >= 6 && view(bytes, 1:6) ==
_FILE_MAGIC
-function _openbytes(bytes::Vector{UInt8})
- return _isfilebytes(bytes) ? readfile(bytes) : readstream(bytes)
+function _openbytes(bytes::Vector{UInt8}; limits::Limits=Limits())
+ return _isfilebytes(bytes) ? readfile(bytes; limits=limits) :
+ readstream(bytes; limits=limits)
end
function _opensource(path::AbstractString; mmap::Bool=true)
@@ -503,34 +505,30 @@ _sourceregions(f::ArrowFile) = AC.OwnerRegion[f.region]
function Table(source; scan::Union{Nothing,Tables.Scan}=nothing,
mmap::Bool=true)
if source isa AbstractArrowSource || source isa SourceFile
sf = source isa SourceFile ? source : SourceFile(source)
- # A stream-format object has no footer to plan from: read it whole
- # (its tail window is already in hand) and scan after decode.
- _isfilesource(sf) || return Table(_wholeobject(sf); scan=scan)
- # The schema up front (from the cached tail): literal lowering,
- # exactly-once output conversion, and DataAPI metadata all need it.
- sch, rfields = _sourceschema(sf)
- theScan = scan === nothing ? Tables.Scan() : scan
- if isempty(rfields)
- # A zero-field object is bytes-tiny; read it whole so the row
- # count survives the read.
- bytes = _wholeobject(sf)
- return _publicscan(
- _materialize_table(readfile(bytes; limits=sf.limits),
AC.OwnerRegion[]),
- sch,
- rfields,
- theScan,
- AC.OwnerRegion[],
- )
- end
- pushscan, pushable = _lowerscan(theScan, rfields)
- if pushable
- got = Tables.scan(sf, pushscan)
- return _wrapscanned(got, sch, rfields, theScan)
+ # Range planning pays off only for a pushable scan over a file-format
+ # object with columns. With a scan the schema comes up front from
+ # the cached tail: literal lowering, exactly-once output conversion,
+ # and DataAPI metadata all need it.
+ if scan !== nothing && _isfilesource(sf)
+ sch, rfields = _sourceschema(sf)
+ if !isempty(rfields)
+ pushscan, pushable = _lowerscan(scan, rfields)
+ pushable &&
+ return _wrapscanned(Tables.scan(sf, pushscan), sch,
rfields, scan)
+ end
end
- full = _wrapscanned(Tables.scan(sf, Tables.Scan()), sch, rfields,
Tables.Scan())
- return _publicscan(full, sch, rfields, theScan, AC.OwnerRegion[])
+ # No scan, a stream-format object (no footer to plan from), a
+ # zero-field file (bytes-tiny; the whole read carries its row count),
+ # or a scan that cannot be pushed down: read the object whole — its
+ # tail window is already in hand — and proceed as with bytes.
+ return _tablefrom(_openbytes(_wholeobject(sf); limits=sf.limits), scan)
end
- src = _opensource(source; mmap=mmap)
+ return _tablefrom(_opensource(source; mmap=mmap), scan)
+end
+
+# A Table from an opened IPC source: the whole thing, or a scan pushed
+# where the format can prove it and evaluated over the rest.
+function _tablefrom(src::Union{IPCStream,ArrowFile},
scan::Union{Nothing,Tables.Scan})
regions = _sourceregions(src)
fields = _corefields(src)
scan === nothing && return _materialize_table(src, regions)
diff --git a/test/cloudstore_tests.jl b/test/cloudstore_tests.jl
index 1e419ee..172233a 100644
--- a/test/cloudstore_tests.jl
+++ b/test/cloudstore_tests.jl
@@ -44,13 +44,12 @@ using CloudBase.CloudTest: Minio
CloudStore.put(bucket, "t.arrows", streambytes;
credentials=credentials)
obj = CloudStore.Object(bucket, "t.arrow"; credentials=credentials)
@test Arrow.sourcelength(ext.CloudObjectSource(obj)) ==
length(filebytes)
- # one range, and the batched form, byte-exact against the object
+ # one range, byte-exact against the object; a bounded concurrency
src = ext.CloudObjectSource(obj)
@test Arrow.readrange(src, 8, 16) == filebytes[9:24]
- @test Arrow.readranges(
- src,
- NTuple{2,Int64}[(0, 6), (100, 50), (length(filebytes) - 6, 6)],
- ) == [filebytes[1:6], filebytes[101:150], filebytes[(end - 5):end]]
+ @test Arrow.readrange(src, length(filebytes) - 6, 6) == filebytes[(end
- 5):end]
+ @test Arrow.readrange(src, 0, 0) == UInt8[]
+ @test Arrow.concurrentreads(src) == ext.CONCURRENT_RANGE_READS > 1
# a projected, windowed scan over the object
t = Arrow.Table(obj; scan=Tables.Scan(select=(:b,), limit=3,
offset=1500))
@test Tables.columnnames(t) == [:b]
@@ -68,6 +67,15 @@ using CloudBase.CloudTest: Minio
@test st.a == mem.a && st.b == mem.b
@test [length(batch.a) for batch in Arrow.Stream(sobj)] == [1000, 1000]
@test [length(batch.a) for batch in Arrow.Stream(obj)] == [1000, 1000]
+ # A handle is pinned to the object version it was made from: after the
+ # key is overwritten, its next range read fails (If-Match) rather than
+ # mixing bytes of two versions across request rounds.
+ stale = ext.CloudObjectSource(obj)
+ CloudStore.put(bucket, "t.arrow", streambytes; credentials=credentials)
+ @test_throws Exception Arrow.readrange(stale, 8, 16)
+ @test_throws Exception Arrow.Table(obj; scan=Tables.Scan(select=(:a,)))
+ fresh = CloudStore.Object(bucket, "t.arrow"; credentials=credentials)
+ @test Arrow.Table(fresh).a == mem.a
end
end
diff --git a/test/facade_tests.jl b/test/facade_tests.jl
index 099d018..eb91a6a 100644
--- a/test/facade_tests.jl
+++ b/test/facade_tests.jl
@@ -42,6 +42,61 @@ function Arrow.readrange(s::_MeteredSource, off, len)
s.fetched[] += len
return s.data[(off + 1):(off + len)]
end
+# A source that logs every request, and a concurrent one whose reads finish
+# in reverse request order (the last-issued read of a round completes first)
+# while a counter records the most reads ever in flight.
+struct _LoggingSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+ requests::Vector{NTuple{2,Int64}}
+end
+Arrow.sourcelength(s::_LoggingSource) = length(s.data)
+function Arrow.readrange(s::_LoggingSource, off, len)
+ push!(s.requests, (Int64(off), Int64(len)))
+ return s.data[(off + 1):(off + len)]
+end
+mutable struct _ConcurrentSource <: Arrow.AbstractArrowSource
+ const data::Vector{UInt8}
+ const limit::Int
+ @atomic inflight::Int
+ @atomic peak::Int
+ @atomic issued::Int
+end
+_ConcurrentSource(data, limit) = _ConcurrentSource(data, limit, 0, 0, 0)
+Arrow.sourcelength(s::_ConcurrentSource) = length(s.data)
+Arrow.concurrentreads(s::_ConcurrentSource) = s.limit
+function Arrow.readrange(s::_ConcurrentSource, off, len)
+ n = @atomic s.inflight += 1
+ while true
+ p = @atomic s.peak
+ (n <= p || (@atomicreplace s.peak p => n).success) && break
+ end
+ order = @atomic s.issued += 1
+ # Later requests of a round return sooner: results must land by request.
+ sleep(0.002 * max(0, s.limit - (order % s.limit)))
+ @atomic s.inflight -= 1
+ return s.data[(off + 1):(off + len)]
+end
+# Sources that violate the contract in each way the reader must refuse.
+struct _ShortSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+end
+Arrow.sourcelength(s::_ShortSource) = length(s.data)
+Arrow.readrange(s::_ShortSource, off, len) = s.data[(off + 1):(off + max(0,
len - 1))]
+struct _LongSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+end
+Arrow.sourcelength(s::_LongSource) = length(s.data)
+Arrow.readrange(s::_LongSource, off, len) = vcat(s.data[(off + 1):(off +
len)], 0x00)
+struct _WrongTypeSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+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
+end
+Arrow.sourcelength(s::_BadLengthSource) = s.reported
+Arrow.readrange(s::_BadLengthSource, off, len) = zeros(UInt8, len)
const MIXED = (
ints=Int64[1, 2, 3, 4],
@@ -172,6 +227,64 @@ end
@test t.b == ["v1501", "v1502", "v1503"]
# The first batch is skipped entirely and column :a is never fetched.
@test fetched[] < length(fb) ÷ 2
+
+ # Request rounds through the public path over an object larger than
+ # the tail window: a pushable scan is the tail, the surviving batch's
+ # metadata, then the selected buffers (three rounds; here one request
+ # each with coalescing), the tail read exactly once; ranges inside the
+ # cached tail window are served from it without a request.
+ wide = [string("v", i, "-", repeat("x", 60)) for i = 1:2000]
+ wio = IOBuffer()
+ Arrow.write(
+ wio,
+ Tables.partitioner([
+ (a=collect(Int64, 1:1000), b=wide[1:1000]),
+ (a=collect(Int64, 1001:2000), b=wide[1001:2000]),
+ ]),
+ )
+ wb = take!(wio)
+ @test length(wb) > 65536
+ log = _LoggingSource(wb, NTuple{2,Int64}[])
+ t2 = Arrow.Table(log; scan=Tables.Scan(select=(:a,),
filter=Tables.col(:a) < 10))
+ @test t2.a == collect(1:9)
+ tailreq = (Int64(length(wb) - 65536), Int64(65536))
+ @test count(==(tailreq), log.requests) == 1
+ @test length(log.requests) == 3
+ # An object no larger than the tail window is entirely in hand after
+ # the tail read: every planned range is served from it.
+ small = _LoggingSource(fb, NTuple{2,Int64}[])
+ t3 = Arrow.Table(small; scan=Tables.Scan(select=(:a,), limit=3,
offset=1500))
+ @test t3.a == [1501, 1502, 1503]
+ @test small.requests == [(Int64(0), Int64(length(fb)))]
+ # No scan (and any unpushable scan): the object is read whole — the
+ # cached tail plus the prefix, two requests, no planning.
+ empty!(log.requests)
+ tw = Arrow.Table(log)
+ @test tw.a == 1:2000 && length(log.requests) == 2
+ @test sum(last, log.requests) == length(wb)
+ empty!(log.requests)
+ tz = Arrow.Table(log; scan=Tables.Scan(select=()))
+ @test length(tz) == 2000 && length(log.requests) == 2
+
+ # Concurrent reads: bounded by the source's limit, results placed by
+ # request even though later requests complete first.
+ cs = _ConcurrentSource(fb, 3)
+ cf = Arrow.SourceFile(cs; tailbytes=1024, coalesce_gap=0)
+ tc = Arrow.Table(cf; scan=Tables.Scan(select=(:a, :b)))
+ @test tc.a == 1:2000 && tc.b == [string("v", i) for i = 1:2000]
+ @test 1 < (@atomic cs.peak) <= 3
+
+ # Contract violations fail closed with ValidationError, never a
+ # wrong table or a stray error type.
+ for bad in (_ShortSource(fb), _LongSource(fb), _WrongTypeSource(fb))
+ @test_throws Arrow.AC.ValidationError Arrow.Table(
+ bad;
+ scan=Tables.Scan(select=(:a,)),
+ )
+ end
+ for reported in (-1, Int128(typemax(Int64)) + 1,
Int128(typemin(Int64)) - 1)
+ @test_throws Arrow.AC.ValidationError
Arrow.Table(_BadLengthSource(reported))
+ end
end
@testset "mmap path and close!" begin
diff --git a/test/scan_battery.jl b/test/scan_battery.jl
index 21e0576..e7de2a5 100644
--- a/test/scan_battery.jl
+++ b/test/scan_battery.jl
@@ -603,14 +603,18 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
# Coalescing: an infinite gap merges every body range into one request;
# a zero gap issues more, smaller requests; both agree with the truth.
+ # (A small tail window, so the ranges are planned rather than served
+ # from the cached tail.)
logbig, srcbig = countingsource(filebytes)
gotbig = Tables.scan(
- SourceFile(srcbig; coalesce_gap=typemax(Int32)),
+ SourceFile(srcbig; tailbytes=256, coalesce_gap=typemax(Int32)),
Tables.Scan(select=(:ints, :strs)),
)
logzero, srczero = countingsource(filebytes)
- gotzero =
- Tables.scan(SourceFile(srczero; coalesce_gap=0),
Tables.Scan(select=(:ints, :strs)))
+ gotzero = Tables.scan(
+ SourceFile(srczero; tailbytes=256, coalesce_gap=0),
+ Tables.Scan(select=(:ints, :strs)),
+ )
want = Tables.scan(full, Tables.Scan(select=(:ints, :strs)))
@assert _tables_equal(gotbig, want) && _tables_equal(gotzero, want)
@assert logbig.requests < logzero.requests
