1use std::collections::HashMap;
16use std::fmt::{Display, Formatter};
17use std::net::SocketAddr;
18use std::sync::{Arc, RwLock};
19use std::time::Duration;
20
21use api::v1::ExplainOptions;
22use api::v1::region::RegionRequestHeader;
23use arc_swap::ArcSwap;
24use auth::UserInfoRef;
25use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
26use common_catalog::{build_db_string, parse_catalog_and_schema_from_db_string};
27use common_recordbatch::cursor::RecordBatchStreamCursor;
28use common_telemetry::warn;
29use common_time::Timezone;
30use common_time::timezone::parse_timezone;
31use datafusion_common::config::ConfigOptions;
32use derive_builder::Builder;
33use sql::dialect::{Dialect, GenericDialect, GreptimeDbDialect, MySqlDialect, PostgreSqlDialect};
34
35use crate::protocol_ctx::ProtocolCtx;
36use crate::session_config::{PGByteaOutputValue, PGDateOrder, PGDateTimeStyle};
37use crate::{MutableInner, ReadPreference};
38
39pub type QueryContextRef = Arc<QueryContext>;
40pub type ConnInfoRef = Arc<ConnInfo>;
41
42const CURSOR_COUNT_WARNING_LIMIT: usize = 10;
43
44#[derive(Debug, Builder, Clone)]
45#[builder(pattern = "owned")]
46#[builder(build_fn(skip))]
47pub struct QueryContext {
48 current_catalog: String,
49 snapshot_seqs: Arc<RwLock<HashMap<u64, u64>>>,
53 sst_min_sequences: Arc<RwLock<HashMap<u64, u64>>>,
55 #[builder(default)]
57 mutable_session_data: Arc<RwLock<MutableInner>>,
58 #[builder(default)]
59 mutable_query_context_data: Arc<RwLock<QueryContextMutableFields>>,
60 sql_dialect: Arc<dyn Dialect + Send + Sync>,
61 #[builder(default)]
62 extensions: HashMap<String, String>,
63 #[builder(default)]
65 configuration_parameter: Arc<ConfigurationVariables>,
66 #[builder(default)]
68 channel: Channel,
69 #[builder(default)]
71 process_id: u32,
72 #[builder(default)]
74 conn_info: ConnInfo,
75 #[builder(default)]
77 protocol_ctx: ProtocolCtx,
78}
79
80#[derive(Debug, Builder, Clone, Default)]
82pub struct QueryContextMutableFields {
83 warning: Option<String>,
84 explain_format: Option<String>,
86 explain_options: Option<ExplainOptions>,
88}
89
90impl Display for QueryContext {
91 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
92 write!(
93 f,
94 "QueryContext{{catalog: {}, schema: {}}}",
95 self.current_catalog(),
96 self.current_schema()
97 )
98 }
99}
100
101impl QueryContextBuilder {
102 pub fn current_schema(mut self, schema: String) -> Self {
103 if self.mutable_session_data.is_none() {
104 self.mutable_session_data = Some(Arc::new(RwLock::new(MutableInner::default())));
105 }
106
107 self.mutable_session_data
109 .as_mut()
110 .unwrap()
111 .write()
112 .unwrap()
113 .schema = schema;
114 self
115 }
116
117 pub fn timezone(mut self, timezone: Timezone) -> Self {
118 if self.mutable_session_data.is_none() {
119 self.mutable_session_data = Some(Arc::new(RwLock::new(MutableInner::default())));
120 }
121
122 self.mutable_session_data
123 .as_mut()
124 .unwrap()
125 .write()
126 .unwrap()
127 .timezone = timezone;
128 self
129 }
130
131 pub fn explain_options(mut self, explain_options: Option<ExplainOptions>) -> Self {
132 self.mutable_query_context_data
133 .get_or_insert_default()
134 .write()
135 .unwrap()
136 .explain_options = explain_options;
137 self
138 }
139
140 pub fn read_preference(mut self, read_preference: ReadPreference) -> Self {
141 self.mutable_session_data
142 .get_or_insert_default()
143 .write()
144 .unwrap()
145 .read_preference = read_preference;
146 self
147 }
148}
149
150impl From<&RegionRequestHeader> for QueryContext {
151 fn from(value: &RegionRequestHeader) -> Self {
152 if let Some(ctx) = &value.query_context {
153 ctx.clone().into()
154 } else {
155 QueryContextBuilder::default().build()
156 }
157 }
158}
159
160impl From<api::v1::QueryContext> for QueryContext {
161 fn from(ctx: api::v1::QueryContext) -> Self {
162 let sequences = ctx.snapshot_seqs.as_ref();
163 QueryContextBuilder::default()
164 .current_catalog(ctx.current_catalog)
165 .current_schema(ctx.current_schema)
166 .timezone(parse_timezone(Some(&ctx.timezone)))
167 .extensions(ctx.extensions)
168 .channel(ctx.channel.into())
169 .snapshot_seqs(Arc::new(RwLock::new(
170 sequences
171 .map(|x| x.snapshot_seqs.clone())
172 .unwrap_or_default(),
173 )))
174 .sst_min_sequences(Arc::new(RwLock::new(
175 sequences
176 .map(|x| x.sst_min_sequences.clone())
177 .unwrap_or_default(),
178 )))
179 .explain_options(ctx.explain)
180 .build()
181 }
182}
183
184impl From<QueryContext> for api::v1::QueryContext {
185 fn from(
186 QueryContext {
187 current_catalog,
188 mutable_session_data: mutable_inner,
189 extensions,
190 channel,
191 snapshot_seqs,
192 sst_min_sequences,
193 mutable_query_context_data,
194 ..
195 }: QueryContext,
196 ) -> Self {
197 let explain = mutable_query_context_data.read().unwrap().explain_options;
198 let mutable_inner = mutable_inner.read().unwrap();
199 api::v1::QueryContext {
200 current_catalog,
201 current_schema: mutable_inner.schema.clone(),
202 timezone: mutable_inner.timezone.to_string(),
203 extensions,
204 channel: channel as u32,
205 snapshot_seqs: Some(api::v1::SnapshotSequences {
206 snapshot_seqs: snapshot_seqs.read().unwrap().clone(),
207 sst_min_sequences: sst_min_sequences.read().unwrap().clone(),
208 }),
209 explain,
210 }
211 }
212}
213
214impl From<&QueryContext> for api::v1::QueryContext {
215 fn from(ctx: &QueryContext) -> Self {
216 ctx.clone().into()
217 }
218}
219
220impl QueryContext {
221 pub fn arc() -> QueryContextRef {
222 Arc::new(QueryContextBuilder::default().build())
223 }
224
225 pub fn create_config_options(&self) -> ConfigOptions {
227 let mut config = ConfigOptions::default();
228 config.execution.time_zone = self.timezone().to_string();
229 config
230 }
231
232 pub fn with(catalog: &str, schema: &str) -> QueryContext {
233 QueryContextBuilder::default()
234 .current_catalog(catalog.to_string())
235 .current_schema(schema.to_string())
236 .build()
237 }
238
239 pub fn with_channel(catalog: &str, schema: &str, channel: Channel) -> QueryContext {
240 QueryContextBuilder::default()
241 .current_catalog(catalog.to_string())
242 .current_schema(schema.to_string())
243 .channel(channel)
244 .build()
245 }
246
247 pub fn with_db_name(db_name: Option<&str>) -> QueryContext {
248 let (catalog, schema) = db_name
249 .map(|db| {
250 let (catalog, schema) = parse_catalog_and_schema_from_db_string(db);
251 (catalog, schema)
252 })
253 .unwrap_or_else(|| {
254 (
255 DEFAULT_CATALOG_NAME.to_string(),
256 DEFAULT_SCHEMA_NAME.to_string(),
257 )
258 });
259 QueryContextBuilder::default()
260 .current_catalog(catalog)
261 .current_schema(schema.to_string())
262 .build()
263 }
264
265 pub fn current_schema(&self) -> String {
266 self.mutable_session_data.read().unwrap().schema.clone()
267 }
268
269 pub fn set_current_schema(&self, new_schema: &str) {
270 self.mutable_session_data.write().unwrap().schema = new_schema.to_string();
271 }
272
273 pub fn current_catalog(&self) -> &str {
274 &self.current_catalog
275 }
276
277 pub fn set_current_catalog(&mut self, new_catalog: &str) {
278 self.current_catalog = new_catalog.to_string();
279 }
280
281 pub fn sql_dialect(&self) -> &(dyn Dialect + Send + Sync) {
282 &*self.sql_dialect
283 }
284
285 pub fn get_db_string(&self) -> String {
286 let catalog = self.current_catalog();
287 let schema = self.current_schema();
288 build_db_string(catalog, &schema)
289 }
290
291 pub fn timezone(&self) -> Timezone {
292 self.mutable_session_data.read().unwrap().timezone.clone()
293 }
294
295 pub fn set_timezone(&self, timezone: Timezone) {
296 self.mutable_session_data.write().unwrap().timezone = timezone;
297 }
298
299 pub fn read_preference(&self) -> ReadPreference {
300 self.mutable_session_data.read().unwrap().read_preference
301 }
302
303 pub fn set_read_preference(&self, read_preference: ReadPreference) {
304 self.mutable_session_data.write().unwrap().read_preference = read_preference;
305 }
306
307 pub fn current_user(&self) -> UserInfoRef {
308 self.mutable_session_data.read().unwrap().user_info.clone()
309 }
310
311 pub fn set_current_user(&self, user: UserInfoRef) {
312 self.mutable_session_data.write().unwrap().user_info = user;
313 }
314
315 pub fn set_extension<S1: Into<String>, S2: Into<String>>(&mut self, key: S1, value: S2) {
316 self.extensions.insert(key.into(), value.into());
317 }
318
319 pub fn extension<S: AsRef<str>>(&self, key: S) -> Option<&str> {
320 self.extensions.get(key.as_ref()).map(|v| v.as_str())
321 }
322
323 pub fn extensions(&self) -> HashMap<String, String> {
324 self.extensions.clone()
325 }
326
327 pub fn quote_style(&self) -> char {
329 if self.sql_dialect().is_delimited_identifier_start('"') {
330 '"'
331 } else if self.sql_dialect().is_delimited_identifier_start('\'') {
332 '\''
333 } else {
334 '`'
335 }
336 }
337
338 pub fn configuration_parameter(&self) -> &ConfigurationVariables {
339 &self.configuration_parameter
340 }
341
342 pub fn channel(&self) -> Channel {
343 self.channel
344 }
345
346 pub fn set_channel(&mut self, channel: Channel) {
347 self.channel = channel;
348 }
349
350 pub fn warning(&self) -> Option<String> {
351 self.mutable_query_context_data
352 .read()
353 .unwrap()
354 .warning
355 .clone()
356 }
357
358 pub fn set_warning(&self, msg: String) {
359 self.mutable_query_context_data.write().unwrap().warning = Some(msg);
360 }
361
362 pub fn explain_format(&self) -> Option<String> {
363 self.mutable_query_context_data
364 .read()
365 .unwrap()
366 .explain_format
367 .clone()
368 }
369
370 pub fn set_explain_format(&self, format: String) {
371 self.mutable_query_context_data
372 .write()
373 .unwrap()
374 .explain_format = Some(format);
375 }
376
377 pub fn explain_verbose(&self) -> bool {
378 self.mutable_query_context_data
379 .read()
380 .unwrap()
381 .explain_options
382 .map(|opts| opts.verbose)
383 .unwrap_or(false)
384 }
385
386 pub fn set_explain_verbose(&self, verbose: bool) {
387 self.mutable_query_context_data
388 .write()
389 .unwrap()
390 .explain_options
391 .get_or_insert_default()
392 .verbose = verbose;
393 }
394
395 pub fn query_timeout(&self) -> Option<Duration> {
396 self.mutable_session_data.read().unwrap().query_timeout
397 }
398
399 pub fn query_timeout_as_millis(&self) -> u128 {
400 let timeout = self.mutable_session_data.read().unwrap().query_timeout;
401 if let Some(t) = timeout {
402 return t.as_millis();
403 }
404 0
405 }
406
407 pub fn set_query_timeout(&self, timeout: Duration) {
408 self.mutable_session_data.write().unwrap().query_timeout = Some(timeout);
409 }
410
411 pub fn insert_cursor(&self, name: String, rb: RecordBatchStreamCursor) {
412 let mut guard = self.mutable_session_data.write().unwrap();
413 guard.cursors.insert(name, Arc::new(rb));
414
415 let cursor_count = guard.cursors.len();
416 if cursor_count > CURSOR_COUNT_WARNING_LIMIT {
417 warn!("Current connection has {} open cursors", cursor_count);
418 }
419 }
420
421 pub fn remove_cursor(&self, name: &str) {
422 let mut guard = self.mutable_session_data.write().unwrap();
423 guard.cursors.remove(name);
424 }
425
426 pub fn get_cursor(&self, name: &str) -> Option<Arc<RecordBatchStreamCursor>> {
427 let guard = self.mutable_session_data.read().unwrap();
428 let rb = guard.cursors.get(name);
429 rb.cloned()
430 }
431
432 pub fn snapshots(&self) -> HashMap<u64, u64> {
433 self.snapshot_seqs.read().unwrap().clone()
434 }
435
436 pub fn get_snapshot(&self, region_id: u64) -> Option<u64> {
437 self.snapshot_seqs.read().unwrap().get(®ion_id).cloned()
438 }
439
440 pub fn auto_string_to_numeric(&self) -> bool {
442 matches!(self.channel, Channel::Mysql)
443 }
444
445 pub fn sst_min_sequence(&self, region_id: u64) -> Option<u64> {
447 self.sst_min_sequences
448 .read()
449 .unwrap()
450 .get(®ion_id)
451 .copied()
452 }
453
454 pub fn process_id(&self) -> u32 {
455 self.process_id
456 }
457
458 pub fn conn_info(&self) -> &ConnInfo {
460 &self.conn_info
461 }
462
463 pub fn protocol_ctx(&self) -> &ProtocolCtx {
464 &self.protocol_ctx
465 }
466
467 pub fn set_protocol_ctx(&mut self, protocol_ctx: ProtocolCtx) {
468 self.protocol_ctx = protocol_ctx;
469 }
470}
471
472impl QueryContextBuilder {
473 pub fn build(self) -> QueryContext {
474 let channel = self.channel.unwrap_or_default();
475 QueryContext {
476 current_catalog: self
477 .current_catalog
478 .unwrap_or_else(|| DEFAULT_CATALOG_NAME.to_string()),
479 snapshot_seqs: self.snapshot_seqs.unwrap_or_default(),
480 sst_min_sequences: self.sst_min_sequences.unwrap_or_default(),
481 mutable_session_data: self.mutable_session_data.unwrap_or_default(),
482 mutable_query_context_data: self.mutable_query_context_data.unwrap_or_default(),
483 sql_dialect: self
484 .sql_dialect
485 .unwrap_or_else(|| Arc::new(GreptimeDbDialect {})),
486 extensions: self.extensions.unwrap_or_default(),
487 configuration_parameter: self
488 .configuration_parameter
489 .unwrap_or_else(|| Arc::new(ConfigurationVariables::default())),
490 channel,
491 process_id: self.process_id.unwrap_or_default(),
492 conn_info: self.conn_info.unwrap_or_default(),
493 protocol_ctx: self.protocol_ctx.unwrap_or_default(),
494 }
495 }
496
497 pub fn set_extension(mut self, key: String, value: String) -> Self {
498 self.extensions
499 .get_or_insert_with(HashMap::new)
500 .insert(key, value);
501 self
502 }
503}
504
505#[derive(Debug, Clone, Default)]
506pub struct ConnInfo {
507 pub client_addr: Option<SocketAddr>,
508 pub channel: Channel,
509}
510
511impl Display for ConnInfo {
512 fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
513 write!(
514 f,
515 "{}[{}]",
516 self.channel,
517 self.client_addr
518 .map(|addr| addr.to_string())
519 .as_deref()
520 .unwrap_or("unknown client addr")
521 )
522 }
523}
524
525impl ConnInfo {
526 pub fn new(client_addr: Option<SocketAddr>, channel: Channel) -> Self {
527 Self {
528 client_addr,
529 channel,
530 }
531 }
532}
533
534#[derive(Debug, PartialEq, Default, Clone, Copy)]
535#[repr(u8)]
536pub enum Channel {
537 #[default]
538 Unknown = 0,
539
540 Mysql = 1,
541 Postgres = 2,
542 HttpSql = 3,
543 Prometheus = 4,
544 Otlp = 5,
545 Grpc = 6,
546 Influx = 7,
547 Opentsdb = 8,
548 Loki = 9,
549 Elasticsearch = 10,
550 Jaeger = 11,
551 Log = 12,
552 Promql = 13,
553}
554
555impl From<u32> for Channel {
556 fn from(value: u32) -> Self {
557 match value {
558 1 => Self::Mysql,
559 2 => Self::Postgres,
560 3 => Self::HttpSql,
561 4 => Self::Prometheus,
562 5 => Self::Otlp,
563 6 => Self::Grpc,
564 7 => Self::Influx,
565 8 => Self::Opentsdb,
566 9 => Self::Loki,
567 10 => Self::Elasticsearch,
568 11 => Self::Jaeger,
569 12 => Self::Log,
570 13 => Self::Promql,
571 _ => Self::Unknown,
572 }
573 }
574}
575
576impl Channel {
577 pub fn dialect(&self) -> Arc<dyn Dialect + Send + Sync> {
578 match self {
579 Channel::Mysql => Arc::new(MySqlDialect {}),
580 Channel::Postgres => Arc::new(PostgreSqlDialect {}),
581 _ => Arc::new(GenericDialect {}),
582 }
583 }
584}
585
586impl Display for Channel {
587 fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
588 write!(f, "{}", self.as_ref())
589 }
590}
591
592impl AsRef<str> for Channel {
593 fn as_ref(&self) -> &str {
594 match self {
595 Channel::Mysql => "mysql",
596 Channel::Postgres => "postgres",
597 Channel::HttpSql => "httpsql",
598 Channel::Prometheus => "prometheus",
599 Channel::Otlp => "otlp",
600 Channel::Grpc => "grpc",
601 Channel::Influx => "influx",
602 Channel::Opentsdb => "opentsdb",
603 Channel::Loki => "loki",
604 Channel::Elasticsearch => "elasticsearch",
605 Channel::Jaeger => "jaeger",
606 Channel::Log => "log",
607 Channel::Promql => "promql",
608 Channel::Unknown => "unknown",
609 }
610 }
611}
612
613#[derive(Default, Debug)]
614pub struct ConfigurationVariables {
615 postgres_bytea_output: ArcSwap<PGByteaOutputValue>,
616 pg_datestyle_format: ArcSwap<(PGDateTimeStyle, PGDateOrder)>,
617 allow_query_fallback: ArcSwap<bool>,
618}
619
620impl Clone for ConfigurationVariables {
621 fn clone(&self) -> Self {
622 Self {
623 postgres_bytea_output: ArcSwap::new(self.postgres_bytea_output.load().clone()),
624 pg_datestyle_format: ArcSwap::new(self.pg_datestyle_format.load().clone()),
625 allow_query_fallback: ArcSwap::new(self.allow_query_fallback.load().clone()),
626 }
627 }
628}
629
630impl ConfigurationVariables {
631 pub fn new() -> Self {
632 Self::default()
633 }
634
635 pub fn set_postgres_bytea_output(&self, value: PGByteaOutputValue) {
636 let _ = self.postgres_bytea_output.swap(Arc::new(value));
637 }
638
639 pub fn postgres_bytea_output(&self) -> Arc<PGByteaOutputValue> {
640 self.postgres_bytea_output.load().clone()
641 }
642
643 pub fn pg_datetime_style(&self) -> Arc<(PGDateTimeStyle, PGDateOrder)> {
644 self.pg_datestyle_format.load().clone()
645 }
646
647 pub fn set_pg_datetime_style(&self, style: PGDateTimeStyle, order: PGDateOrder) {
648 self.pg_datestyle_format.swap(Arc::new((style, order)));
649 }
650
651 pub fn allow_query_fallback(&self) -> bool {
652 **self.allow_query_fallback.load()
653 }
654
655 pub fn set_allow_query_fallback(&self, allow: bool) {
656 self.allow_query_fallback.swap(Arc::new(allow));
657 }
658}
659
660#[cfg(test)]
661mod test {
662 use common_catalog::consts::DEFAULT_CATALOG_NAME;
663
664 use super::*;
665 use crate::Session;
666 use crate::context::Channel;
667
668 #[test]
669 fn test_session() {
670 let session = Session::new(
671 Some("127.0.0.1:9000".parse().unwrap()),
672 Channel::Mysql,
673 Default::default(),
674 100,
675 );
676 assert_eq!(session.user_info().username(), "greptime");
678
679 assert_eq!(session.conn_info().channel, Channel::Mysql);
681 let client_addr = session.conn_info().client_addr.as_ref().unwrap();
682 assert_eq!(client_addr.ip().to_string(), "127.0.0.1");
683 assert_eq!(client_addr.port(), 9000);
684
685 assert_eq!("mysql[127.0.0.1:9000]", session.conn_info().to_string());
686 assert_eq!(100, session.process_id());
687 }
688
689 #[test]
690 fn test_context_db_string() {
691 let context = QueryContext::with("a0b1c2d3", "test");
692 assert_eq!("a0b1c2d3-test", context.get_db_string());
693
694 let context = QueryContext::with(DEFAULT_CATALOG_NAME, "test");
695 assert_eq!("test", context.get_db_string());
696 }
697}