risingwave_frontend/
lib.rs1#![allow(clippy::derive_partial_eq_without_eq)]
16#![feature(map_try_insert)]
17#![feature(negative_impls)]
18#![feature(coroutines)]
19#![feature(proc_macro_hygiene, stmt_expr_attributes)]
20#![feature(trait_alias)]
21#![feature(if_let_guard)]
22#![feature(let_chains)]
23#![feature(assert_matches)]
24#![feature(box_patterns)]
25#![feature(macro_metavar_expr)]
26#![feature(min_specialization)]
27#![feature(extend_one)]
28#![feature(type_alias_impl_trait)]
29#![feature(impl_trait_in_assoc_type)]
30#![feature(error_generic_member_access)]
31#![feature(iterator_try_collect)]
32#![feature(used_with_arg)]
33#![feature(try_trait_v2)]
34#![recursion_limit = "256"]
35
36#[cfg(test)]
37risingwave_expr_impl::enable!();
38#[cfg(test)]
39risingwave_batch_executors::enable!();
40
41#[macro_use]
42mod catalog;
43
44use std::collections::HashSet;
45use std::time::Duration;
46
47pub use catalog::TableCatalog;
48mod binder;
49pub use binder::{Binder, bind_data_type};
50pub mod expr;
51pub mod handler;
52pub use handler::PgResponseStream;
53mod observer;
54pub mod optimizer;
55pub use optimizer::{Explain, OptimizerContext, OptimizerContextRef, PlanRef};
56mod planner;
57use pgwire::net::TcpKeepalive;
58pub use planner::Planner;
59mod scheduler;
60pub mod session;
61mod stream_fragmenter;
62use risingwave_common::config::{MetricLevel, OverrideConfig};
63use risingwave_common::util::meta_addr::MetaAddressStrategy;
64use risingwave_common::util::resource_util::memory::system_memory_available_bytes;
65use risingwave_common::util::tokio_util::sync::CancellationToken;
66pub use stream_fragmenter::build_graph;
67mod utils;
68pub use utils::{WithOptions, WithOptionsSecResolved, explain_stream_graph};
69pub(crate) mod error;
70mod meta_client;
71pub mod test_utils;
72mod user;
73pub mod webhook;
74
75pub mod health_service;
76mod monitor;
77
78pub mod rpc;
79mod telemetry;
80
81use std::ffi::OsString;
82use std::iter;
83use std::sync::Arc;
84
85use clap::Parser;
86use pgwire::pg_server::pg_serve;
87use session::SessionManagerImpl;
88
89#[derive(Parser, Clone, Debug, OverrideConfig)]
91#[command(
92 version,
93 about = "The stateless proxy that parses SQL queries and performs planning and optimizations of query jobs"
94)]
95pub struct FrontendOpts {
96 #[clap(long, env = "RW_LISTEN_ADDR", default_value = "0.0.0.0:4566")]
100 pub listen_addr: String,
101
102 #[clap(long, env = "RW_TCP_KEEPALIVE_IDLE_SECS", default_value = "300")]
105 pub tcp_keepalive_idle_secs: usize,
106
107 #[clap(long, env = "RW_ADVERTISE_ADDR")]
112 pub advertise_addr: Option<String>,
113
114 #[clap(long, env = "RW_META_ADDR", default_value = "http://127.0.0.1:5690")]
116 pub meta_addr: MetaAddressStrategy,
117
118 #[clap(
121 long,
122 env = "RW_PROMETHEUS_LISTENER_ADDR",
123 default_value = "127.0.0.1:2222"
124 )]
125 pub prometheus_listener_addr: String,
126
127 #[clap(
128 long,
129 alias = "health-check-listener-addr",
130 env = "RW_HEALTH_CHECK_LISTENER_ADDR",
131 default_value = "0.0.0.0:6786"
132 )]
133 pub frontend_rpc_listener_addr: String,
134
135 #[clap(long, env = "RW_CONFIG_PATH", default_value = "")]
142 pub config_path: String,
143
144 #[clap(long, hide = true, env = "RW_METRICS_LEVEL")]
149 #[override_opts(path = server.metrics_level)]
150 pub metrics_level: Option<MetricLevel>,
151
152 #[clap(long, hide = true, env = "RW_HEAP_PROFILING_DIR")]
154 #[override_opts(path = server.heap_profiling.dir)]
155 pub heap_profiling_dir: Option<String>,
156
157 #[clap(long, hide = true, env = "ENABLE_BARRIER_READ")]
158 #[override_opts(path = batch.enable_barrier_read)]
159 pub enable_barrier_read: Option<bool>,
160
161 #[clap(
163 long,
164 hide = true,
165 env = "RW_TEMP_SECRET_FILE_DIR",
166 default_value = "./secrets"
167 )]
168 pub temp_secret_file_dir: String,
169
170 #[clap(long, env = "RW_FRONTEND_TOTAL_MEMORY_BYTES", default_value_t = default_frontend_total_memory_bytes())]
172 pub frontend_total_memory_bytes: usize,
173
174 #[clap(long, env = "RW_WEBHOOK_LISTEN_ADDR", default_value = "0.0.0.0:4560")]
177 pub webhook_listen_addr: String,
178
179 #[clap(long, env = "RW_SBC_ADDR", default_value = "")]
184 pub serverless_backfill_controller_addr: String,
185}
186
187impl risingwave_common::opts::Opts for FrontendOpts {
188 fn name() -> &'static str {
189 "frontend"
190 }
191
192 fn meta_addr(&self) -> MetaAddressStrategy {
193 self.meta_addr.clone()
194 }
195}
196
197impl Default for FrontendOpts {
198 fn default() -> Self {
199 FrontendOpts::parse_from(iter::empty::<OsString>())
200 }
201}
202
203use std::future::Future;
204use std::pin::Pin;
205
206use pgwire::memory_manager::MessageMemoryManager;
207use pgwire::pg_protocol::{ConnectionContext, TlsConfig};
208
209use crate::session::SESSION_MANAGER;
210
211pub fn start(
213 opts: FrontendOpts,
214 shutdown: CancellationToken,
215) -> Pin<Box<dyn Future<Output = ()> + Send>> {
216 Box::pin(async move {
219 let listen_addr = opts.listen_addr.clone();
220 let webhook_listen_addr = opts.webhook_listen_addr.parse().unwrap();
221 let tcp_keepalive =
222 TcpKeepalive::new().with_time(Duration::from_secs(opts.tcp_keepalive_idle_secs as _));
223
224 let session_mgr = Arc::new(SessionManagerImpl::new(opts).await.unwrap());
225 SESSION_MANAGER.get_or_init(|| session_mgr.clone());
226 let redact_sql_option_keywords = Arc::new(
227 session_mgr
228 .env()
229 .batch_config()
230 .redact_sql_option_keywords
231 .iter()
232 .map(|s| s.to_lowercase())
233 .collect::<HashSet<_>>(),
234 );
235 let frontend_config = &session_mgr.env().frontend_config();
236 let message_memory_manager = Arc::new(MessageMemoryManager::new(
237 frontend_config.max_total_query_size_bytes,
238 frontend_config.min_single_query_size_bytes,
239 frontend_config.max_single_query_size_bytes,
240 ));
241
242 let webhook_service = crate::webhook::WebhookService::new(webhook_listen_addr);
243 let _task = tokio::spawn(webhook_service.serve());
244 pg_serve(
245 &listen_addr,
246 tcp_keepalive,
247 session_mgr.clone(),
248 ConnectionContext {
249 tls_config: TlsConfig::new_default(),
250 redact_sql_option_keywords: Some(redact_sql_option_keywords),
251 message_memory_manager,
252 },
253 shutdown,
254 )
255 .await
256 .unwrap()
257 })
258}
259
260pub fn default_frontend_total_memory_bytes() -> usize {
261 system_memory_available_bytes()
262}