From ed1f2d9f4e6be2642abaf5e04887b49511cc7a77 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Fri, 4 Sep 2026 05:14:50 +0000 Subject: [PATCH] fix(pipeline): coalesce concurrent pipeline cache misses (#9022) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(pipeline): coalesce concurrent pipeline cache misses The pipeline cache reads with a plain `moka::sync::Cache::get` and falls through to a distributed query on a miss, so when the 10s TTL expires every in-flight write request on a frontend issues its own scan of the single-region `greptime_private.pipelines` table. Concurrent scans per expiry scale with write QPS, and every frontend's burst lands on the same datanode. A user running high-throughput ingestion through a pipeline saw that datanode overloaded. Switch to `moka::future::Cache::try_get_with` so concurrent misses on the same key share one loader. This requires a single-key lookup, so cache entries are now keyed by the requested schema rather than the schema the pipeline is stored under; resolving a request to a stored schema stays in the loader, which is the authoritative path and already handles the empty-schema and multi-schema cases. A lookup for a schema not yet cached costs one extra read, now protected from amplification by the coalescing it enables. `remove_cache` previously only walked the compiled-pipeline cache, so an entry populated by `get_pipeline_str` alone (the pipeline read API) survived deletion until it expired. It now walks all three caches. Also make the TTL configurable as `pipeline.cache_ttl`, default unchanged at 10s. The TTL is what propagates a pipeline change to other frontends, so raising it trades staleness for fewer reads. Refs #9021 Signed-off-by: Dennis Zhuang * fix(pipeline): restore cross-schema semantics broken by the new cache key Keying cache entries by the requested schema dropped two behaviours that the previous stored-schema key provided for free. Creating a new version only wrote the creating request's schema, so another schema on the same frontend kept serving its cached `latest` — an older version — until the entry expired. Since the whole point of making the TTL configurable is to let operators raise it, that window is not bounded by anything useful. Creation now invalidates every schema's `latest` alias for that name before priming the cache, leaving the version-pinned keys alone. The failover cache lost its reach across schemas the same way: a global pipeline (stored under the empty schema) loaded by schema A was cached under `A`, so schema B using it for the first time while the pipeline table was down missed and failed ingestion. The failover cache has no loader and so is not subject to the single-key model of `try_get_with`; it keeps the stored-schema key and the empty-schema-first resolution. Signed-off-by: Dennis Zhuang * refactor(pipeline): drop cache priming on create and fold the sweep helpers Priming the cache on create saved one read on a low-frequency operation and cost a concept: entries were written under the creating request's schema while `PipelineContent.schema` said empty, so the two schemas in play disagreed. Invalidating the `latest` aliases is required regardless — that is what makes a new version visible to other schemas — so dropping the priming loses only the saved read, which coalescing now protects anyway. `insert_and_compile` no longer needs the caller's schema. `remove_cache` and the create-time invalidation collapse into one `invalidate(name, version)`; `None` sweeps only the `latest` aliases, which is exactly what creation wants. That leaves `invalidate_by_suffixes` and `cache_keys` with a single caller each, so both are inlined. Drop the `PipelineOptions` humantime test: `load_config_test` loads both example TOMLs, which now carry `cache_ttl = "10s"`, and would fail the same way if the serde attribute were lost. The `toml` dev-dependency goes with it. The two invalidation tests are now checked to be orthogonal: removing the version suffix fails only the delete test, and sweeping just the compiled cache fails both. Signed-off-by: Dennis Zhuang * fix(pipeline): keep failover populated across a create The `latest` sweep on create clears the failover cache along with the loaded ones, and after dropping the priming there was nothing writing it back. An outage between the create and the first read-back left neither `latest` nor the explicit version with anything to fall back on, failing ingestion — worse than before, since the previous version's failover entry was swept too. Creation now goes through `PipelineCache::on_pipeline_created`, which pairs the sweep with a failover write of the new empty-schema definition. The two must happen together, so they live behind one method rather than at the call site. Also commit the Cargo.lock entry for the dropped `toml` dev-dependency, and trim the comments added over the last few commits down to what the code does not already say. Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang --- Cargo.lock | 2 + config/config.md | 4 + config/frontend.example.toml | 6 + config/standalone.example.toml | 5 + src/frontend/src/frontend.rs | 4 + src/frontend/src/instance/builder.rs | 1 + src/pipeline/Cargo.toml | 5 +- src/pipeline/src/error.rs | 10 + src/pipeline/src/lib.rs | 2 + src/pipeline/src/manager/pipeline_cache.rs | 356 +++++++++++------- src/pipeline/src/manager/pipeline_operator.rs | 7 +- src/pipeline/src/manager/table.rs | 75 ++-- src/pipeline/src/options.rs | 35 ++ src/standalone/Cargo.toml | 1 + src/standalone/src/options.rs | 5 + tests-integration/tests/http.rs | 3 + 16 files changed, 338 insertions(+), 183 deletions(-) create mode 100644 src/pipeline/src/options.rs 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" "#,