hawdb_executor/
external.rs

1// Copyright 2026 Nowledge
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Host-provided physical read operators.
16
17#[doc(hidden)]
18pub mod seed;
19
20use crate::VectorExecutionReport;
21
22mod text;
23mod vector_index;
24use hawdb_core::{HawDBError, Result, RuntimeTaskContext};
25use hawdb_plan_cypher::VectorPhysicalPlan;
26use std::collections::BTreeMap;
27use std::mem::size_of;
28use std::num::NonZeroUsize;
29pub use text::{TextSeedExecutionOutput, TextSeedExecutionRequest, TextSeedExecutionRow};
30pub use vector_index::{
31    VectorIndexExecutionOutput, VectorIndexExecutionRequest, VectorIndexExecutionRow,
32};
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct ExternalReadResultBudget {
36    pub max_rows: usize,
37    pub max_memory_bytes: NonZeroUsize,
38}
39
40#[derive(Debug, Clone, Copy)]
41pub struct ExternalReadResourceContract<'a> {
42    pub priority: u8,
43    pub max_parallelism: NonZeroUsize,
44    pub max_working_memory_bytes: NonZeroUsize,
45    pub result: ExternalReadResultBudget,
46    pub task_context: Option<&'a RuntimeTaskContext>,
47}
48
49impl ExternalReadResourceContract<'_> {
50    pub fn checkpoint(&self) -> Result<()> {
51        match self.task_context {
52            Some(task_context) => task_context.checkpoint().map_err(|reason| {
53                HawDBError::Execution(format!("external read task stopped: {reason}"))
54            }),
55            None => Ok(()),
56        }
57    }
58
59    pub fn reserved_memory_bytes(&self) -> usize {
60        self.max_working_memory_bytes
61            .get()
62            .saturating_add(self.result.max_memory_bytes.get())
63    }
64}
65
66pub struct VectorSeedExecutionRequest<'a> {
67    pub embedding: &'a [f32],
68    pub metadata_filters: &'a BTreeMap<String, String>,
69    pub vector_plan: &'a VectorPhysicalPlan,
70    pub resources: ExternalReadResourceContract<'a>,
71}
72
73pub struct VectorSeedExecutionRow {
74    pub id: String,
75    pub external_id: Option<String>,
76    pub score: f64,
77}
78
79pub struct VectorSeedExecutionOutput {
80    pub rows: Vec<VectorSeedExecutionRow>,
81    pub report: VectorExecutionReport,
82}
83
84impl VectorSeedExecutionOutput {
85    pub fn estimated_memory_bytes(&self) -> usize {
86        self.rows
87            .capacity()
88            .saturating_mul(size_of::<VectorSeedExecutionRow>())
89            .saturating_add(self.rows.iter().fold(0usize, |total, row| {
90                total.saturating_add(row.id.capacity()).saturating_add(
91                    row.external_id
92                        .as_ref()
93                        .map_or(0, |external_id| external_id.capacity()),
94                )
95            }))
96    }
97
98    pub fn validate_result_budget(&self, budget: ExternalReadResultBudget) -> Result<()> {
99        if self.rows.len() > budget.max_rows {
100            return Err(HawDBError::read_budget_exceeded(
101                hawdb_core::ReadBudgetResource::Rows,
102                budget.max_rows,
103                format!(
104                    "external vector read returned {} rows, exceeding result row budget {}",
105                    self.rows.len(),
106                    budget.max_rows
107                ),
108            ));
109        }
110        let memory_bytes = self.estimated_memory_bytes();
111        if memory_bytes > budget.max_memory_bytes.get() {
112            return Err(HawDBError::Execution(format!(
113                "external vector read returned {memory_bytes} estimated bytes, exceeding result memory budget {}",
114                budget.max_memory_bytes
115            )));
116        }
117        Ok(())
118    }
119}
120
121pub trait ExternalReadOperator {
122    fn execute_vector_index(
123        &mut self,
124        request: VectorIndexExecutionRequest<'_>,
125    ) -> Result<Box<dyn VectorIndexExecutionOutput>> {
126        request.resources.checkpoint()?;
127        Err(HawDBError::Execution(
128            "vector index search requires a bound canonical/index read domain".into(),
129        ))
130    }
131
132    fn execute_text_seed(
133        &mut self,
134        request: TextSeedExecutionRequest<'_>,
135    ) -> Result<TextSeedExecutionOutput> {
136        request.resources.checkpoint()?;
137        Err(HawDBError::Execution(
138            "text search capability is unavailable without a text projection".into(),
139        ))
140    }
141
142    fn execute_vector_seed(
143        &mut self,
144        request: VectorSeedExecutionRequest<'_>,
145    ) -> Result<VectorSeedExecutionOutput>;
146}
147
148#[doc(hidden)]
149pub struct NoExternalReadOperator;
150
151impl ExternalReadOperator for NoExternalReadOperator {
152    fn execute_vector_seed(
153        &mut self,
154        _request: VectorSeedExecutionRequest<'_>,
155    ) -> Result<VectorSeedExecutionOutput> {
156        Err(HawDBError::Execution(
157            "vector search capability is unavailable without a search projection".to_string(),
158        ))
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165    use crate::{
166        VectorCompressionMode, VectorExecutionBackend, VectorFallbackReasonCode, VectorScoreSource,
167    };
168    use hawdb_core::RuntimeCancellationToken;
169    use hawdb_plan_cypher::VectorCandidateSource;
170
171    pub(super) fn empty_report() -> VectorExecutionReport {
172        VectorExecutionReport {
173            backend: VectorExecutionBackend::ScalarFlat,
174            compression_mode: VectorCompressionMode::Disabled,
175            candidate_source: VectorCandidateSource::Scalar,
176            backend_selection_reason: None,
177            estimated_raw_vector_bytes: None,
178            filter_selectivity_per_million: None,
179            candidate_score_source: VectorScoreSource::RawVector,
180            final_score_source: VectorScoreSource::RawVector,
181            generated_candidate_count: 0,
182            descriptor_pruned_count: 0,
183            scalar_filtered_count: 0,
184            residual_filtered_count: 0,
185            candidate_scan_rounds: 0,
186            reranked_candidate_count: 0,
187            returned_count: 0,
188            raw_vector_bytes_read: 0,
189            candidate_scan_metrics: None,
190            index_covered_document_count: None,
191            index_candidate_document_count: None,
192            index_coverage_complete: None,
193            fallback_reason_codes: Vec::<VectorFallbackReasonCode>::new(),
194        }
195    }
196
197    #[test]
198    fn external_vector_output_fails_closed_on_row_and_memory_budgets() {
199        let output = VectorSeedExecutionOutput {
200            rows: vec![VectorSeedExecutionRow {
201                id: "memory-1".to_string(),
202                external_id: Some("external-1".to_string()),
203                score: 1.0,
204            }],
205            report: empty_report(),
206        };
207
208        let row_error = output
209            .validate_result_budget(ExternalReadResultBudget {
210                max_rows: 0,
211                max_memory_bytes: NonZeroUsize::new(1024).unwrap(),
212            })
213            .unwrap_err();
214        assert!(matches!(row_error, HawDBError::ReadBudgetExceeded(cause)
215            if cause.resource == hawdb_core::ReadBudgetResource::Rows && cause.limit == 0));
216        let memory_error = output
217            .validate_result_budget(ExternalReadResultBudget {
218                max_rows: 1,
219                max_memory_bytes: NonZeroUsize::MIN,
220            })
221            .unwrap_err();
222        assert!(matches!(memory_error, HawDBError::Execution(_)));
223    }
224
225    #[test]
226    fn external_read_contract_observes_cancellation() {
227        let cancellation = RuntimeCancellationToken::new();
228        let task_context = RuntimeTaskContext::without_deadline(cancellation.clone());
229        let contract = ExternalReadResourceContract {
230            priority: 128,
231            max_parallelism: NonZeroUsize::MIN,
232            max_working_memory_bytes: NonZeroUsize::MIN,
233            result: ExternalReadResultBudget {
234                max_rows: 1,
235                max_memory_bytes: NonZeroUsize::MIN,
236            },
237            task_context: Some(&task_context),
238        };
239
240        assert!(contract.checkpoint().is_ok());
241        assert!(cancellation.cancel());
242        assert!(contract.checkpoint().is_err());
243    }
244}
hawdb_executor/external.rs

hawdb_executor/
external.rs
