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 e446f97 feat!: byte-range reads through AbstractArrowSource;
CloudStore.jl extension
e446f97 is described below
commit e446f9765f74a21c0b5f1d82b8d1d863dfa8cbdc
Author: Jacob Quinn <[email protected]>
AuthorDate: Wed Aug 19 01:10:35 2026 -0600
feat!: byte-range reads through AbstractArrowSource; CloudStore.jl extension
The byte-range read surface is a source interface plus `Tables.Scan`
pushdown on `Arrow.Table`, not a special handle:
- `Arrow.AbstractArrowSource` (src/source.jl): an implementation defines
`sourcelength(src)` and `readrange(src, offset, len)`, and may override
`readranges(src, ranges)` to issue a round's planned ranges concurrently
- `Arrow.Table(src::AbstractArrowSource; scan=…)`: the footer comes from
one tail read (cached on the handle; its trailing magic also decides
file vs stream format — a stream object is read whole), statistics and
the scan window prune batches, only the surviving batches' metadata and
the selected + filter-referenced columns' buffers are requested,
coalesced — three request rounds. The leading magic is no longer
fetched: the footer is the sole authority. `Arrow.Stream(src)` reads the
object whole
- ext/ArrowCloudStoreExt.jl (`[weakdeps] CloudStore`): a
`CloudStore.Object` is a source (its known size, one HTTP `Range` GET per
range, one task per planned range) and `Arrow.Table(obj; …)` /
`Arrow.Stream(obj; …)` accept it directly
- REMOVED from the public surface: `RangedSource`, `RangedFile`,
`fetchranges`. The planner (`SourceFile`, internal) is the same code re-
skinned: `_fetchexact`/`_fetchspans` read through the interface
Tests: the ranged battery runs over `BytesSource`/`CountingSource`
(exact ranges pinned as before); test/cloudstore_tests.jl drives the
extension end to end against CloudBase's Minio server (ranged reads,
projected/windowed and statistics-pruned scans, whole reads of file- and
stream-format objects, `Stream`); CloudStore/CloudBase are test deps.
Docs: manual remote-reads section, reference (`AbstractArrowSource`,
`sourcelength`, `readrange`, `readranges`), DESIGN §2, core-README, README.
Verified on Julia 1.10.11 and 1.12.6: Arrow 421/321 + all batteries +
extension 12/12; trim 6/6; docs; JuliaFormatter no-op; rat.
Co-Authored-By: Claude Fable 5 <[email protected]>
---
Project.toml | 7 ++
README.md | 5 +
docs/dev/DESIGN-scan-ranges-trim.md | 109 +++++++++----------
docs/dev/core-README.md | 19 ++--
docs/src/manual.md | 50 ++++++---
docs/src/reference.md | 7 +-
ext/ArrowCloudStoreExt.jl | 67 ++++++++++++
src/Arrow.jl | 7 +-
src/scan.jl | 203 +++++++++++++++++++-----------------
src/source.jl | 87 ++++++++++++++++
src/table.jl | 55 ++++++----
test/Project.toml | 2 +
test/battery_helpers.jl | 30 ++++--
test/cloudstore_tests.jl | 74 +++++++++++++
test/facade_tests.jl | 51 +++++----
test/runtests.jl | 3 +
test/scan_battery.jl | 106 +++++++++----------
17 files changed, 596 insertions(+), 286 deletions(-)
diff --git a/Project.toml b/Project.toml
index 03682aa..e43f332 100644
--- a/Project.toml
+++ b/Project.toml
@@ -37,8 +37,15 @@ TranscodingStreams = "3bb67fe8-82b1-5028-8e26-92a6c54297fa"
[sources]
ArrowStrings = {path = "src/ArrowStrings"}
+[weakdeps]
+CloudStore = "3365d9ee-d53b-4a56-812d-5344d5b716d7"
+
+[extensions]
+ArrowCloudStoreExt = "CloudStore"
+
[compat]
ArrowStrings = "0.1"
+CloudStore = "1.6"
CodecLz4 = "0.4"
DataAPI = "1"
CodecZstd = "0.8"
diff --git a/README.md b/README.md
index 17ed205..03f393c 100644
--- a/README.md
+++ b/README.md
@@ -40,10 +40,15 @@ This is a pure Julia implementation of the
- `src/ipc_read.jl`, `src/ipc_write.jl` — the IPC stream and file
formats: framing, resource limits, compression, dictionary lifecycles.
- `src/cdata.jl` — the C data and C stream interfaces, import and export.
+- `src/source.jl` — the `AbstractArrowSource` interface: what a
+ byte-range-addressable object (cloud storage, HTTP, …) provides so
+ `Arrow.Table` can fetch only the bytes a scan touches.
- `src/scan.jl` — `Tables.Scan` pushdown over byte ranges plus
footer-carried statistics pruning.
- `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.
- `src/FlatBuffers/` — the vendored FlatBuffers runtime the generated
bindings run over.
- `src/ArrowStrings/` — ArrowStrings.jl, a separate package (to be
diff --git a/docs/dev/DESIGN-scan-ranges-trim.md
b/docs/dev/DESIGN-scan-ranges-trim.md
index 9f9e99e..0d888d6 100644
--- a/docs/dev/DESIGN-scan-ranges-trim.md
+++ b/docs/dev/DESIGN-scan-ranges-trim.md
@@ -19,9 +19,9 @@
# Design: Tables.Scan pushdown, cloud byte-range reads, and the trim contract
-Implemented in `src/scan.jl` and exposed through the facade
-(`Arrow.Table(source; scan=…)`, `RangedFile`); §5 lists what is and is not
-built. The three pieces share one mechanism: **a bound column set drives
+Implemented in `src/scan.jl` and `src/source.jl` and exposed through the
+facade (`Arrow.Table(source; scan=…)` over an `AbstractArrowSource`); §5
+lists what is and is not built. The three pieces share one mechanism: **a
bound column set drives
both what gets decoded and what gets fetched, and every request value is
plain data intended to remain visible to the trim verifier.** Section 4
separates that design intent from what the trim harness actually compiles.
@@ -37,7 +37,7 @@ select/rename/type items, a closed predicate algebra
(`Cmp`/`In`/`IsNull`/
can while materializing; whatever it cannot push it hands to the generic
executor `Tables.scan(table, residual)` (Arrow's pushdown is the internal
`_applyscan(handle, scan) -> (table, residual)`, composed with the executor
-by `Tables.scan(::ArrowFile/::RangedFile, scan)` and by `Arrow.Table(source;
+by `Tables.scan(::ArrowFile/::SourceFile, scan)` and by `Arrow.Table(source;
scan=…)`). Key contract points this design leans on:
- pushed and residual work may **overlap** (inexact pruning keeps the filter
@@ -149,11 +149,13 @@ sequential — cloud-native access is a file-format
feature, stated plainly.
### The fetch protocol
-1. **Head + tail fetch** (two range requests): the eight-byte head (magic
- + padding check) and the last `tailbytes` (default 64 KiB). The tail
- covers footer-length + magic + the whole Footer in almost every real
- file; if `footerlen + 10 > tailbytes`, one exact follow-up fetch.
+1. **Tail fetch** (one range request, cached on the handle): the last
+ `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.
→ 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.
3. **Block metadata fetches**: dictionary metadata plus
`(offset, metaDataLength)` for each statistics-surviving record batch,
@@ -173,9 +175,11 @@ sequential — cloud-native access is a file-format feature,
stated plainly.
was itself derived from the verified buffer table"*: same trust story,
sparse backing.
-Request-count model (what actually matters against cloud latency): `1` head
-+ `1` tail (plus one exact Footer follow-up when the tail is too small)
-+ `⌈candidate metadata spans after coalescing⌉` + `⌈coalesced body ranges⌉`.
+Request-count model (what actually matters against cloud latency): `1` tail
+(plus one exact Footer follow-up when the tail is too small)
++ `⌈candidate metadata spans after coalescing⌉` + `⌈coalesced body ranges⌉`
+— three rounds, each one wall-clock round trip when the source issues a
+round's ranges concurrently.
For a 40-column file reading 3 columns of every batch, this moves roughly
`3/40` of the body bytes plus metadata. Statistics-pruned batches contribute
no requested metadata/body range. Coalescing is an explicit over-read policy,
@@ -184,41 +188,39 @@ gap permits it.
### The interface (no HTTP/CloudStore deps in Arrow)
-Arrow defines a minimal fetcher contract and owns the planner; transports
-live in extensions:
-
- struct RangedSource{F}
- fetch::F # fetch(offset::Int64, len::Int64) ->
Vector{UInt8}
- len::Int64 # total object length, known up front
- end
- fetchranges(s::RangedSource, ranges) -> Vector{Vector{UInt8}}
- # default: serial map over s.fetch; transports override for
- # concurrent range GETs (CloudStore does this well) — concurrency
- # stays in the extension, never in Arrow.
-
-- The entry points are `Tables.scan(RangedFile(source), scan)` and
- `Arrow.Table(RangedFile(source); scan=…)`. A
- production `readfile(::RangedSource; scan=...)` can make the existing
- whole-buffer and `mmapregion` paths trivial `RangedSource`s
- (fetch = copy/subslice), so ONE reader serves local and remote and the
- differential test is free: sparse fetch ≡ whole-file read, plus
- fetch-count/byte-count assertions on a counting test source.
-- **Why a parametric functor field and not an abstract type**: dispatch cost
- is irrelevant (IO-bound), but trim is not — an open abstract type makes
- `fetch` a dynamic call the verifier cannot resolve; a concrete `F` in a
- trimmed app is statically known. This is runtime plumbing, not a `Scan`
- value, so the no-`Function`-fields rule for plain-data requests does not
- apply to it. If extension ergonomics ever demand an abstract type, the
- fallback is a closed core ladder plus an open-only-in-extensions split;
- the functor is simpler and trim-cleaner.
-- Extension shape (not built): an `ArrowCloudStoreExt` loaded with
- CloudStore.jl would construct `RangedSource`s from S3/Azure objects with
- concurrent `fetchranges` and object-length discovery (HEAD). An
- `ArrowHTTPExt` shape is identical. Zero new hard deps either way.
+Arrow defines a minimal source contract (`src/source.jl`) and owns the
+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.
+
+- 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.
+- 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
+ `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
+ request-count/byte-count/range assertions on a counting test source
+ (`test/scan_battery.jl`), and the CloudStore extension end to end against
+ a local S3-compatible server (`test/cloudstore_tests.jl`).
- Explicitly out of scope v1, documented: caching/prefetch policy beyond
- coalescing, retries (the fetcher's job), writers over ranges, stream
- format, mutation detection (ETag pinning is the extension's concern —
- the fetcher closure can bake in `If-Match`).
+ coalescing, retries (the source's job), writers over ranges, stream
+ format (read whole), mutation detection (ETag pinning is the extension's
+ concern).
The ranged reader deliberately uses the Footer schema as its sole schema
authority. It does not parse or cross-check the leading schema message or
@@ -297,9 +299,9 @@ designed for trim but not yet gated by it. The rules in
`core-README.md`
closed algebra). The evaluator uses the same closed-set `isa` ladder
pattern as `layoutspec_of`; `OpNode` rejection keeps the set closed.
`bind` is plain data → plain data.
-- The range planner is arithmetic over `Int64`s; `RangedSource{F}` is
- concrete in any trimmed app. No dynamic registry, no abstract-typed
- fields on the hot path.
+- The range planner is arithmetic over `Int64`s; a trimmed app that names
+ its concrete `AbstractArrowSource` type resolves the source calls
+ statically. No dynamic registry on the hot path.
- **Two-tier public API (mirroring the CSV rewrite)**: the runtime-tagged
core is inherently trim-safe — descriptors are values, accessors use
literal load widths, struct scalars are `Vector{Pair{String,Any}}`. So:
@@ -331,9 +333,9 @@ Implemented (`src/scan.jl`, `src/table.jl`):
corruption-backed never-decoded proofs. `Arrow.Table(source; scan=…)`
routes through it on file-format and ranged inputs; stream-format inputs
scan post-decode with identical results.
-- **RangedSource**: the `RangedSource{F}` contract, the `RangedFile` fetch
- protocol, the coalescing planner, `SparseBody` decode, and
- counting-source proofs (zero planned body ranges for skipped columns,
+- **Byte-range sources**: the `AbstractArrowSource` contract, the
+ `SourceFile` fetch protocol, the coalescing planner, `SparseBody` decode,
+ the CloudStore.jl extension, and counting-source proofs (zero planned body
ranges for skipped columns,
window-excluded batches, and unneeded dictionary bodies, with exact
request-log checks under the fixtures' tail/coalescing settings).
- **Statistics**: `withstatistics`/`statsfile` fold the official statistics
@@ -345,8 +347,9 @@ Implemented (`src/scan.jl`, `src/table.jl`):
tail/coalescing may over-read them); acceptance pins exactness,
degradation, and both lie directions.
-Not implemented: a CloudStore/HTTP transport extension (the fetcher
-contract is the extension point), Stage B's exact facade pushdown, an
+Not implemented: an HTTP transport extension (the source contract is the
+extension point; the CloudStore extension is the model), Stage B's exact
+facade pushdown, an
encode-time `statistics=true` writer keyword, upstream-placement tracking
for statistics, and the scan-and-materialize trim harness. Scan pushdown
depends on Tables.jl's `jq/scan` branch until that API is released.
diff --git a/docs/dev/core-README.md b/docs/dev/core-README.md
index 5b4ba53..e429258 100644
--- a/docs/dev/core-README.md
+++ b/docs/dev/core-README.md
@@ -36,8 +36,10 @@ 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/scan.jl` | `Tables.Scan` pushdown over the file format, sparse
byte-range reads (`RangedSource`/`RangedFile`), embedded per-batch statistics |
+| `src/source.jl` | The `AbstractArrowSource` byte-range source interface
(`sourcelength`, `readrange`, `readranges`) |
+| `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 |
| `src/ArrowStrings/` | ArrowStrings.jl — the shared inline-else-view string
representation (`ArrowString`, `ArrowStringVector` = Utf8View memory); a
separate package, registered on its own like ArrowTypes, that Arrow depends on
through a `[sources]` path entry until its first release |
| `src/ArrowTypes/` | ArrowTypes.jl — the custom-type interface package (not
used by 3.0 yet) |
| `test/` | Core unit tests, facade tests, the four adapter acceptance
batteries, the frozen 2.x-written fixtures, the `--trim=safe` gate |
@@ -247,13 +249,14 @@ filter-referenced columns, prunes whole batches through
the embedded
statistics (one-sided: a pruned batch is provably empty; the filter always
stays in the residual), and consumes `limit`/`offset` exactly when no filter
poisons the window. Projection, filtering, renames, and type conversions
-are the generic `Tables.scan` executor's over the returned residual.
`RangedFile` runs the same
-plan over a byte-range fetcher: it uses the Footer as its sole schema
-authority, validates the full Block index and the complete metadata plan for
-every statistics-surviving record before requesting a body range, and
-requests per-buffer body ranges for exactly the decode set, coalesced under
-`coalesce_gap`. It does not parse or cross-check the leading schema message
-or the optional EOS marker; tail reads and coalescing may physically
+are the generic `Tables.scan` executor's over the returned residual.
`SourceFile` runs the same
+plan over an `AbstractArrowSource`: it uses the Footer (from one cached
+tail read) as its sole schema authority, validates the full Block index and
+the complete metadata plan for every statistics-surviving record before
+requesting a body range, and requests per-buffer body ranges for exactly
+the decode set, coalesced under `coalesce_gap`. It does not fetch the
+leading magic, parse or cross-check the leading schema message, or inspect
+the optional EOS marker; tail reads and coalescing may physically
over-read any unrequested bytes. Embedded batch statistics use the official
Arrow statistics value layout under the `JuliaArrow:batch_statistics.v1`
placement key (placement is scoped out of the upstream spec) and are
diff --git a/docs/src/manual.md b/docs/src/manual.md
index bfd185d..2105b2c 100644
--- a/docs/src/manual.md
+++ b/docs/src/manual.md
@@ -239,26 +239,42 @@ files that do not are simply scanned batch by batch.
The file format is random-access: the footer says where every batch and
buffer lives, so a reader that can fetch byte ranges — from object storage,
over HTTP, or from a local file it prefers not to map whole — needs only
-the ranges its scan touches. [`Arrow.RangedSource`](@ref) is that fetcher
-contract: a function `fetch(offset, len) -> Vector{UInt8}` over an object
-of known total length. [`Arrow.RangedFile`](@ref) wraps one with the fetch
-protocol (the eight-byte head magic, then the footer from the tail, batch
-windowing from the footer's block metadata, dictionary bodies only for the
-columns in play, coalesced body ranges for exactly the decoded columns):
+the ranges its scan touches. [`Arrow.AbstractArrowSource`](@ref) is that
+contract: a byte-addressable object of known length, read through
+[`Arrow.sourcelength`](@ref) and [`Arrow.readrange`](@ref). Given one,
+`Arrow.Table` with a scan fetches the footer from one tail read, keeps only
+the batches the footer's statistics and the scan's window allow, fetches
+those batches' metadata, and then fetches exactly the buffers of the
+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
+directly, and its planned ranges are requested concurrently:
```julia
-src = Arrow.RangedSource(Int64(objectsize)) do offset, len
- fetchbytes(url, offset, len) # your transport: S3, HTTP, ...
+using Arrow, Tables, CloudStore
+obj = CloudStore.Object(bucket, "events/2024-05.arrow"; credentials)
+tbl = Arrow.Table(obj; scan = Scan(select = (:id,), filter = col(:day) > 20))
+```
+
+Any other transport is two methods away:
+
+```julia
+struct HTTPSource <: Arrow.AbstractArrowSource
+ url::String
+ length::Int64
end
-tbl = Arrow.Table(src; scan = Scan(select = (:id,), filter = col(:day) > 20))
+Arrow.sourcelength(s::HTTPSource) = s.length
+Arrow.readrange(s::HTTPSource, offset, len) = fetchbytes(s.url, offset, len)
# a Range GET
+
+tbl = Arrow.Table(HTTPSource(url, objectsize); scan = Scan(select = (:id,)))
```
-Overriding `Arrow.fetchranges(::RangedSource, ranges)` lets a transport
-issue the planned ranges concurrently; the default fetches them serially.
-`RangedFile(src; tailbytes, coalesce_gap, limits)` tunes the initial tail
-read, how close two ranges must be to merge into one request, and the
-resource limits. Arrow.jl has no HTTP or cloud dependency of its own — a
-transport package only has to construct a `RangedSource`.
+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.
## Writing
@@ -437,8 +453,8 @@ compression, metadata — is the same in spirit, with these
differences:
over the mapped bytes; 3.0 materializes columns with concrete element
types (the mapping tables above), and the source may be released with
`Arrow.close!` at any time afterward.
-* **Scan pushdown and byte-range reads** (`Tables.Scan`, `RangedSource`,
- `RangedFile`) are new.
+* **Scan pushdown and byte-range reads** (`Tables.Scan`,
+ `AbstractArrowSource`, the CloudStore.jl extension) are new.
* **Not present in 3.0**: `Arrow.Writer`/`Arrow.append` (incremental and
append-to-file writing), multithreaded encoding (`ntasks`), the
`convert=false` lazy read mode, `Arrow.ToArrow`, and ArrowTypes.jl
diff --git a/docs/src/reference.md b/docs/src/reference.md
index a05ca76..a572a0e 100644
--- a/docs/src/reference.md
+++ b/docs/src/reference.md
@@ -39,9 +39,10 @@ Arrow.DictEncode
## Byte-range reads
```@docs
-Arrow.RangedSource
-Arrow.RangedFile
-Arrow.fetchranges
+Arrow.AbstractArrowSource
+Arrow.sourcelength
+Arrow.readrange
+Arrow.readranges
```
## The C data and C stream interfaces
diff --git a/ext/ArrowCloudStoreExt.jl b/ext/ArrowCloudStoreExt.jl
new file mode 100644
index 0000000..9f749ef
--- /dev/null
+++ b/ext/ArrowCloudStoreExt.jl
@@ -0,0 +1,67 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# 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.
+module ArrowCloudStoreExt
+
+using Arrow
+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.
+"""
+struct CloudObjectSource{O<:Object} <: Arrow.AbstractArrowSource
+ obj::O
+end
+
+Arrow.sourcelength(s::CloudObjectSource) = Int64(s.obj.size)
+
+function Arrow.readrange(s::CloudObjectSource, off, len)
+ len == 0 && return UInt8[]
+ last = off + len - 1
+ obj = s.obj
+ bytes = CloudStore.get(
+ obj.store,
+ obj.key;
+ credentials=obj.credentials,
+ headers=["Range" => "bytes=$(off)-$(last)"],
+ 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...)
+
+end # module
diff --git a/src/Arrow.jl b/src/Arrow.jl
index 5a3e644..03892dd 100644
--- a/src/Arrow.jl
+++ b/src/Arrow.jl
@@ -18,9 +18,9 @@
Arrow.jl — a pure Julia implementation of the Apache Arrow columnar format.
Public surface: `Arrow.Table` reads the IPC stream and file formats (paths,
-`IO`, byte vectors, or a `RangedSource`/`RangedFile` over a byte-range
-fetcher) as Tables.jl tables, with `Tables.Scan` pushdown; `Arrow.Stream`
-iterates a path, `IO`, or byte-vector source one record batch at a time;
+`IO`, byte vectors, or an `AbstractArrowSource` — a byte-range-addressable
+object such as one in cloud storage) as Tables.jl tables, with `Tables.Scan`
+pushdown; `Arrow.Stream` iterates a source one record batch at a time;
`Arrow.write` writes any Tables.jl source; `close!` releases mapped or
foreign storage deterministically.
@@ -91,6 +91,7 @@ const AC = ArrowCore
include("ipc_read.jl")
include("ipc_write.jl")
include("cdata.jl")
+include("source.jl")
include("scan.jl")
# The public facade.
diff --git a/src/scan.jl b/src/scan.jl
index 7c28b71..87bbf85 100644
--- a/src/scan.jl
+++ b/src/scan.jl
@@ -16,8 +16,8 @@
# =============================================================================
# Tables.Scan pushdown over the IPC file adapter, and — further down — the
-# byte-range fetch protocol (`RangedFile`/`RangedSource`) over the same
-# bound column set. Design notes: docs/dev/DESIGN-scan-ranges-trim.md.
+# byte-range fetch protocol (`SourceFile` over an `AbstractArrowSource`) over
+# the same bound column set. Design notes: docs/dev/DESIGN-scan-ranges-trim.md.
#
# Pushdown semantics: the source consumes what it can PROVE and leaves exact
# row evaluation to `Tables.scan`.
@@ -738,42 +738,89 @@ function _applyscan(f::ArrowFile, scan::Tables.Scan)
end
# ===========================================================================
-# Byte-range reads — RangedSource{F}, the planner, and sparse decode
+# Byte-range reads — SourceFile over an AbstractArrowSource, the planner,
+# and sparse decode
# ===========================================================================
"""
- RangedSource{F}
-
-The fetcher contract: `fetch(offset::Int64, len::Int64) ->
-Vector{UInt8}` over a remote or local object of known total `len`, offsets
-0-based. `F` is concrete per instantiation — in a trimmed app the fetch path
-is statically resolvable, which is why this is a parametric functor and not
-an abstract type. A transport (CloudStore, HTTP, …) only needs to construct
-one of these; `fetchranges` has a serial default it may override for
-concurrent range GETs.
+ SourceFile(src::AbstractArrowSource; limits, tailbytes=65536,
coalesce_gap=262144)
+
+The scan-driven, fetch-minimal file handle over a byte-range source:
+`Tables.scan(sf, scan)` (and `Arrow.Table(src; scan=…)`, which builds one)
+runs the fetch protocol — the footer from one tail read, batch windowing
+from block metadata, dictionary bodies only for decode-set ids, and
+per-buffer body ranges for exactly the decode set, coalesced under
+`coalesce_gap`. The source's length is read once, at construction.
+
+Trust note, stated loudly: the ranged reader treats the FOOTER as the sole
+schema authority — it does not fetch the leading magic, parse and
+cross-check the leading schema message, or inspect the optional EOS marker.
+The tail and coalesced requests may physically over-read unrequested bytes.
+The full Footer Block index and global features/message limit are checked
+up front. Per-record limits stay lazy; every surviving candidate's
+metadata-only plan is validated before any planned body range is requested.
"""
-struct RangedSource{F}
- fetch::F
+struct SourceFile{S<:AbstractArrowSource}
+ src::S
len::Int64
+ limits::Limits
+ tailbytes::Int64
+ coalesce_gap::Int64
+ # 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}}}
+end
+function SourceFile(
+ src::AbstractArrowSource;
+ limits::Limits=Limits(),
+ tailbytes::Integer=65536,
+ coalesce_gap::Integer=262144,
+)
+ gap = Int64(coalesce_gap)
+ gap >= 0 || throw(ArgumentError("negative coalesce gap"))
+ len = Int64(sourcelength(src))
+ len >= 0 || throw(ValidationError("source reports a negative length"))
+ return SourceFile(
+ src,
+ len,
+ limits,
+ Int64(max(tailbytes, 32)),
+ gap,
+ Base.RefValue{Union{Nothing,Tuple{Vector{UInt8},Int64}}}(nothing),
+ )
end
-RangedSource(bytes::Vector{UInt8}) =
- RangedSource((off, len) -> bytes[(off + 1):(off + len)],
Int64(length(bytes)))
+# The object's tail window (at most `tailbytes`, the whole object when it is
+# shorter), fetched on first use and cached on the handle.
+function _fetchtail(sf::SourceFile)
+ cached = sf.tail[]
+ cached === nothing || return cached
+ tailstart = max(Int64(0), sf.len - sf.tailbytes)
+ tail = _fetchexact(sf, tailstart, sf.len - tailstart)
+ sf.tail[] = (tail, tailstart)
+ return (tail, tailstart)
+end
-"""
- fetchranges(src::RangedSource, ranges::Vector{NTuple{2,Int64}}) ->
Vector{Vector{UInt8}}
+# 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)
+ tail, _ = _fetchtail(sf)
+ return length(tail) >= 6 && tail[(end - 5):end] ==
Vector{UInt8}(FILE_MAGIC)
+end
-One result vector per requested `(offset, len)`, in order. The default
-fetches serially through `src.fetch`; a transport overrides this method to
-issue the planned ranges concurrently.
-"""
-fetchranges(s::RangedSource, ranges::Vector{NTuple{2,Int64}}) =
- Vector{UInt8}[_fetchexact(s, off, len) for (off, len) in ranges]
+# The whole object as bytes, reusing the cached tail for its final window.
+function _wholeobject(sf::SourceFile)
+ tail, tailstart = _fetchtail(sf)
+ tailstart == 0 && return tail
+ return vcat(_fetchexact(sf, Int64(0), tailstart), tail)
+end
-function _fetchexact(s::RangedSource, off::Int64, len::Int64)
- (off >= 0 && len >= 0 && off <= s.len - len) ||
+# One exact range through the source, bounds-checked against the length
+# read at construction and length-checked on return.
+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 = s.fetch(off, len)
+ bytes = readrange(sf.src, off, len)
length(bytes) == len ||
throw(ValidationError("range fetch returned $(length(bytes)) bytes,
expected $len"))
return bytes
@@ -811,15 +858,17 @@ struct FetchedSpans
end
function _fetchspans(
- src::RangedSource,
+ sf::SourceFile,
ranges::Vector{NTuple{2,Int64}},
gap::Int64;
budget::Union{Nothing,AllocationBudget}=nothing,
what::AbstractString="range fetch",
)
spans = _coalesce(ranges, gap)
+ 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 = fetchranges(src, spans)
+ payloads = readranges(sf.src, spans)
length(payloads) == length(spans) || throw(
ValidationError(
"range fetch returned $(length(payloads)) payloads, expected
$(length(spans))",
@@ -919,54 +968,15 @@ function _parseblockmeta(
return msg, version, header_type
end
-"""
- RangedFile(src::RangedSource; limits, tailbytes=65536, coalesce_gap=262144)
-
-The scan-driven, fetch-minimal file handle: `Tables.scan(rf, scan)` (and
-`Arrow.Table(rf; scan=…)`) runs the fetch protocol — the eight-byte head
-magic then the footer from the tail, batch windowing from block metadata,
-dictionary bodies only for decode-set ids, and per-buffer body ranges for
-exactly the decode set, coalesced under `coalesce_gap`.
-
-Trust note, stated loudly: the ranged reader treats the FOOTER as the sole
-schema authority — it does not parse and cross-check the leading schema
-message or inspect the optional EOS marker. Head, tail, and coalesced requests
-may physically over-read unrequested bytes. The full Footer Block index and
-global features/message limit are checked up front. Per-record limits stay
-lazy; every surviving candidate's metadata-only plan is validated before any
-planned body range is requested.
-"""
-struct RangedFile{F}
- src::RangedSource{F}
- limits::Limits
- tailbytes::Int64
- coalesce_gap::Int64
-end
-function RangedFile(
- src::RangedSource;
- limits::Limits=Limits(),
- tailbytes::Integer=65536,
- coalesce_gap::Integer=262144,
-)
- gap = Int64(coalesce_gap)
- gap >= 0 || throw(ArgumentError("negative coalesce gap"))
- return RangedFile(src, limits, Int64(max(tailbytes, 32)), gap)
-end
-
"Fetch and verify the ranged footer: schema, fields, blocks, id table."
-function _rangedfooter(rf::RangedFile, budget::AllocationBudget)
- src = rf.src
- limits = rf.limits
+function _rangedfooter(sf::SourceFile, budget::AllocationBudget)
+ limits = sf.limits
_requirelittleendian()
_validatelimits(limits)
- L = src.len
+ L = sf.len
L >= Int64(8 + 8 + 4 + 6) ||
throw(ValidationError("file is too short to be an IPC file"))
- head = _fetchexact(src, Int64(0), Int64(8))
- head[1:6] == Vector{UInt8}(FILE_MAGIC) ||
- throw(ValidationError("missing leading ARROW1 magic"))
- tailstart = max(Int64(0), L - rf.tailbytes)
- tail = _fetchexact(src, tailstart, L - tailstart)
+ tail, tailstart = _fetchtail(sf)
tail[(end - 5):end] == Vector{UInt8}(FILE_MAGIC) ||
throw(ValidationError("missing trailing ARROW1 magic"))
footerlen = Int64(reinterpret(Int32, tail[(end - 9):(end - 6)])[1])
@@ -982,7 +992,7 @@ function _rangedfooter(rf::RangedFile,
budget::AllocationBudget)
footerbytes =
footerstart >= tailstart ?
tail[(footerstart - tailstart + 1):(footerstart - tailstart +
footerlen)] :
- _fetchexact(src, footerstart, footerlen)
+ _fetchexact(sf, footerstart, footerlen)
version, features, dictblocks, recordblocks, reserve =
verify_footer(footerbytes, limits, budget.left)
_charge!(budget, reserve, "verified footer expansion")
@@ -1042,7 +1052,7 @@ graph, body-length cross-check), header kind,
footer-version agreement,
and compression rejection.
"""
function _zerofieldblockcount(
- rf::RangedFile,
+ sf::SourceFile,
block::NTuple{3,Int64},
version::Int16,
fields::Vector{Field},
@@ -1050,36 +1060,35 @@ function _zerofieldblockcount(
)
_, metalen, bodylen = block
declared = metalen - 8
- 0 < declared <= rf.limits.max_metadata_bytes || throw(
+ 0 < declared <= sf.limits.max_metadata_bytes || throw(
ValidationError(
- "metadata length $declared outside (0,
$(rf.limits.max_metadata_bytes)]",
+ "metadata length $declared outside (0,
$(sf.limits.max_metadata_bytes)]",
),
)
- 0 <= bodylen <= rf.limits.max_body_bytes || throw(
- ValidationError("body length $bodylen outside [0,
$(rf.limits.max_body_bytes)]"),
+ 0 <= bodylen <= sf.limits.max_body_bytes || throw(
+ ValidationError("body length $bodylen outside [0,
$(sf.limits.max_body_bytes)]"),
)
_charge!(budget, metalen, "metadata range fetch")
- payload = _fetchexact(rf.src, block[1], metalen)
- msg, v, header_type = _parseblockmeta(payload, block, rf.limits, budget)
+ payload = _fetchexact(sf, block[1], metalen)
+ msg, v, header_type = _parseblockmeta(payload, block, sf.limits, budget)
header_type == UInt8(3) ||
throw(ValidationError("footer record block is not a record batch"))
v == version || throw(ValidationError("IPC metadata version changes within
the file"))
rejectexperimentalcompression(msg, v, header_type)
- return _recordbatchmeta(msg.header::Meta.RecordBatch, fields, rf.limits,
bodylen)
+ return _recordbatchmeta(msg.header::Meta.RecordBatch, fields, sf.limits,
bodylen)
end
-"Schema-only ranged read for the facade (the head magic + one tail fetch)."
-function rangedschema(rf::RangedFile)
- budget = AllocationBudget(rf.limits.max_total_allocated_bytes)
- ft = _rangedfooter(rf, budget)
+"Schema-only ranged read for the facade (one tail fetch)."
+function _sourceschema(sf::SourceFile)
+ budget = AllocationBudget(sf.limits.max_total_allocated_bytes)
+ ft = _rangedfooter(sf, budget)
return ft.sch, ft.fields
end
-function _applyscan(rf::RangedFile, scan::Tables.Scan)
- src = rf.src
- limits = rf.limits
+function _applyscan(sf::SourceFile, scan::Tables.Scan)
+ limits = sf.limits
budget = AllocationBudget(limits.max_total_allocated_bytes)
- ft = _rangedfooter(rf, budget)
+ ft = _rangedfooter(sf, budget)
fields = ft.fields
dictids = ft.dictids
fielddictids = ft.fielddictids
@@ -1117,7 +1126,7 @@ function _applyscan(rf::RangedFile, scan::Tables.Scan)
keep = _zerofieldpredicate(scan.filter)
n = _zerofieldwindow(
(
- _zerofieldblockcount(rf, block, version, fields, budget) for
+ _zerofieldblockcount(sf, block, version, fields, budget) for
block in recordblocks
),
keep,
@@ -1176,9 +1185,9 @@ function _applyscan(rf::RangedFile, scan::Tables.Scan)
)
end
metaspans = _fetchspans(
- src,
+ sf,
NTuple{2,Int64}[(bl[1], bl[2]) for bl in metablocks],
- rf.coalesce_gap;
+ sf.coalesce_gap;
budget=budget,
what="metadata range fetch",
)
@@ -1261,12 +1270,12 @@ function _applyscan(rf::RangedFile, scan::Tables.Scan)
try
if !isempty(wanted_dict)
bodyspans = _fetchspans(
- src,
+ sf,
NTuple{2,Int64}[
(dictblocks[i][1] + dictblocks[i][2], dictblocks[i][3]) for
i in wanted_dict
],
- rf.coalesce_gap,
+ sf.coalesce_gap,
)
for i in wanted_dict
block = dictblocks[i]
@@ -1347,7 +1356,7 @@ function _applyscan(rf::RangedFile, scan::Tables.Scan)
NTuple{2,Int64}[(bodystart + off, len) for (off, len) in
wants],
)
end
- bodyspans = _fetchspans(src, bodyranges, rf.coalesce_gap)
+ bodyspans = _fetchspans(sf, bodyranges, sf.coalesce_gap)
parts = Dict{Int,Vector{Any}}(idx => Any[] for idx in decodeidx)
outrows = 0
@@ -1398,7 +1407,7 @@ end
"""
Tables.scan(f::ArrowFile, scan)
- Tables.scan(rf::RangedFile, scan)
+ Tables.scan(sf::SourceFile, scan)
Scan an Arrow file handle: push down what the file format can prove
(`_applyscan` — column pruning, statistics batch pruning, exact
@@ -1406,7 +1415,7 @@ limit/offset windows) and hand the residual to the
generic `Tables.scan`
executor, whose semantics the pushdown must agree with. `Arrow.Table(source;
scan=…)` is the public entry over the same path.
"""
-function Tables.scan(f::Union{ArrowFile,RangedFile}, scan::Tables.Scan)
+function Tables.scan(f::Union{ArrowFile,SourceFile}, scan::Tables.Scan)
table, residual = _applyscan(f, scan)
return Tables.scan(table, residual)
end
diff --git a/src/source.jl b/src/source.jl
new file mode 100644
index 0000000..7373728
--- /dev/null
+++ b/src/source.jl
@@ -0,0 +1,87 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# =============================================================================
+# The byte-range source interface: what a remote (or any byte-addressable)
+# object must provide for `Arrow.Table` to read it with exact range requests.
+# =============================================================================
+
+"""
+ Arrow.AbstractArrowSource
+
+A byte-addressable object of known length — an object in cloud storage, an
+HTTP resource, an in-memory buffer — that [`Arrow.Table`](@ref) reads with
+exact byte-range requests instead of downloading whole. With a
+`Tables.Scan`, the file-format footer is fetched from the tail, the record
+batches are pruned by the footer's statistics and the scan's window, and only
+the buffers of the selected (and filter-referenced) columns are requested,
+coalesced into a few range reads.
+
+An implementation defines two methods:
+
+ Arrow.sourcelength(src)::Integer # total length in bytes
+ Arrow.readrange(src, offset, len)::Vector{UInt8} # `len` bytes from
0-based `offset`
+
+and may override
+
+ Arrow.readranges(src, ranges::Vector{NTuple{2,Int64}}) ->
Vector{Vector{UInt8}}
+
+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.
+
+```julia
+struct BytesSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+end
+Arrow.sourcelength(s::BytesSource) = length(s.data)
+Arrow.readrange(s::BytesSource, offset, len) = s.data[(offset + 1):(offset +
len)]
+
+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
+no footer and are read whole.
+"""
+abstract type AbstractArrowSource end
+
+"""
+ Arrow.sourcelength(src::AbstractArrowSource) -> Integer
+
+The source's total length in bytes. Required of every implementation.
+"""
+function sourcelength end
+
+"""
+ Arrow.readrange(src::AbstractArrowSource, offset, len) -> Vector{UInt8}
+
+`len` bytes starting at 0-based `offset`; the result must have exactly `len`
+elements. Required of every implementation.
+"""
+function readrange end
+
+"""
+ Arrow.readranges(src::AbstractArrowSource,
ranges::Vector{NTuple{2,Int64}}) -> Vector{Vector{UInt8}}
+
+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.
+"""
+readranges(src::AbstractArrowSource, ranges::Vector{NTuple{2,Int64}}) =
+ Vector{UInt8}[readrange(src, off, len) for (off, len) in ranges]
diff --git a/src/table.jl b/src/table.jl
index 3226cf8..bb67f1a 100644
--- a/src/table.jl
+++ b/src/table.jl
@@ -30,17 +30,20 @@
Arrow.Table(source; scan=nothing, mmap=true) -> Table
Read Arrow IPC data as Tables.jl columns. `source` is a file path, an `IO`,
-raw bytes (`Vector{UInt8}`), or an `Arrow.RangedSource`/`Arrow.RangedFile`
-(byte-range reads — see their docs). Both IPC formats are accepted: the file
-format (`ARROW1` magic, random access, footer statistics) and the stream
-format. `mmap=true` memory-maps a file-format path instead of reading it
-into memory; it has no effect on the other source kinds.
+raw bytes (`Vector{UInt8}`), or an [`Arrow.AbstractArrowSource`](@ref) — a
+byte-range-addressable object, such as one in cloud storage, read with
+exact range requests. Both IPC formats are accepted: the file format
+(`ARROW1` magic, random access, footer statistics) and the stream format.
+`mmap=true` memory-maps a file-format path instead of reading it into
+memory; it has no effect on the other source kinds.
`scan` is a `Tables.Scan` pushdown request: only the selected and
-filter-referenced columns are decoded, footer statistics prune batches no row
of which can match the
-filter, and exact limit/offset windows skip whole batches. On the file
-format (and ranged sources) pruning happens before bytes are fetched or
-decoded; on the stream format the scan is applied after decode.
+filter-referenced columns are decoded, footer statistics prune batches no
+row of which can match the filter, and exact limit/offset windows skip
+whole batches. On the file format pruning happens before bytes are decoded,
+and over an `AbstractArrowSource` before they are even fetched — the footer
+comes from one tail read and only the surviving batches' selected buffers
+are requested; on the stream format the scan is applied after decode.
One exception: a scan that cannot run in the storage domain — a filter
literal with no exact storage representation (a cross-domain or
@@ -473,6 +476,10 @@ end
_opensource(io::IO; mmap::Bool=true) = _openbytes(Base.read(io))
_opensource(bytes::Vector{UInt8}; mmap::Bool=true) = _openbytes(bytes)
_opensource(src::Union{IPCStream,ArrowFile}; mmap::Bool=true) = src
+# A byte-range source is read whole: iteration is sequential over every
+# batch, so there is nothing for range planning to skip.
+_opensource(src::AbstractArrowSource; mmap::Bool=true) =
+ _openbytes(_wholeobject(SourceFile(src)))
"Distinct owner regions reachable from a source's decoded batches."
function _sourceregions(s::IPCStream)
@@ -494,18 +501,21 @@ _sourceregions(f::ArrowFile) = AC.OwnerRegion[f.region]
# --- Table construction ------------------------------------------------------
function Table(source; scan::Union{Nothing,Tables.Scan}=nothing,
mmap::Bool=true)
- if source isa RangedSource || source isa RangedFile
- rf = source isa RangedSource ? RangedFile(source) : source
- # One extra tail fetch buys the schema up front: literal lowering,
+ 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 = rangedschema(rf)
+ sch, rfields = _sourceschema(sf)
theScan = scan === nothing ? Tables.Scan() : scan
if isempty(rfields)
- # A zero-field object is bytes-tiny; fetch it whole so the row
+ # A zero-field object is bytes-tiny; read it whole so the row
# count survives the read.
- bytes = _fetchexact(rf.src, Int64(0), rf.src.len)
+ bytes = _wholeobject(sf)
return _publicscan(
- _materialize_table(readfile(bytes; limits=rf.limits),
AC.OwnerRegion[]),
+ _materialize_table(readfile(bytes; limits=sf.limits),
AC.OwnerRegion[]),
sch,
rfields,
theScan,
@@ -514,10 +524,10 @@ function Table(source;
scan::Union{Nothing,Tables.Scan}=nothing, mmap::Bool=true
end
pushscan, pushable = _lowerscan(theScan, rfields)
if pushable
- got = Tables.scan(rf, pushscan)
+ got = Tables.scan(sf, pushscan)
return _wrapscanned(got, sch, rfields, theScan)
end
- full = _wrapscanned(Tables.scan(rf, Tables.Scan()), sch, rfields,
Tables.Scan())
+ full = _wrapscanned(Tables.scan(sf, Tables.Scan()), sch, rfields,
Tables.Scan())
return _publicscan(full, sch, rfields, theScan, AC.OwnerRegion[])
end
src = _opensource(source; mmap=mmap)
@@ -792,10 +802,11 @@ end
"""
Arrow.Stream(source; mmap=true)
-Iterate an IPC source (a file path, an `IO`, or a `Vector{UInt8}`) one
-record batch at a time; each iteration yields an [`Arrow.Table`](@ref) for
-that batch. `mmap=true` memory-maps a file-format path instead of reading
-it into memory; it has no effect on the other source kinds. Satisfies
`Tables.partitions` (each
+Iterate an IPC source (a file path, an `IO`, a `Vector{UInt8}`, or an
+[`Arrow.AbstractArrowSource`](@ref), which is read whole) one record batch
+at a time; each iteration yields an [`Arrow.Table`](@ref) for that batch.
+`mmap=true` memory-maps a file-format path instead of reading it into
+memory; it has no effect on the other source kinds. Satisfies
`Tables.partitions` (each
batch is one partition), so partition-aware sinks — including `Arrow.write`,
which writes one record batch per partition — see the source batch structure.
diff --git a/test/Project.toml b/test/Project.toml
index eb360a1..4b31cbe 100644
--- a/test/Project.toml
+++ b/test/Project.toml
@@ -16,6 +16,8 @@
[deps]
Base64 = "2a0f44e3-6c83-55bd-87e4-b1978d98bd5f"
+CloudBase = "85eb1798-d7c4-4918-bb13-c944d38e27ed"
+CloudStore = "3365d9ee-d53b-4a56-812d-5344d5b716d7"
DataAPI = "9a962f9c-6df0-11e9-0e5d-c546b8b5ee8a"
Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
ArrowStrings = "c38d8858-22a2-449e-9eca-ef92ec15f353"
diff --git a/test/battery_helpers.jl b/test/battery_helpers.jl
index 8370d6f..9a6a5d9 100644
--- a/test/battery_helpers.jl
+++ b/test/battery_helpers.jl
@@ -773,9 +773,15 @@ _viewlong(len::Int, prefix::Vector{UInt8}, bufidx::Int,
off::Int) = vcat(
)
# ---------------------------------------------------------------------------
-# Byte-range fetch accounting for the ranged-scan battery: a RangedSource
-# over in-memory bytes that logs every range and byte it is asked for.
+# Byte-range sources over in-memory bytes for the ranged-scan battery: the
+# plain one, and one that logs every range and byte it is asked for.
# ---------------------------------------------------------------------------
+struct BytesSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+end
+Arrow.sourcelength(s::BytesSource) = length(s.data)
+Arrow.readrange(s::BytesSource, off, len) = s.data[(off + 1):(off + len)]
+
mutable struct FetchLog
requests::Int
bytes::Int64
@@ -783,15 +789,21 @@ mutable struct FetchLog
end
FetchLog() = FetchLog(0, 0, NTuple{2,Int64}[])
+struct CountingSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+ log::FetchLog
+end
+Arrow.sourcelength(s::CountingSource) = length(s.data)
+function Arrow.readrange(s::CountingSource, off, len)
+ s.log.requests += 1
+ s.log.bytes += len
+ push!(s.log.ranges, (off, len))
+ return s.data[(off + 1):(off + len)]
+end
+
function countingsource(bytes::Vector{UInt8})
log = FetchLog()
- fetch = (off, len) -> begin
- log.requests += 1
- log.bytes += len
- push!(log.ranges, (off, len))
- bytes[(off + 1):(off + len)]
- end
- return log, RangedSource(fetch, Int64(length(bytes)))
+ return log, CountingSource(bytes, log)
end
_fetched(log::FetchLog, pos::Int64) =
diff --git a/test/cloudstore_tests.jl b/test/cloudstore_tests.jl
new file mode 100644
index 0000000..1e419ee
--- /dev/null
+++ b/test/cloudstore_tests.jl
@@ -0,0 +1,74 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# The CloudStore.jl extension end to end against a local S3-compatible
+# server (CloudBase's Minio harness): a `CloudStore.Object` is an
+# `Arrow.AbstractArrowSource`, so `Arrow.Table(obj; scan=…)` reads through
+# HTTP Range requests, and the whole-object paths agree with in-memory reads.
+
+module CloudStoreTests
+
+using Test
+using Tables
+using Arrow
+using CloudStore
+using CloudBase.CloudTest: Minio
+
+@testset "CloudStore extension" begin
+ ext = Base.get_extension(Arrow, :ArrowCloudStoreExt)
+ @test ext !== nothing
+ part1 = (a=collect(Int64, 1:1000), b=["v$i" for i = 1:1000], c=rand(1000))
+ part2 = (a=collect(Int64, 1001:2000), b=["v$i" for i = 1001:2000],
c=rand(1000))
+ fio = IOBuffer()
+ Arrow.write(fio, Tables.partitioner([part1, part2]))
+ filebytes = take!(fio)
+ sio = IOBuffer()
+ Arrow.write(sio, Tables.partitioner([part1, part2]); file=false)
+ streambytes = take!(sio)
+ Minio.with() do conf
+ credentials, bucket = conf.credentials, conf.store
+ CloudStore.put(bucket, "t.arrow", filebytes; credentials=credentials)
+ 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
+ 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]]
+ # 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]
+ @test t.b == ["v1501", "v1502", "v1503"]
+ # a filter that the footer statistics can prune to the second batch
+ t2 = Arrow.Table(obj; scan=Tables.Scan(select=(:a,),
filter=Tables.col(:a) > 1990))
+ @test t2.a == collect(1991:2000)
+ # whole-object reads agree with the in-memory reads
+ full = Arrow.Table(obj)
+ mem = Arrow.Table(filebytes)
+ @test Tables.columnnames(full) == Tables.columnnames(mem)
+ @test full.a == mem.a && full.b == mem.b && full.c == mem.c
+ sobj = CloudStore.Object(bucket, "t.arrows"; credentials=credentials)
+ st = Arrow.Table(sobj)
+ @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]
+ end
+end
+
+end # module
diff --git a/test/facade_tests.jl b/test/facade_tests.jl
index ad23030..099d018 100644
--- a/test/facade_tests.jl
+++ b/test/facade_tests.jl
@@ -27,6 +27,22 @@ import DataAPI
using Arrow
using ArrowStrings
+# In-memory byte-range sources: the plain one, and one that meters bytes.
+struct _BytesSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+end
+Arrow.sourcelength(s::_BytesSource) = length(s.data)
+Arrow.readrange(s::_BytesSource, off, len) = s.data[(off + 1):(off + len)]
+struct _MeteredSource <: Arrow.AbstractArrowSource
+ data::Vector{UInt8}
+ fetched::Base.RefValue{Int64}
+end
+Arrow.sourcelength(s::_MeteredSource) = length(s.data)
+function Arrow.readrange(s::_MeteredSource, off, len)
+ s.fetched[] += len
+ return s.data[(off + 1):(off + len)]
+end
+
const MIXED = (
ints=Int64[1, 2, 3, 4],
floats=[1.5, missing, 3.5, 4.5],
@@ -150,11 +166,8 @@ end
)
fb = take!(io)
fetched = Ref(Int64(0))
- src = Arrow.RangedSource(
- (off, len) -> (fetched[] += len; fb[(off + 1):(off + len)]),
- Int64(length(fb)),
- )
- rf = Arrow.RangedFile(src; tailbytes=1024, coalesce_gap=0)
+ src = _MeteredSource(fb, fetched)
+ rf = Arrow.SourceFile(src; tailbytes=1024, coalesce_gap=0)
t = Arrow.Table(rf; scan=Tables.Scan(select=(:b,), limit=3,
offset=1500))
@test t.b == ["v1501", "v1502", "v1503"]
# The first batch is skipped entirely and column :a is never fetched.
@@ -270,7 +283,7 @@ end
metadata=Dict("origin" => "ranged"),
)
fb = take!(io)
- src = Arrow.RangedSource(fb)
+ src = _BytesSource(fb)
t = Arrow.Table(src)
@test t.stamp == [DateTime(2020, 5, 5)]
@test eltype(t.stamp) == Union{Missing,DateTime} || eltype(t.stamp) ==
DateTime
@@ -562,7 +575,7 @@ end
@testset "zero-field counts across all paths" begin
sch = Arrow.AC.Schema(Arrow.AC.Field[])
bytes = Arrow.writefile(sch, [Arrow.AC.RecordBatch(sch,
Arrow.AC.ArrayData[], 3)])
- for source in (bytes, Arrow.RangedSource(bytes))
+ for source in (bytes, _BytesSource(bytes))
t = Arrow.Table(source; scan=Tables.Scan())
@test Tables.rowcount(t) == 3
tw = Arrow.Table(source; scan=Tables.Scan(limit=1, offset=1))
@@ -594,11 +607,8 @@ end
zb;
scan=Tables.Scan(filter=Tables.coleq(Tables.col(:nope), 1)),
)
- # ranged zero-field honors RangedFile limits
- rfz = Arrow.RangedFile(
- Arrow.RangedSource(zb);
- limits=Arrow.Limits(max_array_length=2),
- )
+ # ranged zero-field honors SourceFile limits
+ rfz = Arrow.SourceFile(_BytesSource(zb);
limits=Arrow.Limits(max_array_length=2))
@test_throws Arrow.AC.ValidationError Arrow.Table(rfz)
end
@@ -707,7 +717,7 @@ end
)
got = Tables.scan(af, scan)
@test Tables.rowcount(Tables.columns(got)) == want
- rgot = Tables.scan(Arrow.RangedFile(Arrow.RangedSource(zb)), scan)
+ rgot = Tables.scan(Arrow.SourceFile(_BytesSource(zb)), scan)
@test Tables.rowcount(Tables.columns(rgot)) == want
end
# The facade path allocates nothing proportional to a hostile count:
@@ -727,7 +737,7 @@ end
bad[65:68] .= 0x00
@test_throws Arrow.AC.ValidationError Arrow.readfile(copy(bad))
@test_throws Arrow.AC.ValidationError Tables.scan(
- Arrow.RangedFile(Arrow.RangedSource(copy(bad))),
+ Arrow.SourceFile(_BytesSource(copy(bad))),
Tables.Scan(),
)
# Header reads share ONE cumulative budget, as Limits documents:
@@ -742,7 +752,7 @@ end
Tables.Scan(),
)
@test_throws Arrow.AllocationLimitError Tables.scan(
- Arrow.RangedFile(Arrow.RangedSource(copy(many)); limits=tight),
+ Arrow.SourceFile(_BytesSource(copy(many)); limits=tight),
Tables.Scan(),
)
# A zero-field schema declares no dictionary ids: a footer listing a
@@ -775,13 +785,13 @@ end
append!(doctored, Arrow.FILE_MAGIC)
@test_throws Arrow.AC.ValidationError Arrow.readfile(copy(doctored))
@test_throws Arrow.AC.ValidationError Tables.scan(
- Arrow.RangedFile(Arrow.RangedSource(copy(doctored))),
+ Arrow.SourceFile(_BytesSource(copy(doctored))),
Tables.Scan(),
)
# Structural binding is unconditional: an unsupported predicate node
# rejects even with validate=false, on every facade path.
zs = Arrow.writestream(sch, [Arrow.AC.RecordBatch(sch,
Arrow.AC.ArrayData[], 3)])
- for source in (zb, zs, Arrow.RangedSource(zb))
+ for source in (zb, zs, _BytesSource(zb))
@test_throws ArgumentError Arrow.Table(
source;
scan=Tables.Scan(filter=Tables.OpNode(:custom, Any[]),
validate=false),
@@ -827,8 +837,7 @@ end
ios = IOBuffer()
Arrow.write(ios, (l=rows,); file=false)
sbb = take!(ios)
- for src in
- (Arrow.Table(fbb), Arrow.Table(sbb),
Arrow.Table(Arrow.RangedSource(fbb)))
+ for src in (Arrow.Table(fbb), Arrow.Table(sbb),
Arrow.Table(_BytesSource(fbb)))
for file in (true, false)
out = IOBuffer()
Arrow.write(out, src; file=file)
@@ -1213,7 +1222,7 @@ end
@test isequal(Arrow.Table(bytes).x, [missing, 7])
@test eltype(Arrow.Table(bytes).x) === Union{Missing,Int64}
@test isequal(Arrow.Table(bytes; scan=Tables.Scan()).x, [missing, 7])
- for handle in (Arrow.readfile(bytes),
Arrow.RangedFile(Arrow.RangedSource(bytes)))
+ for handle in (Arrow.readfile(bytes),
Arrow.SourceFile(_BytesSource(bytes)))
got = Tables.scan(handle, Tables.Scan())
@test isequal(got.x, [missing, 7])
@test eltype(got.x) === Union{Missing,Int64}
@@ -1227,7 +1236,7 @@ end
sch2 = AC.Schema([ref])
bytes2 = Arrow.writefile(sch2, AC.RecordBatch[AC.RecordBatch(sch2,
[red], 2)])
@test isequal(Arrow.Table(bytes2).r, [missing, 7])
- for handle in (Arrow.readfile(bytes2),
Arrow.RangedFile(Arrow.RangedSource(bytes2)))
+ for handle in (Arrow.readfile(bytes2),
Arrow.SourceFile(_BytesSource(bytes2)))
@test isequal(Tables.scan(handle, Tables.Scan()).r, [missing, 7])
end
# a dictionary column: a null-free pool under a non-nullable field is
diff --git a/test/runtests.jl b/test/runtests.jl
index a432ad8..1291de5 100644
--- a/test/runtests.jl
+++ b/test/runtests.jl
@@ -22,6 +22,9 @@ include("core_tests.jl")
# The public facade (Arrow.Table / Arrow.Stream / Arrow.write).
include("facade_tests.jl")
+# The CloudStore.jl extension against a local S3-compatible server.
+include("cloudstore_tests.jl")
+
# The adapter acceptance batteries: assertion-dense scripts over the
# package's internals, sharing one module that aliases the package
# namespace wholesale.
diff --git a/test/scan_battery.jl b/test/scan_battery.jl
index 7fad2a8..21e0576 100644
--- a/test/scan_battery.jl
+++ b/test/scan_battery.jl
@@ -198,7 +198,7 @@ function _scan_main()
Tables.Scan(offset=10_000),
Tables.Scan(filter=Tables.col(:ints) > 10_000),
)
- for handle in (af, RangedFile(RangedSource(copy(filebytes))))
+ for handle in (af, SourceFile(BytesSource(copy(filebytes))))
got = Tables.scan(handle, emptyscan)
@assert Tables.rowcount(Tables.columns(got)) == 0
@assert Tables.schema(got) == fullschema sprint(show, emptyscan)
@@ -218,7 +218,7 @@ function _scan_main()
# the source or residualized.
extreme = Tables.Scan(select=(:ints,), offset=typemax(Int),
limit=typemax(Int))
extremewant = Tables.scan(full, extreme)
- for sourcefile in (af, RangedFile(RangedSource(filebytes)))
+ for sourcefile in (af, SourceFile(BytesSource(filebytes)))
got = Tables.scan(sourcefile, extreme)
@assert _tables_equal(got, extremewant)
@assert length(Tables.getcolumn(Tables.columns(got), 1)) == 0
@@ -312,7 +312,7 @@ function _scan_main()
copyto!(badrows, block[1] + 9, meta, 1, length(meta))
shifted = Tables.Scan(select=(:x,), offset=5, limit=1)
@assert _rejects(() -> Tables.scan(readfile(copy(badrows)), shifted))
- @assert _rejects(() ->
Tables.scan(RangedFile(RangedSource(copy(badrows))), shifted))
+ @assert _rejects(() -> Tables.scan(SourceFile(BytesSource(copy(badrows))),
shifted))
println("window row counts require top-level FieldNode agreement ✓")
# Checked buffer-span addition is required before a zero-row window may
@@ -358,7 +358,7 @@ function _scan_main()
end
@assert overflowed
@assert _rejects(
- () -> Tables.scan(RangedFile(RangedSource(copy(ovbytes))),
Tables.Scan(limit=0)),
+ () -> Tables.scan(SourceFile(BytesSource(copy(ovbytes))),
Tables.Scan(limit=0)),
)
println("overflowing variadic buffer spans reject before window exclusion
✓")
@@ -371,7 +371,7 @@ function _scan_main()
AC.RecordBatch(zerosch, ArrayData[], 2),
]
zerobytes = writefile(zerosch, zerobatches)
- for source in (readfile(copy(zerobytes)),
RangedFile(RangedSource(copy(zerobytes))))
+ for source in (readfile(copy(zerobytes)),
SourceFile(BytesSource(copy(zerobytes))))
got = Tables.scan(source, Tables.Scan())
@assert isempty(Tables.columnnames(Tables.columns(got)))
@assert Tables.rowcount(Tables.columns(got)) == 5
@@ -397,14 +397,14 @@ function _scan_main()
edgelimits = Limits(max_array_length=typemax(Int64))
for source in (
readfile(copy(edgebytes); limits=edgelimits),
- RangedFile(RangedSource(copy(edgebytes)); limits=edgelimits),
+ SourceFile(BytesSource(copy(edgebytes)); limits=edgelimits),
)
got = Tables.scan(source, Tables.Scan())
@assert Tables.rowcount(Tables.columns(got)) == typemax(Int)
end
for source in (
readfile(copy(overflowbytes); limits=edgelimits),
- RangedFile(RangedSource(copy(overflowbytes)); limits=edgelimits),
+ SourceFile(BytesSource(copy(overflowbytes)); limits=edgelimits),
)
empty = Tables.scan(source, Tables.Scan(limit=0))
@assert Tables.rowcount(Tables.columns(empty)) == 0
@@ -418,7 +418,7 @@ function _scan_main()
@assert _rejects(() -> _fulltable(readfile(copy(overflowbytes);
limits=edgelimits)))
for source in (
readfile(copy(sentinelbytes); limits=edgelimits),
- RangedFile(RangedSource(copy(sentinelbytes)); limits=edgelimits),
+ SourceFile(BytesSource(copy(sentinelbytes)); limits=edgelimits),
)
@assert _rejects(() -> Tables.scan(source, Tables.Scan(offset=1)))
end
@@ -445,7 +445,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
]
for scan in scans
log, src = countingsource(filebytes)
- got = Tables.scan(RangedFile(src), scan)
+ got = Tables.scan(SourceFile(src), scan)
want = Tables.scan(full, scan)
@assert _tables_equal(got, want) sprint(show, scan)
end
@@ -473,10 +473,10 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
),
)
logall, srcall = countingsource(bigbytes)
- Tables.scan(RangedFile(srcall; tailbytes=256, coalesce_gap=64),
Tables.Scan())
+ Tables.scan(SourceFile(srcall; tailbytes=256, coalesce_gap=64),
Tables.Scan())
logone, srcone = countingsource(bigbytes)
Tables.scan(
- RangedFile(srcone; tailbytes=256, coalesce_gap=64),
+ SourceFile(srcone; tailbytes=256, coalesce_gap=64),
Tables.Scan(select=(:a,)),
)
@assert logone.bytes < logall.bytes ÷ 4 (logone.bytes, logall.bytes)
@@ -493,13 +493,13 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
corrupt[(off + 5):(off + 8)] .= reinterpret(UInt8, Int32[Int32(2) ^ 30])
logc, srcc = countingsource(corrupt)
got = Tables.scan(
- RangedFile(srcc; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcc; tailbytes=256, coalesce_gap=0),
Tables.Scan(select=(:ints,)),
)
@assert isequal(collect(Any, got.ints), collect(Any, full.ints))
@assert !_fetched(logc, off + 5)
@assert _rejects(
- () -> Tables.scan(RangedFile(RangedSource(corrupt)),
Tables.Scan(select=(:strs,))),
+ () -> Tables.scan(SourceFile(BytesSource(corrupt)),
Tables.Scan(select=(:strs,))),
)
println(
"skipped columns add no planned body range " *
@@ -512,7 +512,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
body2 = (block2[1] + block2[2], block2[3])
logw, srcw = countingsource(filebytes)
Tables.scan(
- RangedFile(srcw; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcw; tailbytes=256, coalesce_gap=0),
Tables.Scan(select=(:strs,), limit=5),
)
@assert !any(_fetched(logw, body2[1] + k) for k = 0:8:(body2[2] - 1))
@@ -531,19 +531,19 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
dictblockbody = (dictblock[1] + dictblock[2], dictblock[3])
lognod, srcnod = countingsource(filebytes)
Tables.scan(
- RangedFile(srcnod; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcnod; tailbytes=256, coalesce_gap=0),
Tables.Scan(select=(:ints,)),
)
@assert !any(_fetched(lognod, dictblockbody[1] + k) for k =
0:8:(dictblockbody[2] - 1))
logd, srcd = countingsource(filebytes)
Tables.scan(
- RangedFile(srcd; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcd; tailbytes=256, coalesce_gap=0),
Tables.Scan(select=(:dict,)),
)
@assert any(_fetched(logd, dictblockbody[1] + k) for k =
0:8:(dictblockbody[2] - 1))
logd0, srcd0 = countingsource(filebytes)
Tables.scan(
- RangedFile(srcd0; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcd0; tailbytes=256, coalesce_gap=0),
Tables.Scan(select=(:dict,), limit=0),
)
@assert !any(_fetched(logd0, dictblockbody[1] + k) for k =
0:8:(dictblockbody[2] - 1))
@@ -574,7 +574,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
dvsch = Schema(Field[dvf, dvtailf])
dvbytes = writefile(dvsch, [AC.RecordBatch(dvsch, ArrayData[dvd, dvtaild],
3)])
dvgot = Tables.scan(
- RangedFile(RangedSource(copy(dvbytes))),
+ SourceFile(BytesSource(copy(dvbytes))),
Tables.Scan(select=(:dictview, :tail)),
)
@assert collect(Any, dvgot.dictview) == Any["a", "view", "a"]
@@ -596,7 +596,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logmissing, srcmissing = countingsource(missingdict)
@assert _rejects(
() ->
- Tables.scan(RangedFile(srcmissing; tailbytes=32, coalesce_gap=0),
missingscan),
+ Tables.scan(SourceFile(srcmissing; tailbytes=32, coalesce_gap=0),
missingscan),
)
@assert !any(_fetched(logmissing, block[1] + block[2]) for block in
missingrecords)
println("missing dictionary plans reject before dedicated record-body
requests ✓")
@@ -605,12 +605,12 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
# a zero gap issues more, smaller requests; both agree with the truth.
logbig, srcbig = countingsource(filebytes)
gotbig = Tables.scan(
- RangedFile(srcbig; coalesce_gap=typemax(Int32)),
+ SourceFile(srcbig; coalesce_gap=typemax(Int32)),
Tables.Scan(select=(:ints, :strs)),
)
logzero, srczero = countingsource(filebytes)
gotzero =
- Tables.scan(RangedFile(srczero; coalesce_gap=0),
Tables.Scan(select=(:ints, :strs)))
+ Tables.scan(SourceFile(srczero; 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
@@ -630,7 +630,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
# A tail smaller than the footer forces the exact follow-up fetch.
logt, srct = countingsource(filebytes)
- gott = Tables.scan(RangedFile(srct; tailbytes=32),
Tables.Scan(select=(:ints,)))
+ gott = Tables.scan(SourceFile(srct; tailbytes=32),
Tables.Scan(select=(:ints,)))
@assert isequal(collect(Any, gott.ints), collect(Any, full.ints))
println("undersized tails recover with one exact footer fetch ✓")
@@ -654,7 +654,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
zfull = _fulltable(readfile(copy(zbytes)))
logz, srcz = countingsource(zbytes)
gotz = Tables.scan(
- RangedFile(srcz; tailbytes=256, coalesce_gap=64),
+ SourceFile(srcz; tailbytes=256, coalesce_gap=64),
Tables.Scan(select=(:x,)),
)
@assert isequal(collect(Any, gotz.x), collect(Any, zfull.x))
@@ -673,13 +673,13 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logfixed, srcfixed = countingsource(badfixed)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcfixed; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcfixed; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:ints,)),
),
)
@assert !_fetched(logfixed, fixedoff)
skipped = Tables.scan(
- RangedFile(RangedSource(copy(badfixed)); tailbytes=32, coalesce_gap=0),
+ SourceFile(BytesSource(copy(badfixed)); tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:floats,)),
)
@assert isequal(collect(Any, skipped.floats), collect(Any, full.floats))
@@ -698,7 +698,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logvalid, srcvalid = countingsource(badvalid)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcvalid; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcvalid; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:x,)),
),
)
@@ -792,7 +792,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logemptyoffset, srcemptyoffset = countingsource(bademptyoffset)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcemptyoffset; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcemptyoffset; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:x,)),
),
)
@@ -803,7 +803,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logoffsets, srcoffsets = countingsource(badoffsets)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcoffsets; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcoffsets; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:strs,)),
),
)
@@ -814,7 +814,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logstruct, srcstruct = countingsource(badstruct)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcstruct; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcstruct; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:structs,)),
),
)
@@ -850,7 +850,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logsparse, srcsparse = countingsource(broken)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcsparse; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcsparse; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:u,)),
),
)
@@ -865,7 +865,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logcompressed, srccompressed = countingsource(badcompressed)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srccompressed; tailbytes=32, coalesce_gap=0),
+ SourceFile(srccompressed; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:x,)),
),
)
@@ -876,7 +876,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
logbaddict, srcbaddict = countingsource(baddict)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srcbaddict; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcbaddict; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:dict,)),
),
)
@@ -884,7 +884,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
_fetched(logbaddict, dictblockbody[1] + k) for k =
0:8:(dictblockbody[2] - 1)
)
skippeddict = Tables.scan(
- RangedFile(RangedSource(copy(baddict)); tailbytes=32, coalesce_gap=0),
+ SourceFile(BytesSource(copy(baddict)); tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:ints,)),
)
@assert isequal(collect(Any, skippeddict.ints), collect(Any, full.ints))
@@ -893,7 +893,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
windowpos, _ = _bufferposition(filebytes, 2, 2)
logwindow, srcwindow = countingsource(badwindow)
windowed = Tables.scan(
- RangedFile(srcwindow; tailbytes=32, coalesce_gap=0),
+ SourceFile(srcwindow; tailbytes=32, coalesce_gap=0),
Tables.Scan(select=(:ints,), limit=5),
)
@assert isequal(collect(Any, windowed.ints), collect(Any, full.ints[1:5]))
@@ -907,7 +907,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
@assert _rejects(() -> Tables.scan(readfile(copy(legacyv4)), legacyscan))
loglegacy, srclegacy = countingsource(legacyv4)
@assert _rejects(
- () -> Tables.scan(RangedFile(srclegacy; tailbytes=32, coalesce_gap=0),
legacyscan),
+ () -> Tables.scan(SourceFile(srclegacy; tailbytes=32, coalesce_gap=0),
legacyscan),
)
@assert !_fetched(loglegacy, legacyblock[1] + legacyblock[2])
println("legacy compression rejects before dedicated record-body requests
✓")
@@ -917,9 +917,9 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
badlen = copy(filebytes)
lenpos = length(badlen) - 9
badlen[lenpos:(lenpos + 3)] .= reinterpret(UInt8, Int32[Int32(2) ^ 30])
- @assert _rejects(() -> Tables.scan(RangedFile(RangedSource(badlen)),
Tables.Scan()))
+ @assert _rejects(() -> Tables.scan(SourceFile(BytesSource(badlen)),
Tables.Scan()))
@assert _rejects(
- () -> Tables.scan(RangedFile(RangedSource(filebytes[1:20])),
Tables.Scan()),
+ () -> Tables.scan(SourceFile(BytesSource(filebytes[1:20])),
Tables.Scan()),
)
overlap = copy(filebytes)
@@ -935,7 +935,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
_write_i64!(footerbytes, recordstart + 40, firstblock[3])
copyto!(overlap, footerstart + 1, footerbytes, 1, length(footerbytes))
@assert _rejects(() -> readfile(copy(overlap)))
- @assert _rejects(() -> Tables.scan(RangedFile(RangedSource(overlap)),
Tables.Scan()))
+ @assert _rejects(() -> Tables.scan(SourceFile(BytesSource(overlap)),
Tables.Scan()))
zerobuffer = copy(filebytes)
block = af.recordblocks[1]
@@ -948,7 +948,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
@assert _rejects(() -> readfile(copy(zerobuffer)))
@assert _rejects(
() ->
- Tables.scan(RangedFile(RangedSource(zerobuffer)),
Tables.Scan(select=(:ints,))),
+ Tables.scan(SourceFile(BytesSource(zerobuffer)),
Tables.Scan(select=(:ints,))),
)
println("forged footers and truncated objects fail closed ✓")
@@ -956,13 +956,13 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
# Scan also keeps one aggregate budget across every batch it decompresses.
@assert _rejects(
() -> Tables.scan(
- RangedFile(RangedSource(filebytes);
limits=Limits(max_body_bytes=32)),
+ SourceFile(BytesSource(filebytes);
limits=Limits(max_body_bytes=32)),
Tables.Scan(select=(:ints,)),
),
)
@assert _rejects(
() -> Tables.scan(
- RangedFile(RangedSource(filebytes); limits=Limits(max_messages=1)),
+ SourceFile(BytesSource(filebytes); limits=Limits(max_messages=1)),
Tables.Scan(),
),
)
@@ -970,7 +970,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
intoff, _ = _bufferposition(filebytes, 1, 2)
@assert _rejects(
() -> Tables.scan(
- RangedFile(
+ SourceFile(
srclimit;
limits=Limits(max_buffer_bytes=8),
tailbytes=256,
@@ -1001,7 +1001,7 @@ function _ranged_main(filebytes::Vector{UInt8},
af::ArrowFile, full)
)
@assert _rejects(
() -> Tables.scan(
- RangedFile(RangedSource(largebytes); limits=tight),
+ SourceFile(BytesSource(largebytes); limits=tight),
Tables.Scan(select=(:x,)),
),
)
@@ -1090,7 +1090,7 @@ end
want = Tables.scan(sfull, scan)
@assert _tables_equal(Tables.scan(saf, scan), want) sprint(show, scan)
@assert _tables_equal(
- Tables.scan(RangedFile(RangedSource(copy(sbytes))), scan),
+ Tables.scan(SourceFile(BytesSource(copy(sbytes))), scan),
want,
) sprint(show, scan)
end
@@ -1129,7 +1129,7 @@ end
for scan in floatscans
want = Tables.scan(ffull, scan)
@assert _tables_equal(Tables.scan(faf, scan), want)
- @assert _tables_equal(Tables.scan(RangedFile(RangedSource(fbytes)),
scan), want)
+ @assert _tables_equal(Tables.scan(SourceFile(BytesSource(fbytes)),
scan), want)
end
println("float pruning preserves signed-zero and NaN predicate semantics
✓")
@@ -1186,7 +1186,7 @@ end
block1 = saf.recordblocks[1]
logp, srcp = countingsource(sbytes)
got = Tables.scan(
- RangedFile(srcp; tailbytes=256, coalesce_gap=0),
+ SourceFile(srcp; tailbytes=256, coalesce_gap=0),
Tables.Scan(filter=Tables.col(:x) > 7),
)
@assert isequal(collect(Any, got.x), Any[8, 9, 10])
@@ -1219,7 +1219,7 @@ end
logpruned, srcpruned = countingsource(limitbytes)
@assert isempty(
Tables.scan(
- RangedFile(srcpruned; limits=lazylimits, tailbytes=32,
coalesce_gap=0),
+ SourceFile(srcpruned; limits=lazylimits, tailbytes=32,
coalesce_gap=0),
prunedscan,
).x,
)
@@ -1231,7 +1231,7 @@ end
logkept, srckept = countingsource(limitbytes)
@assert _rejects(
() -> Tables.scan(
- RangedFile(srckept; limits=lazylimits, tailbytes=32,
coalesce_gap=0),
+ SourceFile(srckept; limits=lazylimits, tailbytes=32,
coalesce_gap=0),
keptscan,
),
)
@@ -1265,7 +1265,7 @@ end
endianness=source.schema.endianness,
)
badbytes = writefile(badsch, source.batches)
- for sourcefile in (readfile(copy(badbytes)),
RangedFile(RangedSource(badbytes)))
+ for sourcefile in (readfile(copy(badbytes)),
SourceFile(BytesSource(badbytes)))
got = Tables.scan(sourcefile, Tables.Scan(filter=Tables.col(:x) > 7))
@assert isequal(collect(Any, got.x), Any[8, 9, 10])
end
@@ -1287,7 +1287,7 @@ end
endianness=source.schema.endianness,
)
wrongbytes = writefile(wrongsch, source.batches)
- for sourcefile in (readfile(copy(wrongbytes)),
RangedFile(RangedSource(wrongbytes)))
+ for sourcefile in (readfile(copy(wrongbytes)),
SourceFile(BytesSource(wrongbytes)))
got = Tables.scan(sourcefile, Tables.Scan(filter=Tables.col(:x) > 7))
@assert isequal(collect(Any, got.x), Any[8, 9, 10])
end
@@ -1341,7 +1341,7 @@ end
tight = Limits(max_total_allocated_bytes=cap)
for sourcefile in (
readfile(copy(hugebytes); limits=tight),
- RangedFile(RangedSource(hugebytes); limits=tight),
+ SourceFile(BytesSource(hugebytes); limits=tight),
)
rejected = try
Tables.scan(sourcefile,
Tables.Scan(filter=Tables.coleq(Tables.col(:s), "x")))
@@ -1385,11 +1385,11 @@ end
wides = liarfile(Int64(-1000), Int64(1000))
narrows = liarfile(Int64(6), Int64(7))
trustscan = Tables.Scan(filter=Tables.col(:x) > 8)
- for sourcefile in (readfile(copy(wides)), RangedFile(RangedSource(wides)))
+ for sourcefile in (readfile(copy(wides)), SourceFile(BytesSource(wides)))
wide = Tables.scan(sourcefile, trustscan)
@assert isequal(collect(Any, wide.x), Any[9, 10])
end
- for sourcefile in (readfile(copy(narrows)),
RangedFile(RangedSource(narrows)))
+ for sourcefile in (readfile(copy(narrows)),
SourceFile(BytesSource(narrows)))
narrow = Tables.scan(sourcefile, trustscan)
@assert isempty(narrow.x) # rows 9, 10 silently lost: the trust
boundary
end
