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;
23use hawdb_core::{HawDBError, Result, RuntimeTaskContext};
24use hawdb_plan_cypher::VectorPhysicalPlan;
25use std::collections::BTreeMap;
26use std::mem::size_of;
27use std::num::NonZeroUsize;
28pub use text::{TextSeedExecutionOutput, TextSeedExecutionRequest, TextSeedExecutionRow};
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct ExternalReadResultBudget {
32    pub max_rows: usize,
33    pub max_memory_bytes: NonZeroUsize,
34}
35
36#[derive(Debug, Clone, Copy)]
37pub struct ExternalReadResourceContract<'a> {
38    pub priority: u8,
39    pub max_parallelism: NonZeroUsize,
40    pub max_working_memory_bytes: NonZeroUsize,
41    pub result: ExternalReadResultBudget,
42    pub task_context: Option<&'a RuntimeTaskContext>,
43}
44
45impl ExternalReadResourceContract<'_> {
46    pub fn checkpoint(&self) -> Result<()> {
47        match self.task_context {
48            Some(task_context) => task_context.checkpoint().map_err(|reason| {
49                HawDBError::Execution(format!("external read task stopped: {reason}"))
50            }),
51            None => Ok(()),
52        }
53    }
54
55    pub fn reserved_memory_bytes(&self) -> usize {
56        self.max_working_memory_bytes
57            .get()
58            .saturating_add(self.result.max_memory_bytes.get())
59    }
60}
61
62pub struct VectorSeedExecutionRequest<'a> {
63    pub embedding: &'a [f32],
64    pub metadata_filters: &'a BTreeMap<String, String>,
65    pub vector_plan: &'a VectorPhysicalPlan,
66    pub resources: ExternalReadResourceContract<'a>,
67}
68
69pub struct VectorSeedExecutionRow {
70    pub id: String,
71    pub external_id: Option<String>,
72    pub score: f64,
73}
74
75pub struct VectorSeedExecutionOutput {
76    pub rows: Vec<VectorSeedExecutionRow>,
77    pub report: VectorExecutionReport,
78}
79
80impl VectorSeedExecutionOutput {
81    pub fn estimated_memory_bytes(&self) -> usize {
82        self.rows
83            .capacity()
84            .saturating_mul(size_of::<VectorSeedExecutionRow>())
85            .saturating_add(self.rows.iter().fold(0usize, |total, row| {
86                total.saturating_add(row.id.capacity()).saturating_add(
87                    row.external_id
88                        .as_ref()
89                        .map_or(0, |external_id| external_id.capacity()),
90                )
91            }))
92    }
93
94    pub fn validate_result_budget(&self, budget: ExternalReadResultBudget) -> Result<()> {
95        if self.rows.len() > budget.max_rows {
96            return Err(HawDBError::read_budget_exceeded(
97                hawdb_core::ReadBudgetResource::Rows,
98                budget.max_rows,
99                format!(
100                    "external vector read returned {} rows, exceeding result row budget {}",
101                    self.rows.len(),
102                    budget.max_rows
103                ),
104            ));
105        }
106        let memory_bytes = self.estimated_memory_bytes();
107        if memory_bytes > budget.max_memory_bytes.get() {
108            return Err(HawDBError::Execution(format!(
109                "external vector read returned {memory_bytes} estimated bytes, exceeding result memory budget {}",
110                budget.max_memory_bytes
111            )));
112        }
113        Ok(())
114    }
115}
116
117pub trait ExternalReadOperator {
118    fn execute_text_seed(
119        &mut self,
120        request: TextSeedExecutionRequest<'_>,
121    ) -> Result<TextSeedExecutionOutput> {
122        request.resources.checkpoint()?;
123        Err(HawDBError::Execution(
124            "text search capability is unavailable without a text projection".into(),
125        ))
126    }
127
128    fn execute_vector_seed(
129        &mut self,
130        request: VectorSeedExecutionRequest<'_>,
131    ) -> Result<VectorSeedExecutionOutput>;
132}
133
134#[doc(hidden)]
135pub struct NoExternalReadOperator;
136
137impl ExternalReadOperator for NoExternalReadOperator {
138    fn execute_vector_seed(
139        &mut self,
140        _request: VectorSeedExecutionRequest<'_>,
141    ) -> Result<VectorSeedExecutionOutput> {
142        Err(HawDBError::Execution(
143            "vector search capability is unavailable without a search projection".to_string(),
144        ))
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151    use crate::{
152        VectorCompressionMode, VectorExecutionBackend, VectorFallbackReasonCode, VectorScoreSource,
153    };
154    use hawdb_core::RuntimeCancellationToken;
155    use hawdb_plan_cypher::VectorCandidateSource;
156
157    pub(super) fn empty_report() -> VectorExecutionReport {
158        VectorExecutionReport {
159            backend: VectorExecutionBackend::ScalarFlat,
160            compression_mode: VectorCompressionMode::Disabled,
161            candidate_source: VectorCandidateSource::Scalar,
162            backend_selection_reason: None,
163            estimated_raw_vector_bytes: None,
164            filter_selectivity_per_million: None,
165            candidate_score_source: VectorScoreSource::RawVector,
166            final_score_source: VectorScoreSource::RawVector,
167            generated_candidate_count: 0,
168            descriptor_pruned_count: 0,
169            scalar_filtered_count: 0,
170            residual_filtered_count: 0,
171            candidate_scan_rounds: 0,
172            reranked_candidate_count: 0,
173            returned_count: 0,
174            raw_vector_bytes_read: 0,
175            candidate_scan_metrics: None,
176            index_covered_document_count: None,
177            index_candidate_document_count: None,
178            index_coverage_complete: None,
179            fallback_reason_codes: Vec::<VectorFallbackReasonCode>::new(),
180        }
181    }
182
183    #[test]
184    fn external_vector_output_fails_closed_on_row_and_memory_budgets() {
185        let output = VectorSeedExecutionOutput {
186            rows: vec![VectorSeedExecutionRow {
187                id: "memory-1".to_string(),
188                external_id: Some("external-1".to_string()),
189                score: 1.0,
190            }],
191            report: empty_report(),
192        };
193
194        let row_error = output
195            .validate_result_budget(ExternalReadResultBudget {
196                max_rows: 0,
197                max_memory_bytes: NonZeroUsize::new(1024).unwrap(),
198            })
199            .unwrap_err();
200        assert!(matches!(row_error, HawDBError::ReadBudgetExceeded(cause)
201            if cause.resource == hawdb_core::ReadBudgetResource::Rows && cause.limit == 0));
202        let memory_error = output
203            .validate_result_budget(ExternalReadResultBudget {
204                max_rows: 1,
205                max_memory_bytes: NonZeroUsize::MIN,
206            })
207            .unwrap_err();
208        assert!(matches!(memory_error, HawDBError::Execution(_)));
209    }
210
211    #[test]
212    fn external_read_contract_observes_cancellation() {
213        let cancellation = RuntimeCancellationToken::new();
214        let task_context = RuntimeTaskContext::without_deadline(cancellation.clone());
215        let contract = ExternalReadResourceContract {
216            priority: 128,
217            max_parallelism: NonZeroUsize::MIN,
218            max_working_memory_bytes: NonZeroUsize::MIN,
219            result: ExternalReadResultBudget {
220                max_rows: 1,
221                max_memory_bytes: NonZeroUsize::MIN,
222            },
223            task_context: Some(&task_context),
224        };
225
226        assert!(contract.checkpoint().is_ok());
227        assert!(cancellation.cancel());
228        assert!(contract.checkpoint().is_err());
229    }
230}
hawdb_executor/external.rs

hawdb_executor/
external.rs
