Skip to content

Commit 0e4f849

Browse files
add prism check for various methods
1 parent e139b0f commit 0e4f849

File tree

6 files changed

+9
-10
lines changed

6 files changed

+9
-10
lines changed

src/handlers/http/about.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,8 @@ pub async fn about() -> Json<Value> {
6363
let commit = current_release.commit_hash;
6464
let deployment_id = meta.deployment_id.to_string();
6565
let mode = PARSEABLE.get_server_mode_string();
66-
let staging = if PARSEABLE.options.mode == Mode::Query {
66+
let staging = if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism
67+
{
6768
"".to_string()
6869
} else {
6970
PARSEABLE.options.staging_dir().display().to_string()

src/handlers/http/query.rs

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,6 @@ use crate::query::{execute, CountsRequest, CountsResponse, Query as LogicalQuery
4545
use crate::query::{TableScanVisitor, QUERY_SESSION};
4646
use crate::rbac::Users;
4747
use crate::response::{QueryResponse, TIME_ELAPSED_HEADER};
48-
use crate::storage::object_storage::commit_schema_to_storage;
4948
use crate::storage::ObjectStorageError;
5049
use crate::utils::actix::extract_session_key_from_req;
5150
use crate::utils::time::{TimeParseError, TimeRange};
@@ -173,12 +172,9 @@ pub async fn get_counts(
173172
}
174173

175174
pub async fn update_schema_when_distributed(tables: &Vec<String>) -> Result<(), EventError> {
176-
if PARSEABLE.options.mode == Mode::Query {
175+
if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
177176
for table in tables {
178177
if let Ok(new_schema) = fetch_schema(table).await {
179-
// commit schema merges the schema internally and updates the schema in storage.
180-
commit_schema_to_storage(table, new_schema.clone()).await?;
181-
182178
commit_schema(table, Arc::new(new_schema))?;
183179
}
184180
}

src/parseable/mod.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -214,6 +214,7 @@ impl Parseable {
214214
pub async fn check_or_load_stream(&self, stream_name: &str) -> bool {
215215
!self.streams.contains(stream_name)
216216
&& (self.options.mode != Mode::Query
217+
|| self.options.mode != Mode::Prism
217218
|| !self
218219
.create_stream_and_schema_from_storage(stream_name)
219220
.await
@@ -521,7 +522,7 @@ impl Parseable {
521522
let stream_in_memory_dont_update =
522523
self.streams.contains(stream_name) && !update_stream_flag;
523524
let stream_in_storage_only_for_query_node = !self.streams.contains(stream_name) // check if stream in storage only if not in memory
524-
&& self.options.mode == Mode::Query // and running in query mode
525+
&& (self.options.mode == Mode::Query || self.options.mode == Mode::Prism) // and running in query mode
525526
&& self
526527
.create_stream_and_schema_from_storage(stream_name)
527528
.await?;

src/query/mod.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -467,7 +467,7 @@ pub async fn get_manifest_list(
467467
let mut merged_snapshot: Snapshot = Snapshot::default();
468468

469469
// get a list of manifests
470-
if PARSEABLE.options.mode == Mode::Query {
470+
if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
471471
let path = RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY]);
472472
let obs = glob_storage
473473
.get_objects(

src/query/stream_schema_provider.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -503,7 +503,7 @@ impl TableProvider for StandardTableProvider {
503503
.await?;
504504
};
505505
let mut merged_snapshot = Snapshot::default();
506-
if PARSEABLE.options.mode == Mode::Query {
506+
if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
507507
let path = RelativePathBuf::from_iter([&self.stream, STREAM_ROOT_DIRECTORY]);
508508
let obs = glob_storage
509509
.get_objects(

src/utils/arrow/flight.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,8 @@ pub fn send_to_ingester(start: i64, end: i64) -> bool {
133133
);
134134
let time_filters =
135135
extract_primary_filter(&[Expr::BinaryExpr(ex1), Expr::BinaryExpr(ex2)], &None);
136-
PARSEABLE.options.mode == Mode::Query && is_within_staging_window(&time_filters)
136+
(PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism)
137+
&& is_within_staging_window(&time_filters)
137138
}
138139

139140
fn lit_timestamp_milli(time: i64) -> Expr {

0 commit comments

Comments
 (0)