1use super::{
17 executor::{QueryExecutor, QueryResult},
18 planner::{PlanType, QueryPlan},
19 ParsedQuery,
20};
21use crate::types::DataType;
22use crate::{Error, Result, Value};
23use std::collections::HashMap;
24use std::sync::Arc;
25
26#[cfg(feature = "state_machine")]
35#[derive(Debug)]
36struct PreparedSelect {
37 statement: super::select_ast::SelectStatement,
40 optimizer: Arc<super::select_optimizer::SelectOptimizer>,
42 executor: Arc<super::select_executor::SelectExecutor>,
44 plan_cache: std::sync::Mutex<Option<CachedOptimizedPlan>>,
53}
54
55#[cfg(feature = "state_machine")]
58#[derive(Debug)]
59struct CachedOptimizedPlan {
60 params: Vec<Value>,
61 plan: Arc<super::select_optimizer::OptimizedQueryPlan>,
62}
63
64#[cfg(feature = "state_machine")]
65impl PreparedSelect {
66 async fn plan_for(
72 &self,
73 params: &[Value],
74 ) -> Result<Arc<super::select_optimizer::OptimizedQueryPlan>> {
75 {
76 let guard = self
77 .plan_cache
78 .lock()
79 .map_err(|_| Error::query_execution("prepared plan cache lock poisoned"))?;
80 if let Some(cached) = guard.as_ref() {
81 if params_bit_eq(&cached.params, params) {
82 return Ok(Arc::clone(&cached.plan));
83 }
84 }
85 }
86
87 let mut statement = self.statement.clone();
88 statement.bind_parameters(params)?;
89 let plan = Arc::new(self.optimizer.optimize(statement).await?);
90
91 if let Ok(mut guard) = self.plan_cache.lock() {
92 *guard = Some(CachedOptimizedPlan {
93 params: params.to_vec(),
94 plan: Arc::clone(&plan),
95 });
96 }
97 Ok(plan)
98 }
99}
100
101#[cfg(feature = "state_machine")]
112fn params_bit_eq(a: &[Value], b: &[Value]) -> bool {
113 a.len() == b.len() && a.iter().zip(b).all(|(x, y)| value_bit_eq(x, y))
114}
115
116#[cfg(feature = "state_machine")]
122fn value_bit_eq(a: &Value, b: &Value) -> bool {
123 match (a, b) {
124 (Value::Float(x), Value::Float(y)) => x.to_bits() == y.to_bits(),
125 (Value::Float32(x), Value::Float32(y)) => x.to_bits() == y.to_bits(),
126 (Value::List(x), Value::List(y))
127 | (Value::Set(x), Value::Set(y))
128 | (Value::Tuple(x), Value::Tuple(y)) => {
129 x.len() == y.len() && x.iter().zip(y).all(|(m, n)| value_bit_eq(m, n))
130 }
131 (Value::Map(x), Value::Map(y)) => {
132 x.len() == y.len()
133 && x.iter()
134 .zip(y)
135 .all(|((xk, xv), (yk, yv))| value_bit_eq(xk, yk) && value_bit_eq(xv, yv))
136 }
137 (Value::Frozen(x), Value::Frozen(y)) => value_bit_eq(x, y),
138 (Value::Udt(x), Value::Udt(y)) => {
139 x.type_name == y.type_name
140 && x.keyspace == y.keyspace
141 && x.fields.len() == y.fields.len()
142 && x.fields.iter().zip(&y.fields).all(|(xf, yf)| {
143 xf.name == yf.name
144 && match (&xf.value, &yf.value) {
145 (Some(xv), Some(yv)) => value_bit_eq(xv, yv),
146 (None, None) => true,
147 _ => false,
148 }
149 })
150 }
151 _ => a == b,
154 }
155}
156
157#[derive(Debug)]
159pub struct PreparedQuery {
160 pub cql: String,
162 pub parsed_query: ParsedQuery,
164 pub plan: QueryPlan,
166 pub parameters: Vec<ParameterMetadata>,
168 executor: Arc<QueryExecutor>,
170 #[cfg(feature = "state_machine")]
174 select_pipeline: Option<PreparedSelect>,
175}
176
177#[derive(Debug, Clone)]
179pub struct ParameterMetadata {
180 pub name: Option<String>,
182 pub position: usize,
184 pub expected_type: Option<DataType>,
186 pub optional: bool,
188}
189
190#[derive(Debug)]
192pub struct PreparedContext {
193 pub parameters: HashMap<String, Value>,
195 pub positional_params: Vec<Value>,
197 pub hints: ExecutionHints,
199}
200
201#[derive(Debug, Clone, Default)]
203pub struct ExecutionHints {
204 pub force_index: Option<String>,
206 pub timeout_ms: Option<u64>,
208 pub parallelism: Option<usize>,
210 pub cache_results: bool,
212}
213
214impl PreparedQuery {
215 pub fn new(parsed_query: ParsedQuery, plan: QueryPlan, executor: Arc<QueryExecutor>) -> Self {
217 let cql = parsed_query.cql.clone();
218 let parameters = Self::extract_parameters(&parsed_query);
219
220 Self {
221 cql,
222 parsed_query,
223 plan,
224 parameters,
225 executor,
226 #[cfg(feature = "state_machine")]
227 select_pipeline: None,
228 }
229 }
230
231 #[cfg(feature = "state_machine")]
240 #[allow(clippy::too_many_arguments)]
241 pub(crate) fn new_select(
242 parsed_query: ParsedQuery,
243 plan: QueryPlan,
244 executor: Arc<QueryExecutor>,
245 statement: super::select_ast::SelectStatement,
246 select_marker_count: usize,
247 optimizer: Arc<super::select_optimizer::SelectOptimizer>,
248 select_executor: Arc<super::select_executor::SelectExecutor>,
249 ) -> Self {
250 let cql = parsed_query.cql.clone();
251 let parameters = (0..select_marker_count)
255 .map(|position| ParameterMetadata {
256 name: None,
257 position,
258 expected_type: None,
259 optional: false,
260 })
261 .collect();
262
263 Self {
264 cql,
265 parsed_query,
266 plan,
267 parameters,
268 executor,
269 select_pipeline: Some(PreparedSelect {
270 statement,
271 optimizer,
272 executor: select_executor,
273 plan_cache: std::sync::Mutex::new(None),
274 }),
275 }
276 }
277
278 pub async fn execute(&self, params: &[Value]) -> Result<QueryResult> {
285 self.validate_params(params)?;
286
287 #[cfg(feature = "state_machine")]
288 if let Some(pipeline) = &self.select_pipeline {
289 let plan = pipeline.plan_for(params).await?;
293 return pipeline.executor.execute((*plan).clone()).await;
294 }
295
296 self.executor.execute(&self.plan).await
299 }
300
301 pub async fn execute_named(&self, params: &HashMap<String, Value>) -> Result<QueryResult> {
303 let mut positional_params = Vec::with_capacity(self.parameters.len());
305 for metadata in &self.parameters {
306 let Some(name) = &metadata.name else {
307 continue;
308 };
309 match params.get(name) {
310 Some(value) => positional_params.push(value.clone()),
311 None if metadata.optional => positional_params.push(Value::Null),
312 None => {
313 return Err(Error::query_execution(format!(
314 "Missing required parameter: {}",
315 name
316 )));
317 }
318 }
319 }
320
321 self.execute(&positional_params).await
322 }
323
324 pub async fn execute_with_context(&self, context: &PreparedContext) -> Result<QueryResult> {
342 let hints = &context.hints;
343
344 #[cfg(feature = "state_machine")]
345 if let Some(pipeline) = &self.select_pipeline {
346 if hints.force_index.is_some()
347 || hints.timeout_ms.is_some()
348 || hints.parallelism.is_some()
349 {
350 return Err(Error::query_execution(
351 "Execution hints (force_index/timeout_ms/parallelism) are not supported for \
352 prepared SELECT statements; they apply only to the legacy plan path. Re-run \
353 without hints to bind parameters and use the partition-targeted SELECT path.",
354 ));
355 }
356 self.validate_params(&context.positional_params)?;
360 let plan = pipeline.plan_for(&context.positional_params).await?;
361 return pipeline.executor.execute((*plan).clone()).await;
362 }
363
364 if hints.force_index.is_none() && hints.timeout_ms.is_none() && hints.parallelism.is_none()
366 {
367 return self.executor.execute(&self.plan).await;
368 }
369
370 let mut modified_plan = self.plan.clone();
371 if let Some(force_index) = &hints.force_index {
372 modified_plan.hints.force_index = Some(force_index.clone());
373 }
374 if let Some(timeout) = hints.timeout_ms {
375 modified_plan.hints.timeout_ms = Some(timeout);
376 }
377 if let Some(parallelism) = hints.parallelism {
378 modified_plan.hints.preferred_parallelization = Some(parallelism);
379 }
380 self.executor.execute(&modified_plan).await
381 }
382
383 pub fn parameters(&self) -> &[ParameterMetadata] {
385 &self.parameters
386 }
387
388 pub fn cql(&self) -> &str {
390 &self.cql
391 }
392
393 pub fn plan(&self) -> &QueryPlan {
395 &self.plan
396 }
397
398 pub fn stats(&self) -> PreparedQueryStats {
400 PreparedQueryStats {
401 parameter_count: self.parameters.len(),
402 plan_type: format!("{:?}", self.plan.plan_type),
403 estimated_cost: self.plan.estimated_cost,
404 estimated_rows: self.plan.estimated_rows,
405 cache_friendly: self.is_cache_friendly(),
406 }
407 }
408
409 pub fn is_cache_friendly(&self) -> bool {
411 matches!(
415 self.plan.plan_type,
416 PlanType::PointLookup | PlanType::IndexScan | PlanType::TableScan
417 )
418 }
419
420 fn validate_params(&self, params: &[Value]) -> Result<()> {
422 if params.len() != self.parameters.len() {
423 return Err(Error::query_execution(format!(
424 "Parameter count mismatch: expected {}, got {}",
425 self.parameters.len(),
426 params.len()
427 )));
428 }
429
430 for (i, (param, metadata)) in params.iter().zip(&self.parameters).enumerate() {
431 if let Some(expected_type) = &metadata.expected_type {
432 if !type_matches(param, expected_type) {
433 return Err(Error::query_execution(format!(
434 "Parameter {} type mismatch: expected {:?}, got {:?}",
435 i, expected_type, param
436 )));
437 }
438 }
439 }
440
441 Ok(())
442 }
443
444 fn extract_parameters(parsed_query: &ParsedQuery) -> Vec<ParameterMetadata> {
450 if parsed_query.where_clause.is_none() {
451 return Vec::new();
452 }
453 vec![ParameterMetadata {
454 name: None,
455 position: 0,
456 expected_type: Some(DataType::Integer),
457 optional: false,
458 }]
459 }
460}
461
462fn type_matches(value: &Value, expected_type: &DataType) -> bool {
464 matches!(
465 (value, expected_type),
466 (Value::Integer(_), DataType::Integer)
467 | (Value::Float(_), DataType::Float)
468 | (Value::Text(_), DataType::Text)
469 | (Value::Boolean(_), DataType::Boolean)
470 | (Value::Null, _)
471 )
472}
473
474#[derive(Debug, Clone)]
476pub struct PreparedQueryStats {
477 pub parameter_count: usize,
479 pub plan_type: String,
481 pub estimated_cost: f64,
483 pub estimated_rows: u64,
485 pub cache_friendly: bool,
487}
488
489#[derive(Default)]
491pub struct PreparedQueryBuilder {
492 cql: String,
494 parameters: Vec<ParameterMetadata>,
496 hints: ExecutionHints,
498}
499
500impl PreparedQueryBuilder {
501 pub fn new(cql: &str) -> Self {
503 Self {
504 cql: cql.to_string(),
505 ..Self::default()
506 }
507 }
508
509 pub fn parameter(mut self, name: Option<String>, data_type: DataType, optional: bool) -> Self {
511 self.push_parameter(name, data_type, optional);
512 self
513 }
514
515 pub fn positional_parameter(mut self, data_type: DataType) -> Self {
517 self.push_parameter(None, data_type, false);
518 self
519 }
520
521 pub fn named_parameter(mut self, name: &str, data_type: DataType, optional: bool) -> Self {
523 self.push_parameter(Some(name.to_string()), data_type, optional);
524 self
525 }
526
527 pub fn hints(mut self, hints: ExecutionHints) -> Self {
529 self.hints = hints;
530 self
531 }
532
533 pub fn force_index(mut self, index_name: &str) -> Self {
535 self.hints.force_index = Some(index_name.to_string());
536 self
537 }
538
539 pub fn timeout(mut self, timeout_ms: u64) -> Self {
541 self.hints.timeout_ms = Some(timeout_ms);
542 self
543 }
544
545 pub fn parallelism(mut self, threads: usize) -> Self {
547 self.hints.parallelism = Some(threads);
548 self
549 }
550
551 pub fn cache_results(mut self) -> Self {
553 self.hints.cache_results = true;
554 self
555 }
556
557 pub fn build(
559 self,
560 parsed_query: ParsedQuery,
561 plan: QueryPlan,
562 executor: Arc<QueryExecutor>,
563 ) -> PreparedQuery {
564 PreparedQuery {
565 cql: self.cql,
566 parsed_query,
567 plan,
568 parameters: self.parameters,
569 executor,
570 #[cfg(feature = "state_machine")]
571 select_pipeline: None,
572 }
573 }
574
575 fn push_parameter(&mut self, name: Option<String>, data_type: DataType, optional: bool) {
576 self.parameters.push(ParameterMetadata {
577 name,
578 position: self.parameters.len(),
579 expected_type: Some(data_type),
580 optional,
581 });
582 }
583}
584
585#[cfg(all(test, feature = "state_machine"))]
586mod tests {
587 use super::*;
588 use crate::Config;
589 use std::sync::Arc;
590 use tempfile::TempDir;
591
592 #[tokio::test]
593 #[cfg(feature = "state_machine")]
594 async fn test_prepared_query_creation() {
595 let temp_dir = TempDir::new().unwrap();
596 let config = Config::default();
597 let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
598 let storage = Arc::new(
599 crate::storage::StorageEngine::open(
600 temp_dir.path(),
601 &config,
602 platform,
603 #[cfg(feature = "state_machine")]
604 None,
605 )
606 .await
607 .unwrap(),
608 );
609 let schema = Arc::new(
610 crate::schema::SchemaManager::new(temp_dir.path())
611 .await
612 .unwrap(),
613 );
614 let executor = Arc::new(crate::query::executor::QueryExecutor::new(
615 storage, schema, &config,
616 ));
617
618 let parsed_query = ParsedQuery {
619 query_type: crate::query::QueryType::Select,
620 table: Some(crate::TableId::new("users")),
621 columns: vec!["*".to_string()],
622 where_clause: None,
623 values: vec![],
624 set_clause: std::collections::HashMap::new(),
625 order_by: vec![],
626 limit: None,
627 cql: "SELECT * FROM users".to_string(),
628 };
629
630 let plan = crate::query::planner::QueryPlan {
631 plan_type: crate::query::planner::PlanType::TableScan,
632 table: Some(crate::TableId::new("users")),
633 estimated_cost: 100.0,
634 estimated_rows: 1000,
635 selected_indexes: vec![],
636 steps: vec![],
637 hints: crate::query::planner::QueryHints::default(),
638 };
639
640 let prepared = PreparedQuery::new(parsed_query, plan, executor);
641
642 assert_eq!(prepared.cql(), "SELECT * FROM users");
643 assert_eq!(prepared.parameters().len(), 0);
644 assert!(prepared.is_cache_friendly());
646 }
647
648 #[test]
649 fn test_prepared_query_builder() {
650 let builder = PreparedQueryBuilder::new("SELECT * FROM users WHERE id = ? AND name = ?")
651 .positional_parameter(DataType::Integer)
652 .positional_parameter(DataType::Text)
653 .timeout(5000)
654 .parallelism(4);
655
656 assert_eq!(builder.cql, "SELECT * FROM users WHERE id = ? AND name = ?");
657 assert_eq!(builder.parameters.len(), 2);
658 assert_eq!(builder.hints.timeout_ms, Some(5000));
659 assert_eq!(builder.hints.parallelism, Some(4));
660 }
661
662 #[test]
663 fn test_parameter_metadata() {
664 let metadata = ParameterMetadata {
665 name: Some("user_id".to_string()),
666 position: 0,
667 expected_type: Some(DataType::Integer),
668 optional: false,
669 };
670
671 assert_eq!(metadata.name, Some("user_id".to_string()));
672 assert_eq!(metadata.position, 0);
673 assert!(!metadata.optional);
674 }
675
676 #[test]
677 fn test_execution_hints() {
678 let hints = ExecutionHints {
679 force_index: Some("idx_user_id".to_string()),
680 timeout_ms: Some(10000),
681 parallelism: Some(8),
682 cache_results: true,
683 };
684
685 assert_eq!(hints.force_index, Some("idx_user_id".to_string()));
686 assert_eq!(hints.timeout_ms, Some(10000));
687 assert_eq!(hints.parallelism, Some(8));
688 assert!(hints.cache_results);
689 }
690
691 #[test]
692 fn test_type_matching() {
693 assert!(type_matches(&Value::Integer(42), &DataType::Integer));
694 assert!(type_matches(
695 &Value::text("test".to_string()),
696 &DataType::Text
697 ));
698 assert!(type_matches(&Value::Null, &DataType::Integer));
700 assert!(!type_matches(&Value::Integer(42), &DataType::Text));
701 }
702
703 #[tokio::test]
710 async fn prepared_select_reuses_optimized_plan_across_executes() {
711 use crate::query::select_optimizer::OPTIMIZE_INVOCATIONS;
712
713 let temp_dir = TempDir::new().unwrap();
714 let config = Config::default();
715 let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
716 let storage = Arc::new(
717 crate::storage::StorageEngine::open(temp_dir.path(), &config, platform, None)
718 .await
719 .unwrap(),
720 );
721 let schema = Arc::new(
722 crate::schema::SchemaManager::new(temp_dir.path())
723 .await
724 .unwrap(),
725 );
726 let executor = Arc::new(crate::query::executor::QueryExecutor::new(
727 storage.clone(),
728 schema.clone(),
729 &config,
730 ));
731 let optimizer = Arc::new(super::super::select_optimizer::SelectOptimizer::new(
732 schema.clone(),
733 storage.clone(),
734 ));
735 let select_executor = Arc::new(super::super::select_executor::SelectExecutor::new(
736 schema.clone(),
737 storage.clone(),
738 ));
739
740 let cql = "SELECT * FROM t WHERE id = ?";
741 let statement = super::super::select_parser::parse_select(cql).unwrap();
742 let marker_count = statement.bind_marker_count();
743 assert_eq!(marker_count, 1);
744
745 let parsed_query = ParsedQuery {
746 query_type: crate::query::QueryType::Select,
747 table: Some(crate::TableId::new("t")),
748 columns: vec!["*".to_string()],
749 where_clause: None,
750 values: vec![],
751 set_clause: std::collections::HashMap::new(),
752 order_by: vec![],
753 limit: None,
754 cql: cql.to_string(),
755 };
756 let plan = crate::query::planner::QueryPlan {
757 plan_type: crate::query::planner::PlanType::PointLookup,
758 table: Some(crate::TableId::new("t")),
759 estimated_cost: 1.0,
760 estimated_rows: 1,
761 selected_indexes: vec![],
762 steps: vec![],
763 hints: crate::query::planner::QueryHints::default(),
764 };
765
766 let prepared = PreparedQuery::new_select(
767 parsed_query,
768 plan,
769 executor,
770 statement,
771 marker_count,
772 optimizer,
773 select_executor,
774 );
775
776 OPTIMIZE_INVOCATIONS.with(|c| c.set(0));
778 let params = vec![Value::Integer(1)];
779 for _ in 0..100 {
780 let _ = prepared.execute(¶ms).await;
783 }
784 let after_same = OPTIMIZE_INVOCATIONS.with(|c| c.get());
785 assert!(
786 after_same <= 1,
787 "issue #1587: prepared statement must optimize at most once across 100 \
788 executes with identical params, got {after_same}"
789 );
790
791 let _ = prepared.execute(&[Value::Integer(2)]).await;
793 let after_diff = OPTIMIZE_INVOCATIONS.with(|c| c.get());
794 assert_eq!(
795 after_diff,
796 after_same + 1,
797 "a new parameter tuple must re-optimize (correctness over reuse)"
798 );
799
800 let _ = prepared.execute(&[Value::Integer(2)]).await;
804 let after_repeat = OPTIMIZE_INVOCATIONS.with(|c| c.get());
805 assert_eq!(
806 after_repeat, after_diff,
807 "repeating the last params must hit the memoized plan (no re-optimize)"
808 );
809 }
810
811 #[tokio::test]
819 async fn prepared_select_signed_zero_float_params_do_not_reuse_plan() {
820 use crate::query::select_optimizer::OPTIMIZE_INVOCATIONS;
821
822 let temp_dir = TempDir::new().unwrap();
823 let config = Config::default();
824 let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
825 let storage = Arc::new(
826 crate::storage::StorageEngine::open(temp_dir.path(), &config, platform, None)
827 .await
828 .unwrap(),
829 );
830 let schema = Arc::new(
831 crate::schema::SchemaManager::new(temp_dir.path())
832 .await
833 .unwrap(),
834 );
835 let executor = Arc::new(crate::query::executor::QueryExecutor::new(
836 storage.clone(),
837 schema.clone(),
838 &config,
839 ));
840 let optimizer = Arc::new(super::super::select_optimizer::SelectOptimizer::new(
841 schema.clone(),
842 storage.clone(),
843 ));
844 let select_executor = Arc::new(super::super::select_executor::SelectExecutor::new(
845 schema.clone(),
846 storage.clone(),
847 ));
848
849 let cql = "SELECT * FROM t WHERE id = ?";
850 let statement = super::super::select_parser::parse_select(cql).unwrap();
851 let marker_count = statement.bind_marker_count();
852 assert_eq!(marker_count, 1);
853
854 let parsed_query = ParsedQuery {
855 query_type: crate::query::QueryType::Select,
856 table: Some(crate::TableId::new("t")),
857 columns: vec!["*".to_string()],
858 where_clause: None,
859 values: vec![],
860 set_clause: std::collections::HashMap::new(),
861 order_by: vec![],
862 limit: None,
863 cql: cql.to_string(),
864 };
865 let plan = crate::query::planner::QueryPlan {
866 plan_type: crate::query::planner::PlanType::PointLookup,
867 table: Some(crate::TableId::new("t")),
868 estimated_cost: 1.0,
869 estimated_rows: 1,
870 selected_indexes: vec![],
871 steps: vec![],
872 hints: crate::query::planner::QueryHints::default(),
873 };
874
875 let prepared = PreparedQuery::new_select(
876 parsed_query,
877 plan,
878 executor,
879 statement,
880 marker_count,
881 optimizer,
882 select_executor,
883 );
884
885 OPTIMIZE_INVOCATIONS.with(|c| c.set(0));
886
887 let _ = prepared.execute(&[Value::Float(0.0)]).await;
889 let after_pos = OPTIMIZE_INVOCATIONS.with(|c| c.get());
890 assert_eq!(after_pos, 1, "first float param must optimize once");
891
892 let _ = prepared.execute(&[Value::Float(-0.0)]).await;
896 let after_neg = OPTIMIZE_INVOCATIONS.with(|c| c.get());
897 assert_eq!(
898 after_neg,
899 after_pos + 1,
900 "issue #1587: +0.0 and -0.0 float params must be distinct plan-cache \
901 keys and re-optimize, never reuse the wrong-signed-zero plan"
902 );
903
904 let _ = prepared.execute(&[Value::Float(0.0)]).await;
907 let after_flip_back = OPTIMIZE_INVOCATIONS.with(|c| c.get());
908 assert_eq!(
909 after_flip_back,
910 after_neg + 1,
911 "flipping -0.0 back to +0.0 must re-optimize (distinct bit-exact keys)"
912 );
913
914 let _ = prepared.execute(&[Value::Float(0.0)]).await;
917 let after_repeat = OPTIMIZE_INVOCATIONS.with(|c| c.get());
918 assert_eq!(
919 after_repeat, after_flip_back,
920 "identical float params must still hit the memoized plan (no re-optimize)"
921 );
922 }
923}