Merge branch 'main' into query-param

Author: Devdutt Shenoi

CONTRIBUTING.md

@@ -0,0 +1,68 @@
+### Contributing
+
+This document outlines how you can contribute to Parseable.
+
+Thank you for considering to contribute to Parseable. The goal of this document is to provide everything you need to start your contribution. We encourage all contributions, including but not limited to:
+- Code patches, bug fixes.
+- Tutorials or blog posts.
+- Improving the documentation.
+- Submitting [bug reports](https://github.com/parseablehq/parseable/issues/new).
+
+### Prerequisites
+
+- Your PR has an associated issue. Find an [existing issue](https://github.com/parseablehq/parseable/issues) or open a new issue.
+- You've discussed the with [Parseable community](https://logg.ing/community).
+- You're familiar with GitHub Pull Requests(PR) workflow.
+- You've read the [Parseable documentation](https://www.parseable.com/docs).
+
+### Contribution workflow
+
+- Fork the [Parseable repository](https://help.github.com/en/github/getting-started-with-github/fork-a-repo) in your own GitHub account.
+- [Create a new branch](https://help.github.com/en/github/collaborating-with-issues-and-pull-requests/creating-and-deleting-branches-within-your-repository).
+- Review the Development Workflow section that describes the steps to maintain the repository.
+- Make your changes on your branch.
+- Submit the branch as a Pull Request pointing to the main branch of the Parseable repository. A maintainer should comment and/or review your Pull Request within a few hours.
+- You’ll be asked to review & sign the [Parseable Contributor License Agreement (CLA)](https://github.com/parseablehq/.github/blob/main/CLA.md) on the GitHub PR. Please ensure that you review the document. Once you’re ready, please sign the CLA by adding a comment I have read the CLA Document and I hereby sign the CLA in your Pull Request.
+- Please ensure that the commits in your Pull Request are made by the same GitHub user (which was used to create the PR and add the comment).
+
+### Development workflow
+
+We recommend Linux or macOS as the development platform for Parseable.
+
+#### Setup and run Parseable
+
+Parseable needs Rust 1.77.1 or above. Use the below command to build and run Parseable binary with local mode.
+
+```sh
+cargo run --release local-store
+```
+
+We recommend using the --release flag to test the full performance of Parseable.
+
+#### Running Tests
+
+```sh
+cargo test
+```
+
+This command will be triggered to each PR as a requirement for merging it.
+
+If you get a "Too many open files" error you might want to increase the open file limit using this command:
+
+```sh
+ulimit -Sn 3000
+```
+
+If you get a OpenSSL related error while building Parseable, you might need to install the dependencies using this command:
+
+```sh
+sudo apt install build-essential
+sudo apt-get install libssl-dev
+sudo apt-get install pkg-config
+```
+
+### Git Guidelines
+
+- The PR title should be accurate and descriptive of the changes.
+- The draft PRs are recommended when you want to show that you are working on something and make your work visible. Convert your PR as a draft if your changes are a work in progress. Maintainers will review the PR once you mark your PR as ready for review.
+- The branch related to the PR must be up-to-date with main before merging. We use Bors to automatically enforce this requirement without the PR author having to rebase manually.

Cargo.lock

@@ -0,0 +1,5 @@
+[target.aarch64-unknown-linux-gnu]
+image = "ghcr.io/cross-rs/aarch64-unknown-linux-gnu@sha256:1e2a0291f92a4372cbc22d8994e735473045383f1ce7fa44a16c234ba00187f4"
+
+[target.x86_64-unknown-linux-gnu]
+image = "ghcr.io/cross-rs/x86_64-unknown-linux-gnu@sha256:bf05360bb9d6d4947eed60532ac7a0d7e8fae8f214e9abb801d5941c8fe4918d"

Cross.toml

@@ -31,6 +31,7 @@ use tracing::trace;
 
 use crate::{
     alerts::AggregateCondition,
+    parseable::PARSEABLE,
     query::{TableScanVisitor, QUERY_SESSION},
     rbac::{
         map::SessionKey,
@@ -137,8 +138,9 @@ async fn execute_base_query(
         AlertError::CustomError(format!("Table name not found in query- {}", original_query))
     })?;
 
+    let time_partition = PARSEABLE.get_stream(&stream_name)?.get_time_partition();
     query
-        .get_dataframe(stream_name)
+        .get_dataframe(time_partition.as_ref())
         .await
         .map_err(|err| AlertError::CustomError(err.to_string()))
 }

src/alerts/alerts_utils.rs

@@ -37,7 +37,7 @@ use ulid::Ulid;
 pub mod alerts_utils;
 pub mod target;
 
-use crate::parseable::PARSEABLE;
+use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::map::SessionKey;
 use crate::storage;
@@ -514,17 +514,16 @@ impl AlertConfig {
 
         // for now proceed in a similar fashion as we do in query
         // TODO: in case of multiple table query does the selection of time partition make a difference? (especially when the tables don't have overlapping data)
-        let stream_name = if let Some(stream_name) = query.first_table_name() {
-            stream_name
-        } else {
+        let Some(stream_name) = query.first_table_name() else {
             return Err(AlertError::CustomError(format!(
                 "Table name not found in query- {}",
                 self.query
             )));
         };
 
+        let time_partition = PARSEABLE.get_stream(&stream_name)?.get_time_partition();
         let base_df = query
-            .get_dataframe(stream_name)
+            .get_dataframe(time_partition.as_ref())
             .await
             .map_err(|err| AlertError::CustomError(err.to_string()))?;
 
@@ -704,6 +703,8 @@ pub enum AlertError {
     CustomError(String),
     #[error("Invalid State Change: {0}")]
     InvalidStateChange(String),
+    #[error("{0}")]
+    StreamNotFound(#[from] StreamNotFound),
 }
 
 impl actix_web::ResponseError for AlertError {
@@ -717,6 +718,7 @@ impl actix_web::ResponseError for AlertError {
             Self::DatafusionError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::CustomError(_) => StatusCode::BAD_REQUEST,
             Self::InvalidStateChange(_) => StatusCode::BAD_REQUEST,
+            Self::StreamNotFound(_) => StatusCode::NOT_FOUND,
         }
     }
 

src/alerts/mod.rs

@@ -260,11 +260,12 @@ pub struct Options {
         help = "Set a fixed memory limit for query in GiB"
     )]
     pub query_memory_pool_size: Option<usize>,
-
+    // reduced the max row group size from 1048576
+    // smaller row groups help in faster query performance in multi threaded query
     #[arg(
         long,
         env = "P_PARQUET_ROW_GROUP_SIZE",
-        default_value = "1048576",
+        default_value = "262144",
         help = "Number of rows in a row group"
     )]
     pub row_group_size: usize,

src/cli.rs

@@ -38,7 +38,7 @@ use crate::handlers::http::query::{into_query, update_schema_when_distributed};
 use crate::handlers::livetail::cross_origin_config;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::parseable::PARSEABLE;
-use crate::query::{TableScanVisitor, QUERY_SESSION};
+use crate::query::{execute, TableScanVisitor, QUERY_SESSION};
 use crate::utils::arrow::flight::{
     append_temporary_events, get_query_from_ticket, into_flight_data, run_do_get_rpc,
     send_to_ingester,
@@ -215,8 +215,8 @@ impl FlightService for AirServiceImpl {
             Status::permission_denied("User Does not have permission to access this")
         })?;
         let time = Instant::now();
-        let (records, _) = query
-            .execute(stream_name.clone())
+
+        let (records, _) = execute(query, &stream_name)
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 

src/handlers/airplane.rs

@@ -39,9 +39,9 @@ use crate::handlers::http::fetch_schema;
 use crate::event::commit_schema;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::option::Mode;
-use crate::parseable::PARSEABLE;
+use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::query::error::ExecuteError;
-use crate::query::{CountsRequest, CountsResponse, Query as LogicalQuery};
+use crate::query::{execute, CountsRequest, CountsResponse, Query as LogicalQuery};
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::Users;
 use crate::response::QueryResponse;
@@ -130,7 +130,8 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
 
         return Ok(HttpResponse::Ok().json(response));
     }
-    let (records, fields) = query.execute(table_name.clone()).await?;
+
+    let (records, fields) = execute(query, &table_name).await?;
 
     let response = QueryResponse {
         records,
@@ -318,6 +319,8 @@ Description: {0}"#
     ActixError(#[from] actix_web::Error),
     #[error("Error: {0}")]
     Anyhow(#[from] anyhow::Error),
+    #[error("Error: {0}")]
+    StreamNotFound(#[from] StreamNotFound),
 }
 
 impl actix_web::ResponseError for QueryError {

src/handlers/http/query.rs

@@ -28,7 +28,7 @@ use http::{header::CONTENT_TYPE, HeaderName, HeaderValue, StatusCode};
 use once_cell::sync::Lazy;
 pub use staging::StagingError;
 use streams::StreamRef;
-pub use streams::{StreamNotFound, Streams};
+pub use streams::{Stream, StreamNotFound, Streams};
 use tracing::error;
 
 #[cfg(feature = "kafka")]
@@ -451,23 +451,17 @@ impl Parseable {
                 .await;
         }
 
-        let time_partition_in_days = if !time_partition_limit.is_empty() {
-            Some(validate_time_partition_limit(&time_partition_limit)?)
-        } else {
-            None
-        };
-
-        if let Some(custom_partition) = &custom_partition {
-            validate_custom_partition(custom_partition)?;
-        }
-
-        if !time_partition.is_empty() && custom_partition.is_some() {
+        if !time_partition.is_empty() || !time_partition_limit.is_empty() {
             return Err(StreamError::Custom {
-                msg: "Cannot set both time partition and custom partition".to_string(),
+                msg: "Creating stream with time partition is not supported anymore".to_string(),
                 status: StatusCode::BAD_REQUEST,
             });
         }
 
+        if let Some(custom_partition) = &custom_partition {
+            validate_custom_partition(custom_partition)?;
+        }
+
         let schema = validate_static_schema(
             body,
             stream_name,
@@ -479,7 +473,7 @@ impl Parseable {
         self.create_stream(
             stream_name.to_string(),
             &time_partition,
-            time_partition_in_days,
+            None,
             custom_partition.as_ref(),
             static_schema_flag,
             schema,