Zstd manifest - #1738
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe PR adds zstd-compressed manifest storage, bounded concurrent ranged object reads, manifest-statistics-based query planning, numeric UInt64 series hashes, and Parquet bloom-filter placement configuration. ChangesManifest codec integration
Concurrent object storage
Query planning and scan observability
OTel series-hash typing
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Query
participant StandardTableProvider
participant ObjectStorage
participant DataFusion
Query->>StandardTableProvider: start scan
StandardTableProvider->>ObjectStorage: fetch manifests concurrently
ObjectStorage-->>StandardTableProvider: ordered manifest files
StandardTableProvider->>DataFusion: provide pruned files and statistics
DataFusion-->>Query: build physical plan
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (7)
src/query/stream_schema_provider.rs (2)
757-759: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winEmit scan timings before these early returns.
Both empty-manifest and hot-tier-hit paths bypass the consolidated timing event, leaving common fast paths invisible. Record the timing event before returning.
Also applies to: 778-785
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 757 - 759, Update the early-return paths in the stream schema provider, including the empty-manifest branch and hot-tier-hit branch, to emit the consolidated scan timing event before returning. Reuse the existing timing-event logic and ensure both paths record timing before calling final_plan or returning their results.
816-830: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winKeep routine query telemetry out of warning logs.
These successful hot-path events scale with query/table-scan volume and can obscure actionable warnings while increasing logging overhead.
src/query/stream_schema_provider.rs#L816-L830: lower scan timing logs toinfo!/debug!or warn only above thresholds.src/query/stream_schema_provider.rs#L616-L622: sample or lower normal manifest-pruning statistics.src/query/mod.rs#L1111-L1121: sample or lower normal cache-statistics reporting.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 816 - 830, Lower routine successful query telemetry from warning level: change the scan phase timing log in stream schema provider to info/debug or emit warn only when thresholds are exceeded, and apply the same sampling or lower severity to normal manifest-pruning statistics in src/query/stream_schema_provider.rs:616-622 and cache-statistics reporting in src/query/mod.rs:1111-1121. Preserve warning severity for genuinely actionable or anomalous conditions.src/storage/object_storage.rs (1)
327-337: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider a default implementation that forwards to
get_object.As a required method, every backend must implement it even when ranging is meaningless —
LocalFSalready does exactlyself.get_object(path, tenant_id).await. A default body on the trait would delete that override and give any future backend a correct, if unoptimised, implementation for free.♻️ Suggested default
async fn get_object_ranged( &self, path: &RelativePath, tenant_id: &Option<String>, - ) -> Result<Bytes, ObjectStorageError>; + ) -> Result<Bytes, ObjectStorageError> { + self.get_object(path, tenant_id).await + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/object_storage.rs` around lines 327 - 337, Provide a default implementation for the ObjectStorage::get_object_ranged trait method that forwards directly to self.get_object(path, tenant_id).await. Remove the redundant LocalFS override while preserving any optimized ranged implementations in other backends.src/storage/mod.rs (3)
115-138: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueRestart is immediate; consider a small backoff.
ObjectChangedmeans a writer is actively replacing the object. Restarting with zero delay makes it likely the next attempt lands in the same window, burning all three attempts within milliseconds. A short sleep (even 50–100 ms, optionally scaled byattempt) would materially raise the odds the retry succeeds. Not a correctness issue — the error path is well-formed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/mod.rs` around lines 115 - 138, Add a short asynchronous backoff before retrying after RangedReadError::ObjectChanged in the retry branch of the ranged-read loop. Sleep briefly, optionally increasing the delay with attempt, then continue to the next get_object_ranged_once call; keep the final-attempt error behavior unchanged.
79-93: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
error!fires for benignNotFoundtoo.Several callers treat a missing object as normal control flow —
ObjectStoreMetastore::get_manifestmapsNoSuchKeytoOk(None), andget_overviewsswallows it with.ok(). Because this helper logs unconditionally aterror!before the caller classifies the failure, every optional-object probe will emit a full error record with a source chain. DowngradingNotFoundtodebug!/warn!keeps the transport diagnostics you want without the false alarms.🔊 Suggested severity split
- tracing::error!( - op, - path = %path, - chain = ?chain, - debug = ?err, - "object store error" - ); + if matches!(err, object_store::Error::NotFound { .. }) { + tracing::debug!(op, path = %path, "object store: not found"); + } else { + tracing::error!( + op, + path = %path, + chain = ?chain, + debug = ?err, + "object store error" + ); + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/mod.rs` around lines 79 - 93, Update log_object_store_error to classify object_store::Error::NotFound failures separately and emit them at debug or warn severity, while retaining error! with the existing source-chain diagnostics for all other errors.
52-63: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueProbe range (64 MiB) is 4× the chunk size, which partly defeats the stated goal.
The rationale for ranging is that "every stream short lived" lets
object_storeretry one chunk instead of the object. But the first request can stream up to 64 MiB in a single body — larger than any subsequent chunk — so a mid-transfer reset on the probe still discards the most expensive part of the read, and this path has noObjectChangedretry either. Consider sizing the probe at or nearRANGED_GET_CHUNK_BYTESand letting the reportedmeta.sizedrive the rest; the single-request fast path for small objects is preserved either way.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/mod.rs` around lines 52 - 63, Align RANGED_GET_PROBE_BYTES with RANGED_GET_CHUNK_BYTES so the initial ranged request is no larger than subsequent chunks, while retaining the existing small-object plain-GET fast path and using meta.size to determine remaining ranges.src/storage/s3.rs (1)
949-984: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueHardcoded
GET_OBJECTS_CONCURRENCYbypasses the configured request cap.
P_MAX_OBJECT_STORE_REQUESTSalready exists as the operator-facing knob for in-flight requests, but this fan-out is fixed at 32 regardless of it. On a low cap the extra futures just queue; on a high cap 32 leaves throughput on the table. Deriving the value from the configured maximum would keep the two settings coherent. Same pattern insrc/storage/gcs.rsandsrc/storage/azure_blob.rs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/s3.rs` around lines 949 - 984, The object-fetch fan-out in the shown S3 helper hardcodes GET_OBJECTS_CONCURRENCY instead of honoring the configured request cap. Replace that buffer limit with the configured P_MAX_OBJECT_STORE_REQUESTS-derived value, preserving listing order and existing error handling; apply the same adjustment to the corresponding fetch paths in the GCS and Azure Blob helpers.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/query/mod.rs`:
- Around line 328-331: Remove the unconditional report_parquet_metadata_cache()
call from the query completion paths, including the corresponding logic around
the other referenced occurrences. Update the reporting flow to collect and sort
cache metrics periodically or through an independent throttling mechanism,
ensuring concurrent query completions do not repeatedly perform process-wide O(n
log n) work.
In `@src/query/stream_schema_provider.rs`:
- Around line 51-59: Reduce MANIFEST_FETCH_CONCURRENCY from 64 to a small limit
within the requested 4–8 range, such as 8, and preserve the existing manifest
fetch and decoding flow that uses this constant.
- Around line 590-615: Move the complete manifest pruning phase in the scan
flow—including the Rayon prune-reason calculation, count aggregation, and
parallel filtering/deallocation around manifest_files—into spawn_blocking or the
established planning pool. Await that blocking task from the async path,
preserving the existing pruning order and pruned_by_filter results while
ensuring no synchronous Rayon work runs on the Tokio worker.
In `@src/storage/gcs.rs`:
- Around line 118-138: Remove the duplicate log_object_store_error definition in
src/storage/gcs.rs:118-138 and update its callers to use
crate::storage::log_object_store_error. In src/storage/mod.rs:79-93, make the
shared helper log object_store::Error::NotFound { .. } at debug! while retaining
error! for other failures, preserving the existing full source-chain logging.
In `@src/storage/mod.rs`:
- Around line 350-351: Update REQUEST_TIMEOUT_SECS and RETRY_TIMEOUT_SECS so the
total retry budget can accommodate the configured max_retries: 5 across
120-second request attempts, including reasonable overhead. Preserve bounded
ranged-read behavior while ensuring retries are not prematurely cut off by
retry_timeout.
- Around line 189-235: Prevent the ranged-read path from running when
`first.meta.e_tag` is `None`: after obtaining the probe bytes and before
constructing `ranges`, return an appropriate error instead of issuing unpinned
range requests. Preserve the existing early return for objects whose total size
is at or below `head.len()`, and keep the ETag-pinned ranged path unchanged when
an ETag is present.
---
Nitpick comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 757-759: Update the early-return paths in the stream schema
provider, including the empty-manifest branch and hot-tier-hit branch, to emit
the consolidated scan timing event before returning. Reuse the existing
timing-event logic and ensure both paths record timing before calling final_plan
or returning their results.
- Around line 816-830: Lower routine successful query telemetry from warning
level: change the scan phase timing log in stream schema provider to info/debug
or emit warn only when thresholds are exceeded, and apply the same sampling or
lower severity to normal manifest-pruning statistics in
src/query/stream_schema_provider.rs:616-622 and cache-statistics reporting in
src/query/mod.rs:1111-1121. Preserve warning severity for genuinely actionable
or anomalous conditions.
In `@src/storage/mod.rs`:
- Around line 115-138: Add a short asynchronous backoff before retrying after
RangedReadError::ObjectChanged in the retry branch of the ranged-read loop.
Sleep briefly, optionally increasing the delay with attempt, then continue to
the next get_object_ranged_once call; keep the final-attempt error behavior
unchanged.
- Around line 79-93: Update log_object_store_error to classify
object_store::Error::NotFound failures separately and emit them at debug or warn
severity, while retaining error! with the existing source-chain diagnostics for
all other errors.
- Around line 52-63: Align RANGED_GET_PROBE_BYTES with RANGED_GET_CHUNK_BYTES so
the initial ranged request is no larger than subsequent chunks, while retaining
the existing small-object plain-GET fast path and using meta.size to determine
remaining ranges.
In `@src/storage/object_storage.rs`:
- Around line 327-337: Provide a default implementation for the
ObjectStorage::get_object_ranged trait method that forwards directly to
self.get_object(path, tenant_id).await. Remove the redundant LocalFS override
while preserving any optimized ranged implementations in other backends.
In `@src/storage/s3.rs`:
- Around line 949-984: The object-fetch fan-out in the shown S3 helper hardcodes
GET_OBJECTS_CONCURRENCY instead of honoring the configured request cap. Replace
that buffer limit with the configured P_MAX_OBJECT_STORE_REQUESTS-derived value,
preserving listing order and existing error handling; apply the same adjustment
to the corresponding fetch paths in the GCS and Azure Blob helpers.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 3406ebe6-1b0e-4b68-a732-673a12298bc5
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (16)
Cargo.tomlsrc/catalog/manifest.rssrc/cli.rssrc/metastore/metastore_traits.rssrc/metastore/metastores/object_store_metastore.rssrc/metastore/mod.rssrc/metrics/mod.rssrc/parseable/streams.rssrc/query/mod.rssrc/query/stream_schema_provider.rssrc/storage/azure_blob.rssrc/storage/gcs.rssrc/storage/localfs.rssrc/storage/mod.rssrc/storage/object_storage.rssrc/storage/s3.rs
716434d to
0a288b9
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/query/stream_schema_provider.rs (1)
757-786: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winConsolidated scan-timing log never fires on the two early-return paths.
Both early returns — no manifest files at all (Lines 757-759) and hot tier fully satisfying the query (Lines 778-786) — skip straight to
final_planwithout ever reaching thetracing::warn!("scan phase timings", ...)at Lines 816-830. Given these look like common paths (empty results, hot-tier cache hits), the new consolidated timing instrumentation this PR adds will be silent for a meaningful share of real scans, undermining the observability goal of this change.Consider emitting a (partial) timing log at each early-return site, or restructuring with a single exit point / RAII-style logger that always records whatever phases completed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 757 - 786, Ensure the scan-timing instrumentation executes before both early returns in the manifest-files and hot-tier paths. Update the surrounding flow near get_hottier_exectuion_plan and final_plan so each completed phase emits the consolidated tracing::warn!("scan phase timings", ...) data, or use a single-exit/RAII approach that records partial timings when no manifest files remain.
♻️ Duplicate comments (2)
src/query/stream_schema_provider.rs (2)
47-73: 🩺 Stability & Availability | 🟠 MajorManifest fetch concurrency still 64 — contradicts its own rationale, unresolved from prior review.
The doc comment explains manifests are hundreds of MB each and that a wide fan-out is bandwidth-bound and risks tripping the object store body timeout, yet
MANIFEST_FETCH_CONCURRENCYis64. This is the exact same constant flagged in a prior review round and it hasn't changed. The.buffered(MANIFEST_FETCH_CONCURRENCY)call at lines 541-574 inherits the same risk — up to 64 multi-hundred-MB manifests could be in flight/decoded concurrently, which can exhaust memory and saturate the object store client.Proposed fix
-const MANIFEST_FETCH_CONCURRENCY: usize = 64; +const MANIFEST_FETCH_CONCURRENCY: usize = 8;Also applies to: 538-574
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 47 - 73, Reduce MANIFEST_FETCH_CONCURRENCY from 64 to a deliberately small handful consistent with its documentation, and ensure the .buffered(MANIFEST_FETCH_CONCURRENCY) manifest-fetch pipeline uses that reduced limit. Keep the existing fetch and decoding flow unchanged.
590-615: 🩺 Stability & Availability | 🟠 MajorRayon pruning still runs synchronously on the Tokio worker — unresolved from prior review.
par_iter()/into_par_iter()here block the calling Tokio worker thread until the Rayon global pool finishes; the surrounding comments themselves describe "hundreds of milliseconds" of CPU/deallocation work on tens of thousands of files. This is the same concern raised previously (move the pruning phase throughspawn_blockingor a dedicated planning pool); the code is unchanged in that respect.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 590 - 615, The manifest pruning and discard phase around prune_reason and manifest_files still executes synchronously on the Tokio worker; move the Rayon computation and parallel drops into spawn_blocking or the established dedicated planning pool, await its result, and propagate any join errors while preserving pruning counts and file ordering.
🧹 Nitpick comments (4)
src/storage/mod.rs (2)
115-137: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueRestart-on-rewrite has no backoff.
An object being rewritten in a tight loop causes up to three immediate full re-reads (probe + all chunks) back to back, re-paying the whole transfer each time. A short sleep before retrying would cut the wasted bandwidth without changing semantics.
♻️ Suggested delay before restart
Err(RangedReadError::ObjectChanged) if attempt < RANGED_GET_MAX_ATTEMPTS => { tracing::warn!( path = %log_path, attempt, "object changed during ranged read, retrying" ); + tokio::time::sleep(std::time::Duration::from_millis( + 100 * u64::from(attempt), + )) + .await; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/mod.rs` around lines 115 - 137, Add a short asynchronous backoff before retrying after RangedReadError::ObjectChanged in the get-object retry loop, while preserving the existing retry limit and error handling. Apply the delay only when another attempt remains, before the next get_object_ranged_once call.
52-60: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueProbe size (64 MiB) is 4× the chunk size, so the one request every read makes is the least bounded.
The doc comment justifies ranging as "keeps every stream short lived", but the probe body is the only unavoidable request and is the largest. If severed-body recovery is the goal, a probe closer to
RANGED_GET_CHUNK_BYTES(with compressed manifests still fitting in one request) would be more consistent; the current split mostly optimizes for "one request for compressed manifests".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/mod.rs` around lines 52 - 60, Reduce RANGED_GET_PROBE_BYTES to a value closer to RANGED_GET_CHUNK_BYTES while still accommodating compressed manifests in a single request, and update its documentation to reflect the bounded-request rationale. Keep the subsequent range size and get_object_ranged behavior unchanged.src/storage/s3.rs (1)
910-932: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff
get_object_rangedand the reworkedget_objectsare copy-pasted verbatim into three providers. Both method bodies differ only inself.client, so any future fix (metric labels, ordering, concurrency tuning) has to be applied three times. Factor them into shared helpers insrc/storage/mod.rs— e.g. aget_object_ranged_with_metrics(client, path, tenant_id)and aget_objects_buffered(client, root, prefix, filter_func, tenant_id, fetch)— and have each provider delegate.
src/storage/s3.rs#L910-L932: replace the body with a call to the shared ranged+metrics helper.src/storage/gcs.rs#L702-L724: same delegation.src/storage/azure_blob.rs#L689-L711: same delegation.src/storage/s3.rs#L949-L984: replace the drain-then-bufferedblock with the shared listing/download helper.src/storage/gcs.rs#L740-L775: same delegation.src/storage/azure_blob.rs#L727-L762: same delegation.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/s3.rs` around lines 910 - 932, Deduplicate the provider implementations by adding shared get_object_ranged_with_metrics and get_objects_buffered helpers in src/storage/mod.rs, preserving existing metrics, listing, filtering, fetching, and buffering behavior. Delegate get_object_ranged in src/storage/s3.rs#L910-L932, src/storage/gcs.rs#L702-L724, and src/storage/azure_blob.rs#L689-L711 to the ranged helper; delegate the get_objects implementations in src/storage/s3.rs#L949-L984, src/storage/gcs.rs#L740-L775, and src/storage/azure_blob.rs#L727-L762 to the buffered helper, passing each provider’s client and existing arguments.src/query/stream_schema_provider.rs (1)
816-830: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRoutine per-scan/per-snapshot logs emitted at
warn!level. Both the consolidated scan-timing summary and the manifest-pruning summary fire unconditionally on every request, not on anomalies, but usetracing::warn!; at production query volume this floods WARN-level output that's typically wired to alerting/dashboards.
src/query/stream_schema_provider.rs#L816-L830: changetracing::warn!(...)totracing::info!ortracing::debug!for the "scan phase timings" log (or gate atwarn!only whentotal_msexceeds a threshold).src/query/stream_schema_provider.rs#L616-L622: changetracing::warn!(...)totracing::info!/tracing::debug!for the "manifest stats pruning" log, mirroring the same-severity fix.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 816 - 830, Lower the severity of both routine per-scan logs from warn to info or debug: update the “scan phase timings” tracing call around src/query/stream_schema_provider.rs lines 816-830 and the “manifest stats pruning” call around lines 616-622. Keep their fields and messages unchanged, applying the same severity consistently at both sites.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/catalog/manifest.rs`:
- Around line 122-129: Update decode_manifest_blocking to acquire an owned
permit from an Arc<Semaphore> using acquire_owned(), then move that permit into
the spawn_blocking closure alongside bytes. Keep the permit held until
decode_manifest completes, including when the awaiting future is cancelled, so
detached blocking tasks remain bounded by the semaphore.
In `@src/storage/gcs.rs`:
- Around line 331-337: Update the GET error logging in _get_object, including
the related error path around the referenced lines, to log
object_store::Error::NotFound at debug! severity while preserving error! for all
other failures. Keep the existing log context and error propagation unchanged.
---
Outside diff comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 757-786: Ensure the scan-timing instrumentation executes before
both early returns in the manifest-files and hot-tier paths. Update the
surrounding flow near get_hottier_exectuion_plan and final_plan so each
completed phase emits the consolidated tracing::warn!("scan phase timings", ...)
data, or use a single-exit/RAII approach that records partial timings when no
manifest files remain.
---
Duplicate comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 47-73: Reduce MANIFEST_FETCH_CONCURRENCY from 64 to a deliberately
small handful consistent with its documentation, and ensure the
.buffered(MANIFEST_FETCH_CONCURRENCY) manifest-fetch pipeline uses that reduced
limit. Keep the existing fetch and decoding flow unchanged.
- Around line 590-615: The manifest pruning and discard phase around
prune_reason and manifest_files still executes synchronously on the Tokio
worker; move the Rayon computation and parallel drops into spawn_blocking or the
established dedicated planning pool, await its result, and propagate any join
errors while preserving pruning counts and file ordering.
---
Nitpick comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 816-830: Lower the severity of both routine per-scan logs from
warn to info or debug: update the “scan phase timings” tracing call around
src/query/stream_schema_provider.rs lines 816-830 and the “manifest stats
pruning” call around lines 616-622. Keep their fields and messages unchanged,
applying the same severity consistently at both sites.
In `@src/storage/mod.rs`:
- Around line 115-137: Add a short asynchronous backoff before retrying after
RangedReadError::ObjectChanged in the get-object retry loop, while preserving
the existing retry limit and error handling. Apply the delay only when another
attempt remains, before the next get_object_ranged_once call.
- Around line 52-60: Reduce RANGED_GET_PROBE_BYTES to a value closer to
RANGED_GET_CHUNK_BYTES while still accommodating compressed manifests in a
single request, and update its documentation to reflect the bounded-request
rationale. Keep the subsequent range size and get_object_ranged behavior
unchanged.
In `@src/storage/s3.rs`:
- Around line 910-932: Deduplicate the provider implementations by adding shared
get_object_ranged_with_metrics and get_objects_buffered helpers in
src/storage/mod.rs, preserving existing metrics, listing, filtering, fetching,
and buffering behavior. Delegate get_object_ranged in
src/storage/s3.rs#L910-L932, src/storage/gcs.rs#L702-L724, and
src/storage/azure_blob.rs#L689-L711 to the ranged helper; delegate the
get_objects implementations in src/storage/s3.rs#L949-L984,
src/storage/gcs.rs#L740-L775, and src/storage/azure_blob.rs#L727-L762 to the
buffered helper, passing each provider’s client and existing arguments.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 3a15ad3d-e08d-47c6-b255-362d941c9de4
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (14)
Cargo.tomlsrc/catalog/manifest.rssrc/metastore/metastore_traits.rssrc/metastore/metastores/object_store_metastore.rssrc/metastore/mod.rssrc/metrics/mod.rssrc/parseable/streams.rssrc/query/stream_schema_provider.rssrc/storage/azure_blob.rssrc/storage/gcs.rssrc/storage/localfs.rssrc/storage/mod.rssrc/storage/object_storage.rssrc/storage/s3.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- src/metrics/mod.rs
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/otel/metrics.rs (1)
675-677: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAdd a producer-side regression test for the new field.
The decoder test validates manually constructed JSON, but does not verify that
process_resource_metricsemitsSERIES_HASH_COLUMNas a JSON number and omits the legacy key. An end-to-end assertion here would protect the producer/decoder contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/otel/metrics.rs` around lines 675 - 677, Add a producer-side regression test covering process_resource_metrics that verifies emitted output contains SERIES_HASH_COLUMN as a JSON number and excludes the legacy key, complementing the existing decoder test and protecting the producer/decoder contract.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/event/format/mod.rs`:
- Around line 373-378: Update the SERIES_HASH_COLUMN match arm in the event
field-formatting logic to bind the JSON number and require
number.as_u64().is_some() before overriding the type to DataType::UInt64. Leave
negative, fractional, and otherwise non-unsigned values on the existing schema
path.
---
Nitpick comments:
In `@src/otel/metrics.rs`:
- Around line 675-677: Add a producer-side regression test covering
process_resource_metrics that verifies emitted output contains
SERIES_HASH_COLUMN as a JSON number and excludes the legacy key, complementing
the existing decoder test and protecting the producer/decoder contract.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: edaf900d-4339-4452-b41a-f33d1597f16a
📒 Files selected for processing (3)
src/event/format/json.rssrc/event/format/mod.rssrc/otel/metrics.rs
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/query/stream_schema_provider.rs (1)
487-490: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDowngrade column stats when any file is missing them
A missing
statsentry currently leaves the previous min/max in place, so the finalStatisticscan advertisePrecision::Exactfor bounds that only cover part of the snapshot. Clear that column’s aggregate or mark it inexact/unknown as soon as any file lacks stats, and add a regression test with one populated and one empty column-stats file.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 487 - 490, Update the column-statistics aggregation in the stream schema provider so a missing stats entry clears the affected aggregate or downgrades its bounds to unknown/inexact instead of retaining prior min/max values. Ensure the final Statistics does not report exact partial-snapshot bounds, and add a regression test covering one file with populated column stats and another with empty stats.
♻️ Duplicate comments (1)
src/query/stream_schema_provider.rs (1)
562-579: 🩺 Stability & Availability | 🟠 MajorMove the Rayon pruning phase off the Tokio worker.
Both parallel
collect()calls still execute synchronously insidecollect_from_snapshot, which is awaited directly byscan; the async worker remains occupied until CPU work and deallocation finish. Rayon’s parallel iterator operations are synchronous. (docs.rs)This is the same unresolved issue from the previous review. Move the complete pruning phase into
spawn_blockingor the established planning pool.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/query/stream_schema_provider.rs` around lines 562 - 579, Move the complete pruning phase in collect_from_snapshot, including both Rayon collect operations for prune_reason and manifest_files, into spawn_blocking or the established planning pool. Await that blocking task from the async flow so the Tokio worker is not occupied by synchronous filtering, collection, or deallocation; preserve the existing ordering and pruning behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 487-490: Update the column-statistics aggregation in the stream
schema provider so a missing stats entry clears the affected aggregate or
downgrades its bounds to unknown/inexact instead of retaining prior min/max
values. Ensure the final Statistics does not report exact partial-snapshot
bounds, and add a regression test covering one file with populated column stats
and another with empty stats.
---
Duplicate comments:
In `@src/query/stream_schema_provider.rs`:
- Around line 562-579: Move the complete pruning phase in collect_from_snapshot,
including both Rayon collect operations for prune_reason and manifest_files,
into spawn_blocking or the established planning pool. Await that blocking task
from the async flow so the Tokio worker is not occupied by synchronous
filtering, collection, or deallocation; preserve the existing ordering and
pruning behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 97ea5c3c-ed34-4146-8d42-c12b8cb6587e
📒 Files selected for processing (3)
src/query/stream_schema_provider.rssrc/storage/gcs.rssrc/storage/mod.rs
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/catalog/manifest.rs (1)
317-385: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for
decode_manifest_blocking.The tests cover synchronous codec behavior, but not the async
spawn_blockingwrapper that owns the semaphore permit. Add an async test covering compressed/plain inputs and decode-error propagation through this API.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/catalog/manifest.rs` around lines 317 - 385, Add an async test for decode_manifest_blocking that exercises both compressed output from encode_manifest and plain JSON input, asserting both decode successfully and preserve manifest data. Also pass invalid input through decode_manifest_blocking and assert the decode error is propagated, covering the wrapper’s spawn_blocking and semaphore-permit path.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/catalog/manifest.rs`:
- Around line 317-385: Add an async test for decode_manifest_blocking that
exercises both compressed output from encode_manifest and plain JSON input,
asserting both decode successfully and preserve manifest data. Also pass invalid
input through decode_manifest_blocking and assert the decode error is
propagated, covering the wrapper’s spawn_blocking and semaphore-permit path.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 5a75ade5-9db5-4e23-9bb1-e2c18ca120c1
📒 Files selected for processing (2)
src/catalog/manifest.rssrc/storage/mod.rs
3 participants
compresses manifest files using zstd to help with faster manifest fetch. Also, fetch manifest files concurrently.
Fixes #XXXX.
Description
This PR has:
Summary by CodeRabbit