From 64b50966830d8ffc52e94db59477233436bd9a2f Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Tue, 8 Sep 2026 14:37:11 +0800
Subject: [PATCH] fix(pipeline): restore cache TTL configuration for v1.2
Complete the configuration scope of upstream #9022, including frontend and standalone propagation, examples, generated docs, and config serialization coverage.
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
Cargo.lock | 2 ++
config/config.md | 4 +++
config/frontend.example.toml | 6 ++++
config/standalone.example.toml | 5 +++
src/frontend/src/frontend.rs | 20 +++++++++++
src/frontend/src/instance/builder.rs | 1 +
src/pipeline/Cargo.toml | 3 +-
src/pipeline/src/lib.rs | 2 ++
src/pipeline/src/manager/pipeline_cache.rs | 16 ++++-----
src/pipeline/src/manager/pipeline_operator.rs | 7 +++-
src/pipeline/src/manager/table.rs | 4 ++-
src/pipeline/src/options.rs | 35 +++++++++++++++++++
src/standalone/Cargo.toml | 1 +
src/standalone/src/options.rs | 19 ++++++++++
tests-integration/tests/http.rs | 3 ++
15 files changed, 116 insertions(+), 12 deletions(-)
create mode 100644 src/pipeline/src/options.rs
diff --git a/Cargo.lock b/Cargo.lock
index d255729265..327c34d588 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -10790,6 +10790,7 @@ dependencies = [
"enum_dispatch",
"futures",
"greptime-proto",
+ "humantime-serde",
"itertools 0.14.0",
"jsonb",
"jsonpath-rust 0.7.5",
@@ -14238,6 +14239,7 @@ dependencies = [
"hostname 0.4.1",
"log-store",
"mito2",
+ "pipeline",
"query",
"serde",
"servers",
diff --git a/config/config.md b/config/config.md
index 7f9064300a..452a4efaac 100644
--- a/config/config.md
+++ b/config/config.md
@@ -231,6 +231,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`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. |
@@ -368,6 +370,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 30e5f1ff39..b026704a86 100644
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -398,6 +398,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 8ced8b1790..8deb9d7d8d 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -911,6 +911,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 17435c01a1..d56288468e 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![],
}
@@ -217,6 +221,22 @@ mod tests {
let _parsed: FrontendOptions = toml::from_str(&toml_string).unwrap();
}
+ #[test]
+ fn test_pipeline_cache_ttl_toml_roundtrip() {
+ let options: FrontendOptions = toml::from_str("[pipeline]\ncache_ttl = \"30s\"").unwrap();
+ assert_eq!(
+ options.pipeline.cache_ttl,
+ std::time::Duration::from_secs(30)
+ );
+
+ let serialized = toml::to_string(&options).unwrap();
+ let roundtrip: FrontendOptions = toml::from_str(&serialized).unwrap();
+ assert_eq!(
+ roundtrip.pipeline.cache_ttl,
+ std::time::Duration::from_secs(30)
+ );
+ }
+
#[test]
fn test_http_api_server_defaults_on_when_absent() {
// When `[http]` is not present in the config, the dedicated API server is
diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs
index b8e3a9f282..bcb3e3b91e 100644
--- a/src/frontend/src/instance/builder.rs
+++ b/src/frontend/src/instance/builder.rs
@@ -313,6 +313,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 497ce4542c..620ac36ac7 100644
--- a/src/pipeline/Cargo.toml
+++ b/src/pipeline/Cargo.toml
@@ -41,6 +41,7 @@ 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"
@@ -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/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 98105d543f..e60963e7f8 100644
--- a/src/pipeline/src/manager/pipeline_cache.rs
+++ b/src/pipeline/src/manager/pipeline_cache.rs
@@ -27,8 +27,6 @@ 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.
@@ -54,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()
@@ -211,7 +209,7 @@ mod tests {
async fn test_concurrent_misses_run_one_loader() {
const CONCURRENCY: usize = 8;
- let cache = Arc::new(PipelineCache::new());
+ let cache = Arc::new(PipelineCache::new(Duration::from_secs(60)));
let loads = Arc::new(AtomicUsize::new(0));
let barrier = Arc::new(Barrier::new(CONCURRENCY));
@@ -241,7 +239,7 @@ mod tests {
#[tokio::test]
async fn test_delete_drops_version_pinned_entry() {
- let cache = PipelineCache::new();
+ let cache = PipelineCache::new(Duration::from_secs(60));
let content = content_at(1);
let version = Some(content.version);
@@ -265,7 +263,7 @@ mod tests {
#[tokio::test]
async fn test_create_drops_stale_latest_and_primes_failover() {
- let cache = PipelineCache::new();
+ let cache = PipelineCache::new(Duration::from_secs(60));
let v2 = content_at(2);
cache
@@ -293,7 +291,7 @@ mod tests {
#[tokio::test]
async fn test_failover_serves_global_pipeline_to_unwarmed_schema() {
- let cache = PipelineCache::new();
+ let cache = PipelineCache::new(Duration::from_secs(60));
let content = content_at(1);
cache.insert_failover_cache(content.clone(), true).await;
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 d61856dfb3..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),
}
}
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 16479991f4..744b45cab6 100644
--- a/src/standalone/Cargo.toml
+++ b/src/standalone/Cargo.toml
@@ -33,6 +33,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 637715013a..cbd82a6609 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;
@@ -72,6 +73,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.
@@ -112,6 +115,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![],
}
@@ -158,6 +162,7 @@ impl StandaloneOptions {
logging: cloned_opts.logging,
user_provider: cloned_opts.user_provider,
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()
@@ -196,11 +201,25 @@ impl StandaloneOptions {
#[cfg(test)]
mod tests {
use std::sync::Arc;
+ use std::time::Duration;
use common_event_recorder::EventTypeFilter;
use super::*;
+ #[test]
+ fn test_pipeline_cache_ttl_propagates_to_frontend_options() {
+ let default_options: StandaloneOptions = toml::from_str("").unwrap();
+ assert_eq!(default_options.pipeline.cache_ttl, Duration::from_secs(10));
+
+ let options: StandaloneOptions = toml::from_str("[pipeline]\ncache_ttl = \"30s\"").unwrap();
+ assert_eq!(options.pipeline.cache_ttl, Duration::from_secs(30));
+ assert_eq!(
+ options.frontend_options().pipeline.cache_ttl,
+ Duration::from_secs(30)
+ );
+ }
+
#[test]
fn test_event_recorder_event_types_preserve_filter_semantics() {
let all: StandaloneOptions = toml::from_str("").unwrap();
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index 784aa30e7b..739b24e918 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -2333,6 +2333,9 @@ allow_query_fallback = false
[memory]
enable_heap_profiling = true
+[pipeline]
+cache_ttl = "10s"
+
[event_recorder]
ttl = "2months 29days 2h 52m 48s"
"#,