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