chore: improve query performance

[nikhilsinhaparseable]

• create missing streams concurrently
• create stream metadata and schema for a stream concurrently

this will improve performance for each query call

Summary by CodeRabbit

• Bug Fixes

  • Improved error handling and logging during stream creation and querying, ensuring failures are properly reported and do not go unnoticed.

• Refactor

  • Optimized stream and schema retrieval by performing these operations concurrently, reducing wait times for users during data access.

[coderabbitai]

Walkthrough

The changes refactor stream and schema creation logic to improve concurrency and error handling. The create_streams_for_querier function is now asynchronous, propagates errors, and concurrently spawns tasks to create missing streams. The create_stream_and_schema_from_storage method now fetches stream metadata and schema in parallel using tokio::try_join!, optimizing I/O operations.

Changes

File(s)	Change Summary
src/handlers/http/query.rs	Refactored create_streams_for_querier to async, added error propagation, concurrency, and logging. Updated imports. Updated query to propagate errors from this function.
src/parseable/mod.rs	Refactored create_stream_and_schema_from_storage to fetch stream metadata and schema concurrently using tokio::try_join!. No public API changes.

Sequence Diagram(s)

sequenceDiagram
    participant Querier
    participant Storage
    participant TokioTasks

    Querier->>Querier: create_streams_for_querier()
    Querier->>Storage: fetch storage streams (async)
    Querier->>Querier: compare with existing streams
    alt Missing streams
        loop For each missing stream
            Querier->>TokioTasks: spawn async task to create stream & schema
            TokioTasks->>Storage: fetch stream metadata & schema (parallel)
            Storage-->>TokioTasks: return metadata & schema
            TokioTasks->>Querier: log success or warning
        end
        Querier->>TokioTasks: await all tasks
    end
    Querier->>Querier: return Ok or propagate error

Loading

Possibly related PRs

• parseablehq/parseable#1258: Refactors stream creation logic to use a safer get_or_create approach, which complements the concurrency improvements in this PR.

Suggested reviewers

• de-sh

Poem

In the warren where queries hop and play,
Streams now gather in a faster way.
With tasks that run and warnings logged,
The streams and schemas are never bogged.
Parallel fetches, errors caught in stride—
The code hops forward, with rabbits as guide!
🐇✨

---

📜 Recent review details

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 33ed89c and f9396df.

📒 Files selected for processing (2)

• src/handlers/http/query.rs (3 hunks)
• src/parseable/mod.rs (2 hunks)

🧰 Additional context used

🧠 Learnings (3)

📓 Common learnings

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/query/mod.rs:64-66
Timestamp: 2025-06-18T06:39:04.775Z
Learning: In src/query/mod.rs, QUERY_SESSION_STATE and QUERY_SESSION serve different architectural purposes: QUERY_SESSION_STATE is used for stats calculation and allows dynamic registration of individual parquet files from the staging path (files created every minute), while QUERY_SESSION is used for object store queries with the global schema provider. Session contexts with schema providers don't support registering individual tables/parquets, so both session objects are necessary for their respective use cases.

src/parseable/mod.rs (4)

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:319-331
Timestamp: 2025-06-16T09:50:38.636Z
Learning: In Parseable's Ingest or Query mode, the node_id is always available because it's generated during server initialization itself, before the get_node_id_string() function in streams.rs would be called. This makes the .expect() calls on QUERIER_META.get() and INGESTOR_META.get() safe in this context.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1263
File: src/handlers/http/ingest.rs:300-310
Timestamp: 2025-03-26T06:44:53.362Z
Learning: In Parseable, every stream is always associated with a log_source - no stream can exist without a log_source. For otel-traces and otel-metrics, strict restrictions are implemented where ingestion is rejected if a stream already has a different log_source format. However, regular logs from multiple log_sources can coexist in a single stream.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/sync.rs:78-83
Timestamp: 2025-06-16T04:56:21.613Z
Learning: In Rust async code, `#[tokio::main]` is appropriate when functions are called from separate OS threads (like `thread::spawn()`), as it creates a new Tokio runtime for that thread. The "Cannot start a runtime from within a runtime" error only occurs when `#[tokio::main]` functions are called from within an existing Tokio runtime context.

src/handlers/http/query.rs (3)

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/query/mod.rs:64-66
Timestamp: 2025-06-18T06:39:04.775Z
Learning: In src/query/mod.rs, QUERY_SESSION_STATE and QUERY_SESSION serve different architectural purposes: QUERY_SESSION_STATE is used for stats calculation and allows dynamic registration of individual parquet files from the staging path (files created every minute), while QUERY_SESSION is used for object store queries with the global schema provider. Session contexts with schema providers don't support registering individual tables/parquets, so both session objects are necessary for their respective use cases.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/handlers/http/health_check.rs:81-90
Timestamp: 2025-06-16T02:04:58.990Z
Learning: In the shutdown function in src/handlers/http/health_check.rs, the design approach is to log errors from sync operations rather than propagate them. This is intentional because the shutdown function is called on SIGTERM/SIGINT signals, and the goal is to perform best-effort cleanup (syncing pending files to object storage) while allowing the shutdown to proceed regardless of sync failures. Logging provides debugging information without blocking the shutdown process.

⏰ Context from checks skipped due to timeout of 90000ms (10)

• GitHub Check: Build Default x86_64-pc-windows-msvc
• GitHub Check: Build Default aarch64-apple-darwin
• GitHub Check: Build Default x86_64-apple-darwin
• GitHub Check: Build Default aarch64-unknown-linux-gnu
• GitHub Check: Build Default x86_64-unknown-linux-gnu
• GitHub Check: Build Kafka aarch64-apple-darwin
• GitHub Check: Build Kafka x86_64-unknown-linux-gnu
• GitHub Check: coverage
• GitHub Check: Quest Smoke and Load Tests for Distributed deployments
• GitHub Check: Quest Smoke and Load Tests for Standalone deployments

🔇 Additional comments (3)

src/parseable/mod.rs (1)

274-289: Excellent optimization using concurrent I/O operations!

The refactoring to use tokio::try_join! effectively parallelizes the fetching of stream metadata and schema bytes, reducing the overall latency from sequential execution time to the maximum of both operations. The error handling and conditional deserialization logic are properly preserved.

src/handlers/http/query.rs (2)

437-476: Well-designed refactoring with improved concurrency and error handling!

The changes significantly improve the robustness and performance of stream creation:

• HashSet provides O(1) lookup performance for checking existing streams
• JoinSet enables concurrent stream creation, reducing total execution time
• Proper error handling with warnings allows partial success instead of complete failure
• Early return optimization avoids unnecessary work when all streams exist

The error logging approach is appropriate here as it allows the system to continue operating even if some streams fail to create.

---

130-130: Good improvement in error propagation!

Propagating errors from create_streams_for_querier using the ? operator ensures that stream creation failures are properly handled and reported to the caller, improving system observability.

✨ Finishing Touches

• 📝 Generate Docstrings

---

🪧 Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  • I pushed a fix in commit <commit_id>, please review it.
  • Explain this complex logic.
  • Open a follow-up GitHub issue for this discussion.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  • @coderabbitai explain this code block.
  • @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
  • @coderabbitai read src/utils.ts and explain its main purpose.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
  • @coderabbitai help me debug CodeRabbit configuration file.

Support

Need help? Create a ticket on our support page for assistance with any issues or questions.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (Invoked using PR comments)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai full review to do a full review from scratch and review all the files again.
• @coderabbitai summary to regenerate the summary of the PR.
• @coderabbitai generate docstrings to generate docstrings for this PR.
• @coderabbitai generate sequence diagram to generate a sequence diagram of the changes in this PR.
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai configuration to show the current CodeRabbit configuration for the repository.
• @coderabbitai help to get help.

Other keywords and placeholders

• Add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.
• Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
• Add @coderabbitai anywhere in the PR title to generate the title automatically.

CodeRabbit Configuration File (.coderabbit.yaml)

• You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
• Please see the configuration documentation for more information.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.