feat: impl priority runtime for read

Cargo.toml

@@ -115,10 +115,10 @@ hash_ext = { path = "components/hash_ext" }
 hex = "0.4.3"
 hyperloglog = { git = "https://github.com/jedisct1/rust-hyperloglog.git", rev = "425487ce910f26636fbde8c4d640b538431aad50" }
 id_allocator = { path = "components/id_allocator" }
-influxql-logical-planner = { git = "https://github.com/CeresDB/influxql.git", rev = "acbd3ad7651f2deb74857155bea892f88926da57", package = "iox_query_influxql" }
-influxql-parser = { git = "https://github.com/CeresDB/influxql.git", rev = "acbd3ad7651f2deb74857155bea892f88926da57", package = "influxdb_influxql_parser" }
-influxql-query = { git = "https://github.com/CeresDB/influxql.git", rev = "acbd3ad7651f2deb74857155bea892f88926da57", package = "iox_query" }
-influxql-schema = { git = "https://github.com/CeresDB/influxql.git", rev = "acbd3ad7651f2deb74857155bea892f88926da57", package = "schema" }
+influxql-logical-planner = { git = "https://github.com/CeresDB/influxql.git", rev = "a905863", package = "iox_query_influxql" }
+influxql-parser = { git = "https://github.com/CeresDB/influxql.git", rev = "a905863", package = "influxdb_influxql_parser" }
+influxql-query = { git = "https://github.com/CeresDB/influxql.git", rev = "a905863", package = "iox_query" }
+influxql-schema = { git = "https://github.com/CeresDB/influxql.git", rev = "a905863", package = "schema" }
 interpreters = { path = "interpreters" }
 itertools = "0.10.5"
 lz4_flex = { version = "0.11", default-features = false, features = ["frame"] }

analytic_engine/src/instance/mod.rs

@@ -38,7 +38,7 @@ use generic_error::{BoxError, GenericError};
 use logger::{error, info};
 use macros::define_result;
 use mem_collector::MemUsageCollector;
-use runtime::Runtime;
+use runtime::{PriorityRuntime, Runtime};
 use snafu::{ResultExt, Snafu};
 use table_engine::{engine::EngineRuntimes, predicate::PredicateRef, table::FlushRequest};
 use time_ext::ReadableDuration;
@@ -291,7 +291,7 @@ impl Instance {
     }
 
     #[inline]
-    fn read_runtime(&self) -> &Arc<Runtime> {
+    fn read_runtime(&self) -> &PriorityRuntime {
         &self.runtimes.read_runtime
     }
 

analytic_engine/src/instance/read.rs

@@ -122,14 +122,18 @@ impl Instance {
             None,
         ));
 
+        let runtime = self
+            .read_runtime()
+            .choose_runtime(&request.priority)
+            .clone();
         let sst_read_options_builder = SstReadOptionsBuilder::new(
             ScanType::Query,
             self.scan_options.clone(),
             table_metrics.sst_metrics.clone(),
             table_options.num_rows_per_row_group,
             request.predicate.clone(),
             self.meta_cache.clone(),
-            self.read_runtime().clone(),
+            runtime,
         );
 
         if need_merge_sort {

analytic_engine/src/table/mod.rs

@@ -541,6 +541,8 @@ impl Table for TableImpl {
             projected_schema: request.projected_schema,
             predicate,
             metrics_collector: MetricsCollector::new(GET_METRICS_COLLECTOR_NAME.to_string()),
+            // TODO: pass priority from request
+            priority: Default::default(),
         };
         let mut batch_stream = self
             .read(read_request)

analytic_engine/src/tests/table.rs

@@ -192,6 +192,7 @@ pub fn new_read_all_request_with_order(schema: Schema, opts: ReadOptions) -> Rea
         projected_schema: ProjectedSchema::no_projection(schema),
         predicate: Arc::new(Predicate::empty()),
         metrics_collector: MetricsCollector::default(),
+        priority: Default::default(),
     }
 }
 

analytic_engine/src/tests/util.rs

@@ -26,6 +26,7 @@ use common_types::{
 use futures::stream::StreamExt;
 use logger::info;
 use object_store::config::{LocalOptions, ObjectStoreOptions, StorageOptions};
+use runtime::PriorityRuntime;
 use size_ext::ReadableSize;
 use table_engine::{
     engine::{
@@ -124,7 +125,7 @@ impl<T: WalsOpener> TestContext<T> {
                 .open_wals(
                     &self.config.wal,
                     WalRuntimes {
-                        read_runtime: self.runtimes.read_runtime.clone(),
+                        read_runtime: self.runtimes.read_runtime.high().clone(),
                         write_runtime: self.runtimes.write_runtime.clone(),
                         default_runtime: self.runtimes.default_runtime.clone(),
                     },
@@ -528,7 +529,7 @@ impl Builder {
             _dir: dir,
             config,
             runtimes: Arc::new(EngineRuntimes {
-                read_runtime: runtime.clone(),
+                read_runtime: PriorityRuntime::new(runtime.clone(), runtime.clone()),
                 write_runtime: runtime.clone(),
                 meta_runtime: runtime.clone(),
                 compact_runtime: runtime.clone(),

components/logger/Cargo.toml

@@ -35,6 +35,7 @@ workspace = true
 [dependencies]
 chrono = { workspace = true }
 log = "0.4"
+runtime = { workspace = true }
 serde = { workspace = true }
 slog = { workspace = true }
 slog-async = "2.6"

components/logger/src/lib.rs

@@ -28,6 +28,7 @@ pub use log::{
     debug as log_debug, error as log_error, info as log_info, max_level, trace as log_trace,
     warn as log_warn, SetLoggerError,
 };
+use runtime::Priority;
 use serde::{Deserialize, Serialize};
 pub use slog::Level;
 use slog::{slog_o, Drain, Key, OwnedKVList, Record, KV};
@@ -471,16 +472,18 @@ pub struct SlowTimer<'a> {
     sql: &'a str,
     slow_threshold: Duration,
     start_time: Instant,
+    priority: Option<Priority>,
 }
 
 impl<'a> Drop for SlowTimer<'a> {
     fn drop(&mut self) {
         let cost = self.elapsed();
         if cost > self.slow_threshold {
             slow_query!(
-                "Normal query elapsed:{:?}, id:{}, query:{}",
+                "Normal query elapsed:{:?}, id:{}, priority:{:?}, query:{}",
                 cost,
                 self.request_id,
+                self.priority,
                 self.sql,
             );
         }
@@ -494,6 +497,7 @@ impl<'a> SlowTimer<'a> {
             sql,
             slow_threshold: threshold,
             start_time: Instant::now(),
+            priority: None,
         }
     }
 
@@ -504,6 +508,10 @@ impl<'a> SlowTimer<'a> {
     pub fn start_time(&self) -> Instant {
         self.start_time
     }
+
+    pub fn priority(&mut self, priority: Priority) {
+        self.priority = Some(priority);
+    }
 }
 
 #[macro_export(local_inner_macros)]

components/runtime/src/lib.rs

@@ -30,8 +30,10 @@ use tokio::{
 };
 
 mod metrics;
+mod priority_runtime;
+
+pub use priority_runtime::{Priority, PriorityRuntime};
 
-// TODO(yingwen): Use opaque error type
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]
 pub enum Error {