use std::sync::Arc;
use crate::executor::streaming::query_registry::CancelToken;
use crate::executor::streaming::transaction_scope::CancelReason;
use linkrs_core::types::{CharsetInfo, SpaceInfo, Timestamp};
use linkrs_core::IdGenerator;
use super::QueryRequestContext;
pub struct QueryContext {
rctx: Arc<QueryRequestContext>,
cancel_token: CancelToken,
id_gen: IdGenerator,
space_info: Option<SpaceInfo>,
charset_info: Option<Box<CharsetInfo>>,
snapshot_ts: Option<Timestamp>,
isolation_level: Option<linkrs_core::types::TransactionIsolationLevel>,
arena_enabled: bool,
password_history_depth: usize,
}
pub(super) struct ContextParams {
pub cancel_token: CancelToken,
pub id_gen: IdGenerator,
pub space_info: Option<SpaceInfo>,
pub charset_info: Option<Box<CharsetInfo>>,
pub snapshot_ts: Option<Timestamp>,
pub isolation_level: Option<linkrs_core::types::TransactionIsolationLevel>,
pub arena_enabled: bool,
pub password_history_depth: usize,
}
impl QueryContext {
pub fn new(rctx: Arc<QueryRequestContext>) -> Self {
Self {
rctx,
cancel_token: CancelToken::new(),
id_gen: IdGenerator::new(0),
space_info: None,
charset_info: None,
snapshot_ts: None,
isolation_level: None,
arena_enabled: false,
password_history_depth: 0,
}
}
pub fn builder(rctx: Arc<QueryRequestContext>) -> super::QueryContextBuilder {
super::QueryContextBuilder::new(rctx)
}
pub(super) fn from_builder(rctx: Arc<QueryRequestContext>, params: ContextParams) -> Self {
Self {
rctx,
cancel_token: params.cancel_token,
id_gen: params.id_gen,
space_info: params.space_info,
charset_info: params.charset_info,
snapshot_ts: params.snapshot_ts,
isolation_level: params.isolation_level,
arena_enabled: params.arena_enabled,
password_history_depth: params.password_history_depth,
}
}
pub fn request_context(&self) -> &QueryRequestContext {
&self.rctx
}
pub fn request_context_arc(&self) -> Arc<QueryRequestContext> {
self.rctx.clone()
}
pub fn rctx(&self) -> &QueryRequestContext {
&self.rctx
}
pub fn cancel_token(&self) -> CancelToken {
self.cancel_token.clone()
}
pub fn charset_info(&self) -> Option<&CharsetInfo> {
self.charset_info.as_ref().map(|ci| ci.as_ref())
}
pub fn set_charset_info(&mut self, charset_info: CharsetInfo) {
self.charset_info = Some(Box::new(charset_info));
}
pub fn snapshot_ts(&self) -> Option<Timestamp> {
self.snapshot_ts
}
pub fn isolation_level(&self) -> Option<linkrs_core::types::TransactionIsolationLevel> {
self.isolation_level
}
pub fn password_history_depth(&self) -> usize {
self.password_history_depth
}
pub fn authenticated_user(&self) -> Option<&str> {
self.rctx.user_name.as_deref()
}
pub fn gen_id(&self) -> i64 {
self.id_gen.id()
}
pub fn current_id(&self) -> i64 {
self.id_gen.current_value()
}
pub fn space_info(&self) -> Option<&SpaceInfo> {
self.space_info.as_ref()
}
pub fn set_space_info(&mut self, space_info: SpaceInfo) {
self.space_info = Some(space_info);
}
pub fn space_id(&self) -> Option<u64> {
self.space_info.as_ref().map(|s| s.space_id)
}
pub fn space_name(&self) -> Option<String> {
self.space_info.as_ref().map(|s| s.space_name.clone())
}
pub fn mark_killed(&self) {
self.cancel_token.cancel(CancelReason::UserKill);
}
pub fn is_killed(&self) -> bool {
self.cancel_token.is_cancelled()
}
pub fn exist_parameter(&self, param: &str) -> bool {
self.rctx.get_parameter(param).is_some()
}
pub fn query(&self) -> &str {
&self.rctx.query
}
pub fn parameters(&self) -> &std::collections::HashMap<String, linkrs_core::Value> {
&self.rctx.parameters
}
pub fn reset(&mut self) {
self.cancel_token.clear();
self.id_gen.reset(0);
self.space_info = None;
self.charset_info = None;
self.snapshot_ts = None;
self.isolation_level = None;
log::info!("Query context has been reset");
}
pub fn has_arena(&self) -> bool {
self.arena_enabled
}
}
impl std::fmt::Debug for QueryContext {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("QueryContext")
.field("rctx", &self.rctx)
.field("space_id", &self.space_id())
.field("snapshot_ts", &self.snapshot_ts)
.field("isolation_level", &self.isolation_level)
.field("killed", &self.is_killed())
.field("has_arena", &self.arena_enabled)
.finish()
}
}
impl Default for QueryContext {
fn default() -> Self {
Self::new(Arc::new(QueryRequestContext::default()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn query_context_is_send_and_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<QueryContext>();
assert_send_sync::<Arc<QueryContext>>();
}
#[test]
fn mark_killed_cancels_shared_token() {
let rctx = Arc::new(QueryRequestContext::new("MATCH (n) RETURN n".to_string()));
let ctx = QueryContext::new(rctx);
assert!(!ctx.is_killed());
let runtime_token = ctx.cancel_token();
assert!(!runtime_token.is_cancelled());
ctx.mark_killed();
assert!(ctx.is_killed());
assert!(runtime_token.is_cancelled());
assert_eq!(runtime_token.reason(), Some(CancelReason::UserKill));
}
#[test]
fn runtime_adopts_context_token() {
use crate::executor::streaming::runtime::ExecutionRuntime;
let rctx = Arc::new(QueryRequestContext::new("MATCH (n) RETURN n".to_string()));
let ctx = QueryContext::new(rctx);
let runtime = ExecutionRuntime::default_budget();
runtime.set_cancel_token(ctx.cancel_token());
assert!(!runtime.is_cancelled());
ctx.mark_killed();
assert!(runtime.is_cancelled());
assert!(runtime.ensure_not_cancelled().is_err());
}
#[test]
fn reset_clears_cancellation() {
let rctx = Arc::new(QueryRequestContext::new("MATCH (n) RETURN n".to_string()));
let mut ctx = QueryContext::new(rctx);
ctx.mark_killed();
assert!(ctx.is_killed());
ctx.reset();
assert!(!ctx.is_killed());
}
}