Chunked + concurrent get_range for long ranges (#33)

CHANGELOG.md

@@ -8,6 +8,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **Chunked + concurrent `get_range`** (#33). New `chunk` kwarg (a
+  `Dates.Period`, e.g. `Year(1)`) splits `[start_dt, end_dt)` into half-open
+  chunks fetched concurrently under a semaphore (`concurrency` kwarg, default
+  8), hiding the ~25–30s fixed per-request server-side assembly latency that
+  made long continuous-symbol pulls take hours sequentially. Records are
+  concatenated in time order; chunks that fail after retries are warned about
+  and recorded in the new `DBNStore.failed_ranges` field for targeted retry
+  (the call throws only if every chunk failed). `chunk` requires an explicit
+  `end_dt` and calendar endpoints, and is incompatible with `limit`.
+  `DBNStore` gained the `failed_ranges` field (always empty outside chunked
+  mode); the 2-argument constructors are unchanged.
 - **`BentoTimeoutError`.** HTTP read timeouts are now mapped to a `BentoError`
   subtype whose message names the remedy (raise `Historical(timeout = ...)` or
   reduce the range) instead of surfacing HTTP.jl's bare

docs/src/guide/historical.md

@@ -46,6 +46,44 @@ metadata header with a vector of decoded records. Convert it with
 [`to_dataframe`](@ref), [`to_csv`](@ref), [`to_json`](@ref),
 [`to_parquet`](@ref), or write it back to disk with [`to_file`](@ref).
 
+## Chunked, concurrent long-range fetches
+
+Each `get_range` request carries a fixed server-side assembly latency — for
+continuous-symbol queries, ~25–30s regardless of payload size. A multi-year
+pull as one request risks the read timeout; as sequential per-year requests it
+accumulates the fixed latency a hundred times over. Pass `chunk` to split the
+range and fetch chunks concurrently (up to `concurrency`, default 8):
+
+```julia
+store = get_range(client;
+    dataset  = "GLBX.MDP3",
+    schema   = Schema.STATISTICS,
+    symbols  = ["ES.n.0", "ZT.n.0", "ZF.n.0", "ZN.n.0"],
+    stype_in = SType.CONTINUOUS,
+    start_dt = Date(2010, 1, 1),
+    end_dt   = Date(2025, 1, 1),
+    chunk    = Year(1))
+```
+
+Records come back concatenated in time order. Chunks that fail after the
+client's usual retries are warned about and recorded in `store.failed_ranges`
+rather than sinking the whole call (it only throws if every chunk failed), so
+a retry loop is one line per range:
+
+```julia
+for (s, e) in store.failed_ranges
+    retry_store = get_range(client; dataset = "GLBX.MDP3", schema = Schema.STATISTICS,
+                            symbols = ["ES.n.0"], stype_in = SType.CONTINUOUS,
+                            start_dt = s, end_dt = e)
+    append!(store.records, retry_store.records)
+end
+```
+
+`chunk` requires an explicit `end_dt` and calendar endpoints
+(`DateTime`/`Date`/parseable string), and is incompatible with `limit`. For
+multi-GB pulls, [`submit_job`](@ref) → [`batch_download`](@ref) remains the
+better tool.
+
 ## Streaming records: `foreach_record`
 
 For larger queries (millions of records), don't materialize the whole vector.

docs/src/performance.md

@@ -54,6 +54,9 @@ bookkeeping regardless of element type — so the win is in dispatch, not GC.
 
 - **Historical, small query (≤ 1 GB):** [`get_range`](@ref) → [`DBNStore`](@ref)
   → conversion helpers. Easy and ergonomic.
+- **Historical, long range / many fixed-latency requests:** [`get_range`](@ref)
+  with `chunk = Year(1)` (or similar) — chunks fetch concurrently, hiding the
+  ~25–30s per-request server-side assembly latency.
 - **Historical, large query:** [`foreach_record`](@ref) with a typed callback,
   or [`submit_job`](@ref) + [`batch_download`](@ref) and decode the resulting
   files with `DBN.foreach_record` directly.

src/historical/timeseries.jl

@@ -1,7 +1,8 @@
 """
     get_range(client; dataset, schema, symbols, start_dt, end_dt=nothing,
               stype_in=SType.RAW_SYMBOL, stype_out=SType.INSTRUMENT_ID,
-              limit=nothing, typed=true, size_hint=nothing)
+              limit=nothing, typed=true, size_hint=nothing,
+              chunk=nothing, concurrency=8)
 
 Fetch a time range of records from the Databento Historical API. Returns a [`DBNStore`](@ref).
 
@@ -31,6 +32,33 @@ exact hint avoids that waste.
     that exceeds the client's read timeout (default 600s) the call raises
     [`BentoTimeoutError`](@ref) — construct `Historical(timeout = ...)` with a
     larger budget or reduce the range.
+
+# Chunked, concurrent fetching
+
+Every `get_range` request carries a fixed server-side assembly latency that for
+continuous-symbol queries can run ~25–30s regardless of payload size, so a
+long range fetched as one request risks the read timeout, and fetched as
+sequential chunks accumulates an hour of fixed latency over a hundred calls.
+Pass `chunk` (a `Dates.Period`, e.g. `Year(1)` or `Month(1)`) to split
+`[start_dt, end_dt)` into half-open chunks fetched concurrently — at most
+`concurrency` (default 8) requests in flight at once:
+
+```julia
+store = get_range(client; dataset = "GLBX.MDP3", schema = Schema.STATISTICS,
+                  symbols = ["ES.n.0"], stype_in = SType.CONTINUOUS,
+                  start_dt = Date(2010, 1, 1), end_dt = Date(2025, 1, 1),
+                  chunk = Year(1))
+```
+
+Records are concatenated in time order. A chunk that fails after the client's
+usual retries is warned about and recorded in the returned store's
+`failed_ranges` field (a `Vector{Tuple{DateTime,DateTime}}`) so you can
+re-fetch exactly the missing ranges; the call only throws if *every* chunk
+failed. Constraints: `chunk` requires an explicit `end_dt`, `start_dt`/`end_dt`
+as `DateTime`/`Date`/parseable strings (not unix-ns integers), and is
+incompatible with `limit` (a record cap can't be apportioned across chunks);
+`size_hint` is ignored in chunked mode. For multi-GB pulls `submit_job` remains
+the better tool.
 """
 function get_range(c::Historical;
                    dataset::AbstractString,
@@ -42,7 +70,29 @@ function get_range(c::Historical;
                    stype_out::SType.T = SType.INSTRUMENT_ID,
                    limit::Union{Nothing,Integer} = nothing,
                    typed::Bool = true,
-                   size_hint::Union{Nothing,Integer} = nothing)
+                   size_hint::Union{Nothing,Integer} = nothing,
+                   chunk::Union{Nothing,Dates.Period} = nothing,
+                   concurrency::Integer = 8)
+    chunk === nothing &&
+        return _get_range_single(c; dataset, schema, symbols, start_dt, end_dt,
+                                 stype_in, stype_out, limit, typed, size_hint)
+    return _get_range_chunked(c; dataset, schema, symbols, start_dt, end_dt,
+                              stype_in, stype_out, limit, typed, chunk, concurrency)
+end
+
+# The original single-request fetch: one HTTP GET, buffer the zstd payload,
+# decode in memory. Chunked mode calls this once per chunk.
+function _get_range_single(c::Historical;
+                           dataset::AbstractString,
+                           schema::Schema.T,
+                           symbols,
+                           start_dt,
+                           end_dt,
+                           stype_in::SType.T,
+                           stype_out::SType.T,
+                           limit::Union{Nothing,Integer},
+                           typed::Bool,
+                           size_hint::Union{Nothing,Integer})
     query = (
         dataset     = String(dataset),
         symbols     = symbols_str(symbols),
@@ -72,6 +122,132 @@ function get_range(c::Historical;
     end
 end
 
+# Normalize a chunk endpoint to DateTime. Chunk boundaries are computed with
+# Period arithmetic, so the endpoints must be calendar values — unix-ns
+# integers are accepted by the single-request path (ts_str passes them
+# through) but can't be split without assuming a timezone, so reject them.
+_chunk_datetime(d::DateTime) = d
+_chunk_datetime(d::Date) = DateTime(d)
+_chunk_datetime(s::AbstractString) =
+    try
+        DateTime(s)
+    catch
+        DateTime(Date(s))
+    end
+_chunk_datetime(x) =
+    throw(ArgumentError("chunked get_range requires start_dt/end_dt as DateTime, " *
+                        "Date, or a parseable string; got $(typeof(x))"))
+
+# Merge per-chunk response metadata into one header for the combined store.
+# Identity fields (version/dataset/schema/stypes/ts_out) are taken from the
+# first success; the time range spans all chunks; the symbol-resolution
+# vectors are unioned because they legitimately differ per date-chunk under
+# continuous/parent symbology (a contract that rolls mid-range maps
+# differently in different chunks).
+function _merge_chunk_metadata(mds::Vector{DBN.Metadata})
+    md1 = first(mds)
+    end_ts_vals = Int64[md.end_ts for md in mds if md.end_ts !== nothing]
+    return DBN.Metadata(
+        md1.version, md1.dataset, md1.schema,
+        minimum(md.start_ts for md in mds),
+        isempty(end_ts_vals) ? nothing : maximum(end_ts_vals),
+        nothing,   # a per-chunk limit would be meaningless for the union
+        md1.stype_in, md1.stype_out, md1.ts_out,
+        unique(reduce(vcat, (md.symbols for md in mds))),
+        unique(reduce(vcat, (md.partial for md in mds))),
+        unique(reduce(vcat, (md.not_found for md in mds))),
+        unique(reduce(vcat, (md.mappings for md in mds))),
+    )
+end
+
+function _get_range_chunked(c::Historical;
+                            dataset::AbstractString,
+                            schema::Schema.T,
+                            symbols,
+                            start_dt,
+                            end_dt,
+                            stype_in::SType.T,
+                            stype_out::SType.T,
+                            limit::Union{Nothing,Integer},
+                            typed::Bool,
+                            chunk::Dates.Period,
+                            concurrency::Integer)
+    limit === nothing ||
+        throw(ArgumentError("limit is not supported with chunk: a record cap cannot " *
+                            "be apportioned across chunk requests"))
+    end_dt === nothing &&
+        throw(ArgumentError("chunked get_range requires an explicit end_dt"))
+    concurrency >= 1 || throw(ArgumentError("concurrency must be ≥ 1"))
+
+    start = _chunk_datetime(start_dt)
+    stop  = _chunk_datetime(end_dt)
+    start < stop || throw(ArgumentError("start_dt must be before end_dt"))
+
+    # Half-open [t, min(t+chunk, stop)) chunks compose exactly: Databento
+    # ranges are themselves [start, end), so no record is duplicated or lost
+    # at the seams and the concatenation needs no dedup. Positivity is checked
+    # at *every* boundary, not just the first: calendar arithmetic means a
+    # period's effective length varies by date, so a single upfront check
+    # cannot rule out a later non-advancing step (which would otherwise append
+    # zero-length chunks forever).
+    chunks = Tuple{DateTime,DateTime}[]
+    let t = start
+        while t < stop
+            nxt = min(t + chunk, stop)
+            nxt > t || throw(ArgumentError(
+                "chunk must advance time at every boundary; $chunk does not advance from $t"))
+            push!(chunks, (t, nxt))
+            t = nxt
+        end
+    end
+
+    # Fetch concurrently, gated by a semaphore. Each task writes only its own
+    # index slot, so no lock is needed and chunk order (= time order) is
+    # preserved for assembly. HTTPClient is safe to share across tasks, and
+    # its built-in 429 retry-with-backoff absorbs rate-limit brushes from the
+    # concurrency. Threads.@spawn (vs @async) lets the CPU-bound zstd+decode
+    # work parallelize when threads are available; on a single-threaded
+    # runtime it still overlaps the network waits.
+    sem = Base.Semaphore(concurrency)
+    results = Vector{Any}(nothing, length(chunks))
+    @sync for (i, (cs, ce)) in pairs(chunks)
+        Threads.@spawn Base.acquire(sem) do
+            results[i] = try
+                _get_range_single(c; dataset, schema, symbols,
+                                  start_dt = cs, end_dt = ce,
+                                  stype_in, stype_out,
+                                  limit = nothing, typed, size_hint = nothing)
+            catch e
+                e isa InterruptException && rethrow()
+                hint = e isa BentoTimeoutError ? " (consider a smaller chunk)" : ""
+                @warn "get_range chunk failed; continuing with remaining chunks$hint" start_dt = cs end_dt = ce exception = e
+                e
+            end
+        end
+    end
+
+    success_idx = [i for i in eachindex(results) if results[i] isa DBNStore]
+    if isempty(success_idx)
+        # An empty store would mask total failure — surface the first error.
+        throw(first(e for e in results if e isa Exception))
+    end
+
+    R = eltype(results[first(success_idx)].records)
+    records = Vector{R}()
+    sizehint!(records, sum(i -> length(results[i].records), success_idx))
+    for i in success_idx   # index order == chunk order == time order
+        append!(records, results[i].records)
+    end
+
+    failed = Tuple{DateTime,DateTime}[chunks[i] for i in eachindex(results)
+                                      if !(results[i] isa DBNStore)]
+    isempty(failed) ||
+        @warn "get_range: $(length(failed)) of $(length(chunks)) chunks failed; their ranges are in the returned store's failed_ranges for retry"
+
+    md = _merge_chunk_metadata(DBN.Metadata[results[i].metadata for i in success_idx])
+    return DBNStore{R}(md, records, failed)
+end
+
 """
     record_type_for_schema(schema)
 

src/store.jl

@@ -8,13 +8,23 @@ When the schema is type-pure (the common case), `T` is the concrete record
 struct (e.g. `DBN.TradeMsg`) and the records vector is type-stable. For
 mixed-record streams or when `get_range` is called with `typed=false`, `T`
 falls back to the `DBN.DBNRecord` Union.
+
+`failed_ranges` is only populated by chunked `get_range` calls (the `chunk`
+kwarg): each entry is the `(start_dt, end_dt)` of a chunk whose request failed
+after retries, so the caller can re-fetch exactly the missing ranges. It is
+always empty for single-request fetches.
 """
 struct DBNStore{T}
     metadata::DBN.Metadata
     records::Vector{T}
+    failed_ranges::Vector{Tuple{DateTime,DateTime}}
 end
 
-DBNStore(metadata, records::Vector{T}) where {T} = DBNStore{T}(metadata, records)
+# 2-arg forms preserve the pre-failed_ranges construction API.
+DBNStore(metadata, records::Vector{T}) where {T} =
+    DBNStore{T}(metadata, records, Tuple{DateTime,DateTime}[])
+DBNStore{T}(metadata, records::Vector{T}) where {T} =
+    DBNStore{T}(metadata, records, Tuple{DateTime,DateTime}[])
 
 Base.length(s::DBNStore) = length(s.records)
 Base.iterate(s::DBNStore, st...) = iterate(s.records, st...)
@@ -27,7 +37,75 @@ Base.getindex(s::DBNStore, i) = getindex(s.records, i)
 function Base.show(io::IO, s::DBNStore{T}) where {T}
     print(io, "DBNStore{", T, "}(dataset=", s.metadata.dataset,
               ", schema=", s.metadata.schema,
-              ", n=", length(s.records), ")")
+              ", n=", length(s.records))
+    isempty(s.failed_ranges) || print(io, ", failed_chunks=", length(s.failed_ranges))
+    print(io, ")")
+end
+
+# Typed record loop that tolerates interleaved non-schema records.
+#
+# Historical responses are *mostly* type-pure, but the gateway can legitimately
+# interleave control records — ErrorMsg (e.g. partial continuous-symbol
+# resolution), SystemMsg, SymbolMappingMsg — with the data (#30). DBN's
+# `_foreach_record_impl` throws on the first rtype mismatch, which forced
+# callers back to `typed=false` (10× slower) and silently discarded the
+# ErrorMsg text explaining *why* data was missing. This loop keeps the
+# zero-allocation typed hot path (peek header → read straight into a reused
+# Ref{T} buffer) and handles everything else by category:
+#
+#   - ErrorMsg            → per-record @warn (rare, and the content is the
+#                           whole point: it explains gaps in the response)
+#   - SystemMsg / SymbolMappingMsg → decode + @debug (routine noise)
+#   - other known rtypes  → skip by header length, count, one summary @warn
+#   - unknown raw rtypes  → skip by header length, count, one summary @warn
+#
+# Mismatched data records get a single post-loop summary, never a per-record
+# warn — a wrong `record_type` override against a million-record stream must
+# not emit a million log lines. The summary (rather than silence) matters for
+# data-loss visibility: a skipped data rtype means the caller's type and the
+# stream disagree.
+#
+# Unlike DBN's loop, there is deliberately no `finally` that closes
+# `decoder.io` — our decoders wrap HTTP bodies / TranscodingStreams whose
+# lifecycle is owned by `open_stream` / the caller, not by this function.
+function _foreach_typed_tolerant(f, decoder::DBN.DBNDecoder, ::Type{T}) where {T}
+    expected = DBN._type_to_rtype_stream(T)
+    buffer = Ref{T}()   # reused per record: the typed read is allocation-free
+    skipped = 0
+    while !eof(decoder.io)
+        hd_result = DBN.read_record_header(decoder.io)
+        # Unknown raw rtype byte: read_record_header returns a tuple after
+        # consuming only the 2 (length, rtype) bytes; length is in 4-byte units.
+        if hd_result isa Tuple
+            _, _, len_units = hd_result
+            skip(decoder.io, Int(len_units) * DBN.LENGTH_MULTIPLIER - 2)
+            skipped += 1
+            continue
+        end
+        hd = hd_result
+        rt = hd.rtype
+        if rt == expected ||
+           (T === DBN.OHLCVMsg && (rt == DBN.RType.OHLCV_1S_MSG || rt == DBN.RType.OHLCV_1M_MSG ||
+                                   rt == DBN.RType.OHLCV_1H_MSG || rt == DBN.RType.OHLCV_1D_MSG))
+            buffer[] = DBN._read_typed_record_stream(decoder, T, hd)
+            f(buffer[])
+        elseif rt == DBN.RType.ERROR_MSG
+            rec = DBN.read_error_msg(decoder, hd)
+            @warn "Databento gateway error record interleaved in response stream" err = rec.err
+        elseif rt == DBN.RType.SYSTEM_MSG || rt == DBN.RType.SYMBOL_MAPPING_MSG
+            rec = DBN.read_record_dispatch(decoder, hd, rt)
+            @debug "Skipping control record in typed decode" rtype = rt
+        else
+            # Known data rtype that doesn't match the schema's record type.
+            # The full 16-byte header is already consumed; hd.length is in
+            # 4-byte units and covers the whole record.
+            skip(decoder.io, Int(hd.length) * DBN.LENGTH_MULTIPLIER - 16)
+            skipped += 1
+        end
+    end
+    skipped > 0 &&
+        @warn "Typed decode skipped records whose rtype did not match the schema's record type" expected_type = T skipped
+    return nothing
 end
 
 # Typed record loop that tolerates interleaved non-schema records.