blob: e6591135e0f9bd65b878d1a48a589bda2e3782b0 [file]
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
//! Query execution context for error reporting
//!
//! This module provides QueryContext which mirrors Spark's SQLQueryContext
//! for providing SQL text, line/position information, and error location
//! pointers in exception messages.
use serde::{Deserialize, Serialize};
use std::sync::Arc;
/// Based on Spark's SQLQueryContext for error reporting.
///
/// Contains information about where an error occurred in a SQL query,
/// including the full SQL text, line/column positions, and object context.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct QueryContext {
/// Full SQL query text
#[serde(rename = "sqlText")]
pub sql_text: Arc<String>,
/// Start offset in SQL text (0-based, character index)
#[serde(rename = "startIndex")]
pub start_index: i32,
/// Stop offset in SQL text (0-based, character index, inclusive)
#[serde(rename = "stopIndex")]
pub stop_index: i32,
/// Object type (e.g., "VIEW", "Project", "Filter")
#[serde(rename = "objectType", skip_serializing_if = "Option::is_none")]
pub object_type: Option<String>,
/// Object name (e.g., view name, column name)
#[serde(rename = "objectName", skip_serializing_if = "Option::is_none")]
pub object_name: Option<String>,
/// Line number in SQL query (1-based)
pub line: i32,
/// Column position within the line (0-based)
#[serde(rename = "startPosition")]
pub start_position: i32,
}
impl QueryContext {
#[allow(clippy::too_many_arguments)]
pub fn new(
sql_text: String,
start_index: i32,
stop_index: i32,
object_type: Option<String>,
object_name: Option<String>,
line: i32,
start_position: i32,
) -> Self {
Self {
sql_text: Arc::new(sql_text),
start_index,
stop_index,
object_type,
object_name,
line,
start_position,
}
}
/// Convert a character index to a byte offset in the SQL text.
/// Returns None if the character index is out of range.
fn char_index_to_byte_offset(&self, char_index: usize) -> Option<usize> {
self.sql_text
.char_indices()
.nth(char_index)
.map(|(byte_offset, _)| byte_offset)
}
/// Generate a summary string showing SQL fragment with error location.
/// (From SQLQueryContext.summary)
///
/// Format example:
/// ```text
/// == SQL of VIEW v1 (line 1, position 8) ==
/// SELECT a/b FROM t
/// ^^^
/// ```
pub fn format_summary(&self) -> String {
let start_char = self.start_index.max(0) as usize;
// stop_index is inclusive; fragment covers [start, stop]
let stop_char = (self.stop_index + 1).max(0) as usize;
let fragment = match (
self.char_index_to_byte_offset(start_char),
// stop_char may equal sql_text.chars().count() (one past the end)
self.char_index_to_byte_offset(stop_char).or_else(|| {
if stop_char == self.sql_text.chars().count() {
Some(self.sql_text.len())
} else {
None
}
}),
) {
(Some(start_byte), Some(stop_byte)) => &self.sql_text[start_byte..stop_byte],
_ => "",
};
// Build the header line
let mut summary = String::from("== SQL");
if let Some(obj_type) = &self.object_type {
if !obj_type.is_empty() {
summary.push_str(" of ");
summary.push_str(obj_type);
if let Some(obj_name) = &self.object_name {
if !obj_name.is_empty() {
summary.push(' ');
summary.push_str(obj_name);
}
}
}
}
summary.push_str(&format!(
" (line {}, position {}) ==\n",
self.line,
self.start_position + 1 // Convert 0-based to 1-based for display
));
// Add the SQL text with fragment highlighted
summary.push_str(&self.sql_text);
summary.push('\n');
// Add caret pointer
let caret_position = self.start_position.max(0) as usize;
summary.push_str(&" ".repeat(caret_position));
// fragment.chars().count() gives the correct display width for non-ASCII
summary.push_str(&"^".repeat(fragment.chars().count().max(1)));
summary
}
/// Returns the SQL fragment that caused the error.
pub fn fragment(&self) -> String {
let start_char = self.start_index.max(0) as usize;
let stop_char = (self.stop_index + 1).max(0) as usize;
match (
self.char_index_to_byte_offset(start_char),
self.char_index_to_byte_offset(stop_char).or_else(|| {
if stop_char == self.sql_text.chars().count() {
Some(self.sql_text.len())
} else {
None
}
}),
) {
(Some(start_byte), Some(stop_byte)) => self.sql_text[start_byte..stop_byte].to_string(),
_ => String::new(),
}
}
}
use std::collections::HashMap;
use std::sync::RwLock;
/// Map that stores QueryContext information for expressions during execution.
///
/// This map is populated during plan deserialization and accessed
/// during error creation to attach SQL context to exceptions.
#[derive(Debug)]
pub struct QueryContextMap {
/// Map from expression ID to QueryContext
contexts: RwLock<HashMap<u64, Arc<QueryContext>>>,
}
impl QueryContextMap {
pub fn new() -> Self {
Self {
contexts: RwLock::new(HashMap::new()),
}
}
/// Register a QueryContext for an expression ID.
///
/// If the expression ID already exists, it will be replaced.
///
/// # Arguments
/// * `expr_id` - Unique expression identifier from protobuf
/// * `context` - QueryContext containing SQL text and position info
pub fn register(&self, expr_id: u64, context: QueryContext) {
let mut contexts = self.contexts.write().unwrap();
contexts.insert(expr_id, Arc::new(context));
}
/// Get the QueryContext for an expression ID.
///
/// Returns None if no context is registered for this expression.
///
/// # Arguments
/// * `expr_id` - Expression identifier to look up
pub fn get(&self, expr_id: u64) -> Option<Arc<QueryContext>> {
let contexts = self.contexts.read().unwrap();
contexts.get(&expr_id).cloned()
}
/// Clear all registered contexts.
///
/// This is typically called after plan execution completes to free memory.
pub fn clear(&self) {
let mut contexts = self.contexts.write().unwrap();
contexts.clear();
}
/// Return the number of registered contexts (for debugging/testing)
pub fn len(&self) -> usize {
let contexts = self.contexts.read().unwrap();
contexts.len()
}
/// Check if the map is empty
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl Default for QueryContextMap {
fn default() -> Self {
Self::new()
}
}
/// Create a new session-scoped QueryContextMap.
///
/// This should be called once per SessionContext during plan creation
/// and passed to expressions that need query context for error reporting.
pub fn create_query_context_map() -> Arc<QueryContextMap> {
Arc::new(QueryContextMap::new())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_query_context_creation() {
let ctx = QueryContext::new(
"SELECT a/b FROM t".to_string(),
7,
9,
Some("Divide".to_string()),
Some("a/b".to_string()),
1,
7,
);
assert_eq!(*ctx.sql_text, "SELECT a/b FROM t");
assert_eq!(ctx.start_index, 7);
assert_eq!(ctx.stop_index, 9);
assert_eq!(ctx.object_type, Some("Divide".to_string()));
assert_eq!(ctx.object_name, Some("a/b".to_string()));
assert_eq!(ctx.line, 1);
assert_eq!(ctx.start_position, 7);
}
#[test]
fn test_query_context_serialization() {
let ctx = QueryContext::new(
"SELECT a/b FROM t".to_string(),
7,
9,
Some("Divide".to_string()),
Some("a/b".to_string()),
1,
7,
);
let json = serde_json::to_string(&ctx).unwrap();
let deserialized: QueryContext = serde_json::from_str(&json).unwrap();
assert_eq!(ctx, deserialized);
}
#[test]
fn test_format_summary() {
let ctx = QueryContext::new(
"SELECT a/b FROM t".to_string(),
7,
9,
Some("VIEW".to_string()),
Some("v1".to_string()),
1,
7,
);
let summary = ctx.format_summary();
assert!(summary.contains("== SQL of VIEW v1 (line 1, position 8) =="));
assert!(summary.contains("SELECT a/b FROM t"));
assert!(summary.contains("^^^")); // Three carets for "a/b"
}
#[test]
fn test_format_summary_without_object() {
let ctx = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
let summary = ctx.format_summary();
assert!(summary.contains("== SQL (line 1, position 8) =="));
assert!(summary.contains("SELECT a/b FROM t"));
}
#[test]
fn test_fragment() {
let ctx = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
assert_eq!(ctx.fragment(), "a/b");
}
#[test]
fn test_arc_string_sharing() {
let ctx1 = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
let ctx2 = ctx1.clone();
// Arc should share the same allocation
assert!(Arc::ptr_eq(&ctx1.sql_text, &ctx2.sql_text));
}
#[test]
fn test_json_with_optional_fields() {
let ctx = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
let json = serde_json::to_string(&ctx).unwrap();
// Should not serialize objectType and objectName when None
assert!(!json.contains("objectType"));
assert!(!json.contains("objectName"));
}
#[test]
fn test_map_register_and_get() {
let map = QueryContextMap::new();
let ctx = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
map.register(1, ctx.clone());
let retrieved = map.get(1).unwrap();
assert_eq!(*retrieved.sql_text, "SELECT a/b FROM t");
assert_eq!(retrieved.start_index, 7);
}
#[test]
fn test_map_get_nonexistent() {
let map = QueryContextMap::new();
assert!(map.get(999).is_none());
}
#[test]
fn test_map_clear() {
let map = QueryContextMap::new();
let ctx = QueryContext::new("SELECT a/b FROM t".to_string(), 7, 9, None, None, 1, 7);
map.register(1, ctx);
assert_eq!(map.len(), 1);
map.clear();
assert_eq!(map.len(), 0);
assert!(map.is_empty());
}
// Verify that fragment() and format_summary() correctly handle SQL text that
// contains multi-byte characters
#[test]
fn test_fragment_non_ascii_accented() {
// "é" is a 2-byte UTF-8 sequence (U+00E9).
// SQL: "SELECT café FROM t"
// 0123456789...
// char indices: c=7, a=8, f=9, é=10, ' '=11 ... FROM = 12..
// start_index=7, stop_index=10 should yield "café"
let sql = "SELECT café FROM t".to_string();
let ctx = QueryContext::new(sql, 7, 10, None, None, 1, 7);
assert_eq!(ctx.fragment(), "café");
}
}