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`,<br/>`alter_database`, `drop_database`, `create_flow`, `drop_flow`,<br/>`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,<br/>`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,<br/>`create_view`, `drop_view`, `admin_function`.<br/>When omitted, all current and future event types are recorded.<br/>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.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>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<br/>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`.<br/>When omitted, all current and future event types are recorded.<br/>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::<StatementExecutorRef>(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<HashMap<String, PipelineTableRef>>,
+    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"
 "#,