perf: don't construct a tokio runtime for each query #1226
Conversation
WalkthroughThis pull request refactors the query execution flow across multiple modules. The changes remove the use of blocking tasks in favor of direct asynchronous calls. In the airplane and HTTP handlers, the query execution now obtains a stream directly from a central parser module and executes queries asynchronously with improved error handling. Additionally, the public interface is expanded with a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant AirService
participant Parseable
participant Query
Client->>AirService: GET query request
AirService->>Parseable: get_stream(stream_name)
Parseable-->>AirService: stream / error
AirService->>Query: query.execute(&stream)
Query->>stream: get_time_partition()
Query-->>AirService: results / error
AirService-->>Client: response
Suggested labels
Suggested reviewers
Poem
✨ Finishing Touches
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/alerts/alerts_utils.rs (1)
34-34: Avoid potential confusion with global usage
By referencingPARSEABLEdirectly, you rely on a global context, which may make testing or concurrency more complex. Consider documenting concurrency safety or injecting dependencies if feasible.src/query/mod.rs (1)
65-77: Dedicated async runtime for queries
Providing a specializedQUERY_RUNTIMEcan improve performance and isolation for query processing. Ensure it does not conflict with any existing runtime usage within your application.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/alerts/alerts_utils.rs(2 hunks)src/alerts/mod.rs(4 hunks)src/handlers/airplane.rs(2 hunks)src/handlers/http/query.rs(3 hunks)src/parseable/mod.rs(1 hunks)src/query/mod.rs(7 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- src/handlers/airplane.rs
- src/parseable/mod.rs
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
🔇 Additional comments (13)
src/alerts/alerts_utils.rs (1)
141-143: Time partition usage is consistent
Retrievingtime_partitionfrom the stream ensures precise data filtering. This approach is consistent with the rest of the code for retrieving time-based partitions.src/handlers/http/query.rs (3)
42-44: New imports for streamlined error handling and query execution
ImportingStreamNotFound,PARSEABLE, andexecuteis a clear approach, aligning with the new asynchronous query strategy.
133-134: Switch to the newexecutefunction
Replacing the old blocking query approach withexecute(query, &table_name).await?improves performance by leveraging the dedicated async runtime.
322-323: Extended error types
The addition ofStreamNotFoundtoQueryErrorimproves clarity when indicating that a stream was not located.src/alerts/mod.rs (3)
40-40: Introduced global reference for parseable
ImportingPARSEABLEandStreamNotFoundensures consistent usage of the global context and specialized error in the alerts system.
706-707: NewStreamNotFoundvariant
Defining a dedicated error variant clarifies potential missing stream issues within the alerts module.
721-721: Specific HTTP status for missing streams
MappingStreamNotFoundtoNOT_FOUNDstatus code provides precise error responses for clients.src/query/mod.rs (6)
45-45: AddRuntimeimport
Importingtokio::runtime::Runtimeis crucial for creating the dedicated query execution environment.
150-150: Revised signature forexecute
Allowing an optionaltime_partitionreference clarifies partition usage, enabling fallback behavior if no partition is provided.
173-176:get_dataframewith optional time partition
This adaptation ensures consistent interface design across the module, especially for partition-based queries.
185-185:final_logical_plannow supports optional partition
Extending the function signature to acceptOption<&String>aligns with the rest of the code handling partitions gracefully.
505-506:transformfunction parameter
By passingOption<&String>for the partition, the code can branch between a custom partition key and the default timestamp key.
563-563:table_contains_any_time_filtersincludes optional partition
Recognizing both user-specified partition and default timestamp is essential for correct filter application.
Signed-off-by: Devdutt Shenoi <devdutt@outlook.in>
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
src/query/mod.rs (1)
69-80: Consider handling the JoinError instead of using expect.The implementation correctly spawns query execution on the dedicated runtime. However, the
.expect("The Join should have been successful")on line 79 could lead to a panic if the spawned task fails. Consider handling this error case more gracefully.pub async fn execute( query: Query, stream_name: &str, ) -> Result<(Vec<RecordBatch>, Vec<String>), ExecuteError> { let time_partition = PARSEABLE.get_stream(stream_name)?.get_time_partition(); QUERY_RUNTIME .spawn(async move { query.execute(time_partition.as_ref()).await }) .await - .expect("The Join should have been successful") + .map_err(|e| ExecuteError::Datafusion(DataFusionError::Execution(format!("Join error: {}", e))))? }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/query/mod.rs(7 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: coverage
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
🔇 Additional comments (6)
src/query/mod.rs (6)
64-66: Good initialization of a dedicated runtime for queries.The static runtime provides a single, shared Tokio runtime for all queries, which aligns with the PR's objective of avoiding runtime creation for each query. The use of
Lazy::new()ensures the runtime is only initialized once when needed.
151-154: Signature change aligns with the centralized runtime approach.The method now accepts a reference to a time partition instead of a stream name string, which fits well with the new architecture where time partition resolution happens in the outer
executefunction. This change helps avoid redundant work.
176-179: Consistent parameter changes to support the new approach.The
get_dataframemethod has been updated to match the pattern used inexecute, accepting a time partition reference instead of a stream name. This ensures consistency across the API and supports the centralized time partition resolution.
188-188: Parameter type simplification is appropriate.Changed from taking a reference to an Option to directly taking an Option of a reference, which is a more ergonomic approach and aligns with the parameter changes in other methods.
504-509: Parameter type consistency maintained throughout codebase.The
transformfunction's signature has been updated to match the changes in related functions, ensuring consistency across the module. This type of parameter is more efficient as it avoids unnecessary cloning.
564-567: Simplified time_partition check withis_some_and.The code now uses
is_some_andfor a more concise check against the time partition field. This improves readability while maintaining functionality.Also applies to: 580-581
Fixes #XXXX.
Description
This PR has:
Summary by CodeRabbit
New Features
StreamNotFound, for enhanced error reporting related to stream retrieval issues.Refactor