diff --git a/Cargo.lock b/Cargo.lock index 5d7ac01047..17d5882e5e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10785,6 +10785,7 @@ dependencies = [ "enum_dispatch", "futures", "greptime-proto", + "humantime-serde", "itertools 0.14.0", "jsonb", "jsonpath-rust 0.7.5", @@ -14301,6 +14302,7 @@ dependencies = [ "hostname 0.4.1", "log-store", "mito2", + "pipeline", "query", "serde", "servers", diff --git a/config/config.md b/config/config.md index 11e846ac76..319110809d 100644 --- a/config/config.md +++ b/config/config.md @@ -244,6 +244,8 @@ | `slow_query.sample_ratio` | Float | Unset | The sampling ratio of slow query log. The value should be in the range of (0, 1]. | | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. | | `tracing.tokio_console_addr` | String | Unset | The tokio console address. | +| `pipeline` | -- | -- | The pipeline options. | +| `pipeline.cache_ttl` | String | `10s` | Time to live of the local pipeline cache. Default is `10s`. | | `event_recorder` | -- | -- | Configuration options for the event recorder. | | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. | | `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `create_database`,
`alter_database`, `drop_database`, `create_flow`, `drop_flow`,
`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,
`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,
`create_view`, `drop_view`, `admin_function`, `reconcile_table`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | @@ -388,6 +390,8 @@ | `tracing.tokio_console_addr` | String | Unset | The tokio console address. | | `memory` | -- | -- | The memory options. | | `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.
When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
is set to "prof:true,prof_active:false". The official image adds this env variable.
Default is true. | +| `pipeline` | -- | -- | The pipeline options. | +| `pipeline.cache_ttl` | String | `10s` | Time to live of the frontend-local pipeline cache. A pipeline created or deleted on
another frontend takes effect on this one after at most this duration. | | `event_recorder` | -- | -- | Configuration options for the event recorder. | | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. | | `event_recorder.event_types` | Array | -- | Event types to record. Current available event type: `admin_function`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 9175fa598c..e85773c52e 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -428,6 +428,12 @@ ttl = "90d" ## Default is true. enable_heap_profiling = true +## The pipeline options. +[pipeline] +## Time to live of the frontend-local pipeline cache. A pipeline created or deleted on +## another frontend takes effect on this one after at most this duration. +cache_ttl = "10s" + ## Configuration options for the event recorder. [event_recorder] ## TTL for the events table that will be used to store the events. Default is `90d`. diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 5458e0ed83..0444863727 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -956,6 +956,11 @@ default_ratio = 1.0 ## @toml2docs:none-default #+ tokio_console_addr = "127.0.0.1" +## The pipeline options. +[pipeline] +## Time to live of the local pipeline cache. Default is `10s`. +cache_ttl = "10s" + ## Configuration options for the event recorder. [event_recorder] ## TTL for the events table that will be used to store the events. Default is `90d`. diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index 18cedc0545..bb15abd7bb 100644 --- a/src/frontend/src/frontend.rs +++ b/src/frontend/src/frontend.rs @@ -22,6 +22,7 @@ use common_options::datanode::DatanodeClientOptions; use common_options::memory::MemoryOptions; use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions}; use meta_client::MetaClientOptions; +use pipeline::PipelineOptions; use query::options::QueryOptions; use serde::{Deserialize, Serialize}; use servers::grpc::GrpcOptions; @@ -75,6 +76,8 @@ pub struct FrontendOptions { pub query: QueryOptions, pub slow_query: SlowQueryOptions, pub memory: MemoryOptions, + /// The pipeline options. + pub pipeline: PipelineOptions, /// The event recorder options. pub event_recorder: EventRecorderOptions, /// Environment variable keys to read and report in heartbeat messages. @@ -108,6 +111,7 @@ impl Default for FrontendOptions { query: QueryOptions::default(), slow_query: SlowQueryOptions::default(), memory: MemoryOptions::default(), + pipeline: PipelineOptions::default(), event_recorder: EventRecorderOptions::default(), heartbeat_env_vars: vec![], } diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index 0f87dc424f..e49b75bbf8 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -334,6 +334,7 @@ impl FrontendBuilder { statement_executor.clone(), self.catalog_manager.clone(), query_engine.clone(), + &self.options.pipeline, )); plugins.insert::(statement_executor.clone()); diff --git a/src/pipeline/Cargo.toml b/src/pipeline/Cargo.toml index d1a1f88ca8..620ac36ac7 100644 --- a/src/pipeline/Cargo.toml +++ b/src/pipeline/Cargo.toml @@ -41,11 +41,12 @@ dyn-fmt = "0.4" enum_dispatch = "0.3" futures.workspace = true greptime-proto.workspace = true +humantime-serde.workspace = true itertools.workspace = true jsonb.workspace = true jsonpath-rust = "0.7.5" lazy_static.workspace = true -moka = { workspace = true, features = ["sync"] } +moka = { workspace = true, features = ["future"] } once_cell.workspace = true operator.workspace = true ordered-float.workspace = true @@ -53,6 +54,7 @@ paste.workspace = true prometheus.workspace = true query.workspace = true regex.workspace = true +serde = { version = "1.0", features = ["derive"] } serde_json.workspace = true session.workspace = true snafu.workspace = true @@ -67,7 +69,6 @@ yaml-rust = "0.4" catalog = { workspace = true, features = ["testing"] } criterion = { workspace = true, features = ["html_reports"] } rayon = "1.0" -serde = { version = "1.0", features = ["derive"] } session = { workspace = true, features = ["testing"] } [[bench]] diff --git a/src/pipeline/src/error.rs b/src/pipeline/src/error.rs index 36abcaae9e..0da65d2543 100644 --- a/src/pipeline/src/error.rs +++ b/src/pipeline/src/error.rs @@ -699,6 +699,15 @@ pub enum Error { location: Location, }, + /// `try_get_with` shares one loader across concurrent misses, so its error + /// arrives behind an `Arc`. + #[snafu(display("Failed to load pipeline into cache: {}", error))] + CacheLoad { + error: std::sync::Arc, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to collect record batch"))] CollectRecords { #[snafu(implicit)] @@ -895,6 +904,7 @@ impl ErrorExt for Error { fn status_code(&self) -> StatusCode { use Error::*; match self { + CacheLoad { error, .. } => error.status_code(), CastType { .. } => StatusCode::Unexpected, PipelineTableNotFound { .. } => StatusCode::TableNotFound, InsertPipeline { source, .. } => source.status_code(), diff --git a/src/pipeline/src/lib.rs b/src/pipeline/src/lib.rs index c3a45d4fe4..e323859d02 100644 --- a/src/pipeline/src/lib.rs +++ b/src/pipeline/src/lib.rs @@ -19,6 +19,7 @@ pub mod error; mod etl; mod manager; mod metrics; +pub mod options; mod tablesuffix; pub use etl::ctx_req::{ContextOpt, ContextReq}; @@ -35,6 +36,7 @@ pub use manager::{ IdentityTimeIndex, PipelineContext, PipelineDefinition, PipelineInfo, PipelineRef, PipelineTableRef, PipelineVersion, PipelineWay, SelectInfo, pipeline_operator, table, util, }; +pub use options::PipelineOptions; #[macro_export] macro_rules! unwrap_or_continue_if_err { diff --git a/src/pipeline/src/manager/pipeline_cache.rs b/src/pipeline/src/manager/pipeline_cache.rs index 2abe24e94b..e60963e7f8 100644 --- a/src/pipeline/src/manager/pipeline_cache.rs +++ b/src/pipeline/src/manager/pipeline_cache.rs @@ -12,14 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::future::Future; use std::sync::Arc; use std::time::Duration; -use common_telemetry::debug; use datatypes::timestamp::TimestampNanosecond; -use moka::sync::Cache; +use moka::future::Cache; -use crate::error::{MultiPipelineWithDiffSchemaSnafu, Result}; +use crate::error::{CacheLoadSnafu, MultiPipelineWithDiffSchemaSnafu, Result}; use crate::etl::Pipeline; use crate::manager::PipelineVersion; use crate::table::EMPTY_SCHEMA_NAME; @@ -27,11 +27,14 @@ use crate::util::{generate_pipeline_cache_key, generate_pipeline_cache_key_suffi /// Pipeline table cache size. const PIPELINES_CACHE_SIZE: u64 = 10000; -/// Pipeline table cache time to live. -const PIPELINES_CACHE_TTL: Duration = Duration::from_secs(10); /// Pipeline cache is located on a separate file on purpose, /// to encapsulate inner cache. Only public methods are exposed. +/// +/// `pipelines` and `original_pipelines` are keyed by the *requested* schema so +/// a lookup is a single key probe, as [`Cache::try_get_with`] requires; +/// resolving it to a stored schema is the loader's job. `failover_cache` has no +/// loader and keeps the stored-schema key. pub(crate) struct PipelineCache { pipelines: Cache>, original_pipelines: Cache, @@ -49,16 +52,16 @@ pub struct PipelineContent { } impl PipelineCache { - pub(crate) fn new() -> Self { + pub(crate) fn new(ttl: Duration) -> Self { Self { pipelines: Cache::builder() .max_capacity(PIPELINES_CACHE_SIZE) - .time_to_live(PIPELINES_CACHE_TTL) + .time_to_live(ttl) .name("pipelines") .build(), original_pipelines: Cache::builder() .max_capacity(PIPELINES_CACHE_SIZE) - .time_to_live(PIPELINES_CACHE_TTL) + .time_to_live(ttl) .name("original_pipelines") .build(), failover_cache: Cache::builder() @@ -68,163 +71,242 @@ impl PipelineCache { } } - pub(crate) fn insert_pipeline_cache( + /// Concurrent misses on the same key share one `init` call. + pub(crate) async fn get_pipeline_with( &self, schema: &str, name: &str, version: PipelineVersion, - pipeline: Arc, - with_latest: bool, - ) { - insert_cache_generic( - &self.pipelines, - schema, - name, - version, - pipeline.clone(), - with_latest, - ); + init: impl Future>>, + ) -> Result> { + let key = generate_pipeline_cache_key(schema, name, version); + self.pipelines + .try_get_with(key, init) + .await + .map_err(|error| CacheLoadSnafu { error }.build()) } - pub(crate) fn insert_pipeline_str_cache(&self, pipeline: &PipelineContent, with_latest: bool) { - let schema = pipeline.schema.as_str(); - let name = pipeline.name.as_str(); - let version = pipeline.version; - insert_cache_generic( - &self.original_pipelines, - schema, - name, - Some(version), - pipeline.clone(), - with_latest, - ); - insert_cache_generic( - &self.failover_cache, - schema, - name, - Some(version), - pipeline.clone(), - with_latest, - ); - } - - pub(crate) fn get_pipeline_cache( + /// Concurrent misses on the same key share one `init` call. + pub(crate) async fn get_pipeline_str_with( &self, schema: &str, name: &str, version: PipelineVersion, - ) -> Result>> { - get_cache_generic(&self.pipelines, schema, name, version) + init: impl Future>, + ) -> Result { + let key = generate_pipeline_cache_key(schema, name, version); + self.original_pipelines + .try_get_with(key, init) + .await + .map_err(|error| CacheLoadSnafu { error }.build()) } - pub(crate) fn get_failover_cache( + /// Resolves across schemas, unlike the loaded caches: a pipeline stored + /// under the empty schema is reachable from any schema. + pub(crate) async fn get_failover_cache( &self, schema: &str, name: &str, version: PipelineVersion, ) -> Result> { - get_cache_generic(&self.failover_cache, schema, name, version) - } + for key in [ + generate_pipeline_cache_key(EMPTY_SCHEMA_NAME, name, version), + generate_pipeline_cache_key(schema, name, version), + ] { + if let Some(content) = self.failover_cache.get(&key).await { + return Ok(Some(content)); + } + } - pub(crate) fn get_pipeline_str_cache( - &self, - schema: &str, - name: &str, - version: PipelineVersion, - ) -> Result> { - get_cache_generic(&self.original_pipelines, schema, name, version) - } - - // remove cache with version and latest in all schemas - pub(crate) fn remove_cache(&self, name: &str, version: PipelineVersion) { - let version_suffix = generate_pipeline_cache_key_suffix(name, version); - let latest_suffix = generate_pipeline_cache_key_suffix(name, None); - - let ks = self - .pipelines + // Stored under some other schema; unambiguous only if exactly one has it. + let suffix = generate_pipeline_cache_key_suffix(name, version); + let mut found = self + .failover_cache .iter() - .filter_map(|(k, _)| { - if k.ends_with(&version_suffix) || k.ends_with(&latest_suffix) { - Some(k.clone()) - } else { - None - } - }) + .filter(|(k, _)| k.ends_with(&suffix)) .collect::>(); - for k in ks { - let k = k.as_str(); - self.pipelines.remove(k); - self.original_pipelines.remove(k); - self.failover_cache.remove(k); - } - } -} - -fn insert_cache_generic( - cache: &Cache, - schema: &str, - name: &str, - version: PipelineVersion, - value: T, - with_latest: bool, -) { - let k = generate_pipeline_cache_key(schema, name, version); - cache.insert(k, value.clone()); - if with_latest { - let k = generate_pipeline_cache_key(schema, name, None); - cache.insert(k, value); - } -} - -fn get_cache_generic( - cache: &Cache, - schema: &str, - name: &str, - version: PipelineVersion, -) -> Result> { - // lets try empty schema first - let emp_key = generate_pipeline_cache_key(EMPTY_SCHEMA_NAME, name, version); - if let Some(value) = cache.get(&emp_key) { - return Ok(Some(value)); - } - // use input schema - let schema_k = generate_pipeline_cache_key(schema, name, version); - if let Some(value) = cache.get(&schema_k) { - return Ok(Some(value)); - } - - // try all schemas - let suffix_key = generate_pipeline_cache_key_suffix(name, version); - let mut ks = cache - .iter() - .filter(|e| e.0.ends_with(&suffix_key)) - .collect::>(); - - match ks.len() { - 0 => Ok(None), - 1 => { - let (_, value) = ks.remove(0); - Ok(Some(value)) - } - _ => { - debug!( - "caches keys: {:?}, emp key: {:?}, schema key: {:?}, suffix key: {:?}", - cache.iter().map(|e| e.0).collect::>(), - emp_key, - schema_k, - suffix_key - ); - MultiPipelineWithDiffSchemaSnafu { + match found.len() { + 0 => Ok(None), + 1 => Ok(Some(found.remove(0).1)), + _ => MultiPipelineWithDiffSchemaSnafu { name: name.to_string(), current_schema: schema.to_string(), - schemas: ks + schemas: found .iter() .filter_map(|(k, _)| k.split_once('/').map(|k| k.0)) .collect::>() .join(","), } - .fail()? + .fail(), + } + } + + pub(crate) async fn insert_failover_cache(&self, content: PipelineContent, with_latest: bool) { + let versioned = + generate_pipeline_cache_key(&content.schema, &content.name, Some(content.version)); + let latest = generate_pipeline_cache_key(&content.schema, &content.name, None); + + self.failover_cache.insert(versioned, content.clone()).await; + if with_latest { + self.failover_cache.insert(latest, content).await; + } + } + + /// Dropping the stale `latest` aliases also clears the failover entries, so + /// the new version is written back: an outage before the first read-back + /// would otherwise have nothing to fall back on. + pub(crate) async fn on_pipeline_created(&self, content: PipelineContent) { + self.invalidate(&content.name, None).await; + self.insert_failover_cache(content, true).await; + } + + /// Sweeps every schema and all three caches: the `latest` alias always, + /// plus `version` when given. + pub(crate) async fn invalidate(&self, name: &str, version: PipelineVersion) { + let mut suffixes = vec![generate_pipeline_cache_key_suffix(name, None)]; + if version.is_some() { + suffixes.push(generate_pipeline_cache_key_suffix(name, version)); + } + + let ks = self + .pipelines + .iter() + .map(|(k, _)| k) + .chain(self.original_pipelines.iter().map(|(k, _)| k)) + .chain(self.failover_cache.iter().map(|(k, _)| k)) + .filter(|k| suffixes.iter().any(|suffix| k.ends_with(suffix))) + .collect::>(); + + for k in ks { + let k = k.as_str(); + self.pipelines.invalidate(k).await; + self.original_pipelines.invalidate(k).await; + self.failover_cache.invalidate(k).await; } } } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::sync::Barrier; + + use super::*; + + /// Stored under the empty schema, i.e. visible from every schema. + fn content_at(version: i64) -> PipelineContent { + PipelineContent { + name: "p".to_string(), + content: "transform:".to_string(), + version: TimestampNanosecond::new(version), + schema: EMPTY_SCHEMA_NAME.to_string(), + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_misses_run_one_loader() { + const CONCURRENCY: usize = 8; + + let cache = Arc::new(PipelineCache::new(Duration::from_secs(60))); + let loads = Arc::new(AtomicUsize::new(0)); + let barrier = Arc::new(Barrier::new(CONCURRENCY)); + + let handles = (0..CONCURRENCY) + .map(|_| { + let (cache, loads, barrier) = (cache.clone(), loads.clone(), barrier.clone()); + tokio::spawn(async move { + barrier.wait().await; + cache + .get_pipeline_str_with("db", "p", None, async { + loads.fetch_add(1, Ordering::SeqCst); + // Hold the loader open so every caller is waiting on it. + tokio::time::sleep(Duration::from_millis(100)).await; + Ok(content_at(1)) + }) + .await + .unwrap() + }) + }) + .collect::>(); + + for handle in handles { + assert_eq!(handle.await.unwrap(), content_at(1)); + } + assert_eq!(loads.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_delete_drops_version_pinned_entry() { + let cache = PipelineCache::new(Duration::from_secs(60)); + let content = content_at(1); + let version = Some(content.version); + + cache + .get_pipeline_str_with("db", "p", version, async { Ok(content.clone()) }) + .await + .unwrap(); + + cache.invalidate("p", version).await; + + let loads = AtomicUsize::new(0); + cache + .get_pipeline_str_with("db", "p", version, async { + loads.fetch_add(1, Ordering::SeqCst); + Ok(content.clone()) + }) + .await + .unwrap(); + assert_eq!(loads.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_create_drops_stale_latest_and_primes_failover() { + let cache = PipelineCache::new(Duration::from_secs(60)); + let v2 = content_at(2); + + cache + .get_pipeline_str_with("a", "p", None, async { Ok(content_at(1)) }) + .await + .unwrap(); + cache.insert_failover_cache(content_at(1), true).await; + + cache.on_pipeline_created(v2.clone()).await; + + let loads = AtomicUsize::new(0); + let cached = cache + .get_pipeline_str_with("a", "p", None, async { + loads.fetch_add(1, Ordering::SeqCst); + Ok(v2.clone()) + }) + .await + .unwrap(); + assert_eq!(loads.load(Ordering::SeqCst), 1); + assert_eq!(cached.version, v2.version); + + let failover = cache.get_failover_cache("b", "p", None).await.unwrap(); + assert_eq!(failover.map(|c| c.version), Some(v2.version)); + } + + #[tokio::test] + async fn test_failover_serves_global_pipeline_to_unwarmed_schema() { + let cache = PipelineCache::new(Duration::from_secs(60)); + let content = content_at(1); + + cache.insert_failover_cache(content.clone(), true).await; + + let found = cache.get_failover_cache("b", "p", None).await.unwrap(); + assert_eq!(found, Some(content.clone())); + + // A same-named pipeline under another schema must not shadow the global one. + let schema_local = PipelineContent { + schema: "x".to_string(), + ..content_at(2) + }; + cache.insert_failover_cache(schema_local, true).await; + + let found = cache.get_failover_cache("b", "p", None).await.unwrap(); + assert_eq!(found, Some(content)); + } +} diff --git a/src/pipeline/src/manager/pipeline_operator.rs b/src/pipeline/src/manager/pipeline_operator.rs index 4f60ce4e40..c9731b36f2 100644 --- a/src/pipeline/src/manager/pipeline_operator.rs +++ b/src/pipeline/src/manager/pipeline_operator.rs @@ -14,7 +14,7 @@ use std::collections::HashMap; use std::sync::{Arc, RwLock}; -use std::time::Instant; +use std::time::{Duration, Instant}; use api::v1::CreateTableExpr; use catalog::{CatalogManagerRef, RegisterSystemTableRequest}; @@ -39,6 +39,7 @@ use crate::metrics::{ METRIC_PIPELINE_CREATE_HISTOGRAM, METRIC_PIPELINE_DELETE_HISTOGRAM, METRIC_PIPELINE_RETRIEVE_HISTOGRAM, }; +use crate::options::PipelineOptions; use crate::table::{PIPELINE_TABLE_NAME, PipelineTable}; /// PipelineOperator is responsible for managing pipelines. @@ -55,6 +56,7 @@ pub struct PipelineOperator { catalog_manager: CatalogManagerRef, query_engine: QueryEngineRef, tables: RwLock>, + cache_ttl: Duration, } impl PipelineOperator { @@ -97,6 +99,7 @@ impl PipelineOperator { self.statement_executor.clone(), table, self.query_engine.clone(), + self.cache_ttl, )), ); } @@ -172,6 +175,7 @@ impl PipelineOperator { statement_executor: StatementExecutorRef, catalog_manager: CatalogManagerRef, query_engine: QueryEngineRef, + options: &PipelineOptions, ) -> Self { Self { inserter, @@ -179,6 +183,7 @@ impl PipelineOperator { catalog_manager, tables: RwLock::new(HashMap::new()), query_engine, + cache_ttl: options.cache_ttl, } } diff --git a/src/pipeline/src/manager/table.rs b/src/pipeline/src/manager/table.rs index 8720446a64..0b81493b14 100644 --- a/src/pipeline/src/manager/table.rs +++ b/src/pipeline/src/manager/table.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::sync::Arc; +use std::time::Duration; use api::v1::value::ValueData; use api::v1::{ @@ -76,13 +77,14 @@ impl PipelineTable { statement_executor: StatementExecutorRef, table: TableRef, query_engine: QueryEngineRef, + cache_ttl: Duration, ) -> Self { Self { inserter, statement_executor, table, query_engine, - cache: PipelineCache::new(), + cache: PipelineCache::new(cache_ttl), } } @@ -262,21 +264,12 @@ impl PipelineTable { name: &str, input_version: PipelineVersion, ) -> Result> { - if let Some(pipeline) = self.cache.get_pipeline_cache(schema, name, input_version)? { - return Ok(pipeline); - } - - let pipeline_content = self.get_pipeline_str(schema, name, input_version).await?; - let compiled_pipeline = Arc::new(Self::compile_pipeline(&pipeline_content.content)?); - - self.cache.insert_pipeline_cache( - &pipeline_content.schema, - name, - Some(pipeline_content.version), - compiled_pipeline.clone(), - input_version.is_none(), - ); - Ok(compiled_pipeline) + self.cache + .get_pipeline_with(schema, name, input_version, async { + let pipeline_content = self.get_pipeline_str(schema, name, input_version).await?; + Ok(Arc::new(Self::compile_pipeline(&pipeline_content.content)?)) + }) + .await } /// Get a original pipeline by name. @@ -287,13 +280,19 @@ impl PipelineTable { name: &str, input_version: PipelineVersion, ) -> Result { - if let Some(pipeline) = self - .cache - .get_pipeline_str_cache(schema, name, input_version)? - { - return Ok(pipeline); - } + self.cache + .get_pipeline_str_with(schema, name, input_version, async { + self.load_pipeline_str(schema, name, input_version).await + }) + .await + } + async fn load_pipeline_str( + &self, + schema: &str, + name: &str, + input_version: PipelineVersion, + ) -> Result { let mut pipeline_vec; match self.find_pipeline(name, input_version).await { Ok(p) => { @@ -312,7 +311,8 @@ impl PipelineTable { .inc(); return self .cache - .get_failover_cache(schema, name, input_version)? + .get_failover_cache(schema, name, input_version) + .await? .context(PipelineNotFoundSnafu { name, version: input_version, @@ -338,7 +338,8 @@ impl PipelineTable { let pipeline_content = pipeline_vec.remove(0); self.cache - .insert_pipeline_str_cache(&pipeline_content, input_version.is_none()); + .insert_failover_cache(pipeline_content.clone(), input_version.is_none()) + .await; return Ok(pipeline_content); } @@ -359,12 +360,12 @@ impl PipelineTable { })?; self.cache - .insert_pipeline_str_cache(&pipeline_content, input_version.is_none()); + .insert_failover_cache(pipeline_content.clone(), input_version.is_none()) + .await; Ok(pipeline_content) } /// Insert a pipeline into the pipeline table and compile it. - /// The compiled pipeline will be inserted into the cache. /// Newly created pipelines will be saved under empty schema. pub async fn insert_and_compile( &self, @@ -378,25 +379,14 @@ impl PipelineTable { .insert_pipeline_to_pipeline_table(name, content_type, pipeline) .await?; - { - self.cache.insert_pipeline_cache( - EMPTY_SCHEMA_NAME, - name, - Some(TimestampNanosecond(version)), - compiled_pipeline.clone(), - true, - ); - - let pipeline_content = PipelineContent { + self.cache + .on_pipeline_created(PipelineContent { name: name.to_string(), content: pipeline.to_string(), version: TimestampNanosecond(version), schema: EMPTY_SCHEMA_NAME.to_string(), - }; - - self.cache - .insert_pipeline_str_cache(&pipeline_content, true); - } + }) + .await; Ok((version, compiled_pipeline)) } @@ -464,8 +454,7 @@ impl PipelineTable { output ); - // remove cache with version and latest - self.cache.remove_cache(name, version); + self.cache.invalidate(name, version).await; Ok(Some(())) } diff --git a/src/pipeline/src/options.rs b/src/pipeline/src/options.rs new file mode 100644 index 0000000000..6e34f0ce93 --- /dev/null +++ b/src/pipeline/src/options.rs @@ -0,0 +1,35 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(default)] +pub struct PipelineOptions { + /// Time to live of the frontend-local pipeline cache. A pipeline created or + /// deleted on another frontend takes effect on this one after at most this + /// duration. Default: "10s". + #[serde(with = "humantime_serde")] + pub cache_ttl: Duration, +} + +impl Default for PipelineOptions { + fn default() -> Self { + Self { + cache_ttl: Duration::from_secs(10), + } + } +} diff --git a/src/standalone/Cargo.toml b/src/standalone/Cargo.toml index 836f891fa4..3dcb1d5f10 100644 --- a/src/standalone/Cargo.toml +++ b/src/standalone/Cargo.toml @@ -35,6 +35,7 @@ frontend.workspace = true hostname.workspace = true log-store.workspace = true mito2.workspace = true +pipeline.workspace = true query.workspace = true serde.workspace = true servers.workspace = true diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index 4f31ac2ebd..9f20d73952 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -28,6 +28,7 @@ use frontend::service_config::{ PromStoreOptions, }; use mito2::config::MitoConfig; +use pipeline::PipelineOptions; use query::options::QueryOptions; use serde::{Deserialize, Serialize}; use servers::grpc::GrpcOptions; @@ -73,6 +74,8 @@ pub struct StandaloneOptions { pub slow_query: SlowQueryOptions, pub query: QueryOptions, pub memory: MemoryOptions, + /// The pipeline options. + pub pipeline: PipelineOptions, /// The event recorder options. pub event_recorder: EventRecorderOptions, /// Environment variable keys to read and report in heartbeat messages. @@ -114,6 +117,7 @@ impl Default for StandaloneOptions { slow_query: SlowQueryOptions::default(), query: QueryOptions::default(), memory: MemoryOptions::default(), + pipeline: PipelineOptions::default(), event_recorder: EventRecorderOptions::default(), heartbeat_env_vars: vec![], } @@ -162,6 +166,7 @@ impl StandaloneOptions { user_provider: cloned_opts.user_provider, query: cloned_opts.query, slow_query: cloned_opts.slow_query, + pipeline: cloned_opts.pipeline, event_recorder: cloned_opts.event_recorder, heartbeat_env_vars: cloned_opts.heartbeat_env_vars.clone(), ..Default::default() diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index cbb7b8374f..8312d0d5aa 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -2408,6 +2408,9 @@ experimental_memory_pool_policy = "greedy" [memory] enable_heap_profiling = true +[pipeline] +cache_ttl = "10s" + [event_recorder] ttl = "2months 29days 2h 52m 48s" "#,