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    /// Caller admits MAX_VECTOR_INDEX_DESCRIPTOR_BYTES before requesting a copy.
123    fn vector_index_definition(
124        &mut self,
125        _index: &str,
126    ) -> Result<(
127        hawdb_core::VectorIndexDescriptor,
128        hawdb_core::VectorIndexReadIdentity,
129    )> {
130        Err(HawDBError::Execution(
131            "vector eligibility requires a bound index definition".into(),
132        ))
133    }
134
135    fn execute_vector_index(
136        &mut self,
137        request: VectorIndexExecutionRequest<'_>,
138    ) -> Result<Box<dyn VectorIndexExecutionOutput>> {
139        request.resources.checkpoint()?;
140        Err(HawDBError::Execution(
141            "vector index search requires a bound canonical/index read domain".into(),
142        ))
143    }
144
145    fn execute_complete_text_seed(
146        &mut self,
147        request: TextSeedExecutionRequest<'_>,
148    ) -> Result<TextSeedExecutionOutput> {
149        request.resources.checkpoint()?;
150        Err(HawDBError::Execution(
151            "complete text search requires a provider with source-exhaustion support".into(),
152        ))
153    }
154
155    fn execute_text_seed(
156        &mut self,
157        request: TextSeedExecutionRequest<'_>,
158    ) -> Result<TextSeedExecutionOutput> {
159        request.resources.checkpoint()?;
160        Err(HawDBError::Execution(
161            "text search capability is unavailable without a text projection".into(),
162        ))
163    }
164
165    fn execute_vector_seed(
166        &mut self,
167        request: VectorSeedExecutionRequest<'_>,
168    ) -> Result<VectorSeedExecutionOutput>;
169}
170
171#[doc(hidden)]
172pub struct NoExternalReadOperator;
173
174impl ExternalReadOperator for NoExternalReadOperator {
175    fn execute_vector_seed(
176        &mut self,
177        _request: VectorSeedExecutionRequest<'_>,
178    ) -> Result<VectorSeedExecutionOutput> {
179        Err(HawDBError::Execution(
180            "vector search capability is unavailable without a search projection".to_string(),
181        ))
182    }
183}
184
185#[cfg(test)]
186mod tests {
187    use super::*;
188    use crate::{
189        VectorCompressionMode, VectorExecutionBackend, VectorFallbackReasonCode, VectorScoreSource,
190    };
191    use hawdb_core::RuntimeCancellationToken;
192    use hawdb_plan_cypher::VectorCandidateSource;
193
194    pub(super) fn empty_report() -> VectorExecutionReport {
195        VectorExecutionReport {
196            backend: VectorExecutionBackend::ScalarFlat,
197            compression_mode: VectorCompressionMode::Disabled,
198            candidate_source: VectorCandidateSource::Scalar,
199            backend_selection_reason: None,
200            estimated_raw_vector_bytes: None,
201            filter_selectivity_per_million: None,
202            candidate_score_source: VectorScoreSource::RawVector,
203            final_score_source: VectorScoreSource::RawVector,
204            generated_candidate_count: 0,
205            descriptor_pruned_count: 0,
206            scalar_filtered_count: 0,
207            residual_filtered_count: 0,
208            candidate_scan_rounds: 0,
209            reranked_candidate_count: 0,
210            returned_count: 0,
211            raw_vector_bytes_read: 0,
212            candidate_scan_metrics: None,
213            index_covered_document_count: None,
214            index_candidate_document_count: None,
215            index_coverage_complete: None,
216            fallback_reason_codes: Vec::<VectorFallbackReasonCode>::new(),
217        }
218    }
219
220    #[test]
221    fn external_vector_output_fails_closed_on_row_and_memory_budgets() {
222        let output = VectorSeedExecutionOutput {
223            rows: vec![VectorSeedExecutionRow {
224                id: "memory-1".to_string(),
225                external_id: Some("external-1".to_string()),
226                score: 1.0,
227            }],
228            report: empty_report(),
229        };
230
231        let row_error = output
232            .validate_result_budget(ExternalReadResultBudget {
233                max_rows: 0,
234                max_memory_bytes: NonZeroUsize::new(1024).unwrap(),
235            })
236            .unwrap_err();
237        assert!(matches!(row_error, HawDBError::ReadBudgetExceeded(cause)
238            if cause.resource == hawdb_core::ReadBudgetResource::Rows && cause.limit == 0));
239        let memory_error = output
240            .validate_result_budget(ExternalReadResultBudget {
241                max_rows: 1,
242                max_memory_bytes: NonZeroUsize::MIN,
243            })
244            .unwrap_err();
245        assert!(matches!(memory_error, HawDBError::Execution(_)));
246    }
247
248    #[test]
249    fn external_read_contract_observes_cancellation() {
250        let cancellation = RuntimeCancellationToken::new();
251        let task_context = RuntimeTaskContext::without_deadline(cancellation.clone());
252        let contract = ExternalReadResourceContract {
253            priority: 128,
254            max_parallelism: NonZeroUsize::MIN,
255            max_working_memory_bytes: NonZeroUsize::MIN,
256            result: ExternalReadResultBudget {
257                max_rows: 1,
258                max_memory_bytes: NonZeroUsize::MIN,
259            },
260            task_context: Some(&task_context),
261        };
262
263        assert!(contract.checkpoint().is_ok());
264        assert!(cancellation.cancel());
265        assert!(contract.checkpoint().is_err());
266    }
267}
hawdb_executor/external.rs

hawdb_executor/
external.rs
