1use std::{
2 cmp::Ordering,
3 slice,
4 time::{SystemTime, UNIX_EPOCH},
5};
6
7use radixdb_core::{DataType, ExternalTypeRef, Value};
8use radixdb_plugin_abi as abi;
9use thiserror::Error;
10
11use crate::{
12 ObjectId, PluginRegistry, RegisteredExternalType, RegisteredFunction, RegisteredTypeRef,
13};
14
15const DEFAULT_WORK_UNITS: u32 = 1_000_000;
16const MAX_NORMALIZED_PREDICATE_BYTES: usize = 1024 * 1024;
17
18#[derive(Debug, Error, Clone, PartialEq, Eq)]
19pub enum PluginInvocationError {
20 #[error("external type {0:02x?} is not present in the immutable plugin registry")]
21 UnknownType([u8; 16]),
22 #[error("native function {0:02x?} is not present in the immutable plugin registry")]
23 UnknownFunction([u8; 16]),
24 #[error("operator class {0:02x?} is not present in the immutable plugin registry")]
25 UnknownOperatorClass([u8; 16]),
26 #[error("planner support {0:02x?} is not present in the immutable plugin registry")]
27 UnknownPlannerSupport([u8; 16]),
28 #[error("external value codec version {actual} does not match admitted version {expected}")]
29 CodecVersion { expected: u32, actual: u32 },
30 #[error("external type does not provide the required {0} capability")]
31 MissingCapability(&'static str),
32 #[error("external value exceeds admitted payload limit")]
33 PayloadLimit,
34 #[error("plugin callback failed with status {status}: {diagnostic}")]
35 Callback { status: u32, diagnostic: String },
36 #[error("plugin callback violated the ABI result contract: {0}")]
37 Contract(&'static str),
38}
39
40impl PluginInvocationError {
41 pub const fn status_code(&self) -> abi::RadixAbiStatusV1 {
42 match self {
43 Self::UnknownType(_)
44 | Self::UnknownFunction(_)
45 | Self::UnknownOperatorClass(_)
46 | Self::UnknownPlannerSupport(_)
47 | Self::MissingCapability(_) => abi::RADIX_STATUS_INVALID_ARGUMENT,
48 Self::CodecVersion { .. } | Self::Contract(_) => abi::RADIX_STATUS_CONTRACT_VIOLATION,
49 Self::PayloadLimit => abi::RADIX_STATUS_LIMIT_EXCEEDED,
50 Self::Callback { status, .. } => *status,
51 }
52 }
53}
54
55#[derive(Debug, Clone, Copy)]
56pub enum NormalizedPredicateArgument<'a> {
57 IndexedColumn,
58 Constant(&'a Value),
59}
60
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct CandidateKeySpan {
63 pub start: Value,
64 pub end: Value,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Hash)]
68pub struct PlannerPlanIdentity {
69 pub registry_generation: u64,
70 pub support_id: ObjectId,
71 pub support_semantic_revision: u32,
72 pub support_fingerprint: [u8; 32],
73 pub target_function_id: ObjectId,
74 pub function_semantic_revision: u32,
75 pub operator_class_id: ObjectId,
76 pub operator_class_semantic_revision: u32,
77 pub key_codec_revision: u32,
78 pub operator_class_fingerprint: [u8; 32],
79}
80
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct CandidatePlan {
83 pub spans: Vec<CandidateKeySpan>,
84 pub requires_recheck: bool,
85 pub estimated_rows: u64,
86 pub cost_hint: u32,
87 pub identity: PlannerPlanIdentity,
88}
89
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub enum PlannerSupportOutcome {
92 Plan(CandidatePlan),
93 Fallback { reason: String },
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct HashComponent {
98 pub kind: u16,
99 pub bytes: Vec<u8>,
100}
101
102#[derive(Default)]
103struct CallState {
104 staged: Vec<Vec<u8>>,
105 staged_nulls: Vec<bool>,
106 finished: bool,
107 output_bytes: usize,
108 max_output_bytes: u32,
109 max_items: u32,
110 work_units: u32,
111 diagnostic: String,
112 hash_components: Vec<HashComponent>,
113 hash_bytes: usize,
114 cancel_check: Option<fn() -> bool>,
115 deadline_unix_ns: u64,
116}
117
118impl CallState {
119 fn new(max_output_bytes: u32, max_items: u32) -> Self {
120 Self {
121 max_output_bytes,
122 max_items,
123 deadline_unix_ns: u64::MAX,
124 ..Self::default()
125 }
126 }
127
128 fn with_limits(mut self, limits: InvocationLimits) -> Self {
129 self.cancel_check = limits.cancel_check;
130 self.deadline_unix_ns = limits.deadline_unix_ns;
131 self
132 }
133
134 fn handle(&mut self) -> u64 {
135 self as *mut Self as usize as u64
136 }
137
138 fn diagnostic_sink(&mut self) -> abi::RadixAbiDiagnosticSinkV1 {
139 abi::RadixAbiDiagnosticSinkV1 {
140 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiDiagnosticSinkV1>(0),
141 handle: self.handle(),
142 max_detail_bytes: abi::RADIX_MAX_DIAGNOSTIC_BYTES,
143 reserved: 0,
144 write: Some(write_diagnostic),
145 }
146 }
147
148 fn context(
149 &mut self,
150 diagnostics: &abi::RadixAbiDiagnosticSinkV1,
151 ) -> abi::RadixAbiCallContextV1 {
152 abi::RadixAbiCallContextV1 {
153 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiCallContextV1>(0),
154 handle: self.handle(),
155 deadline_unix_ns: self.deadline_unix_ns,
156 max_output_bytes: self.max_output_bytes.max(1),
157 max_work_units: DEFAULT_WORK_UNITS,
158 check_cancelled: Some(check_cancelled),
159 charge_work: Some(charge_work),
160 diagnostics,
161 }
162 }
163
164 fn result_builder(&mut self) -> abi::RadixAbiResultBuilderV1 {
165 abi::RadixAbiResultBuilderV1 {
166 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiResultBuilderV1>(0),
167 handle: self.handle(),
168 max_bytes: self.max_output_bytes.max(1),
169 max_items: self.max_items.max(1),
170 write: Some(write_result),
171 finish: Some(finish_result),
172 }
173 }
174
175 fn hash_sink(&mut self) -> abi::RadixAbiHashSinkV1 {
176 abi::RadixAbiHashSinkV1 {
177 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiHashSinkV1>(0),
178 handle: self.handle(),
179 max_components: abi::RADIX_MAX_HASH_COMPONENTS,
180 max_bytes: abi::RADIX_MAX_HASH_BYTES,
181 append: Some(append_hash),
182 }
183 }
184
185 fn callback_error(&self, status: abi::RadixAbiStatusV1) -> PluginInvocationError {
186 PluginInvocationError::Callback {
187 status,
188 diagnostic: if self.diagnostic.is_empty() {
189 "plugin returned no diagnostic".to_owned()
190 } else {
191 self.diagnostic.clone()
192 },
193 }
194 }
195}
196
197unsafe fn state(handle: u64) -> &'static mut CallState {
198 unsafe { &mut *(handle as usize as *mut CallState) }
201}
202
203unsafe extern "C" fn check_cancelled(_handle: u64) -> abi::RadixAbiStatusV1 {
204 let state = unsafe { state(_handle) };
205 let cancelled = state.cancel_check.is_some_and(|check| check());
206 let expired = state.deadline_unix_ns != u64::MAX
207 && SystemTime::now()
208 .duration_since(UNIX_EPOCH)
209 .map_or(true, |now| {
210 now.as_nanos() >= u128::from(state.deadline_unix_ns)
211 });
212 if cancelled || expired {
213 abi::RADIX_STATUS_CANCELLED
214 } else {
215 abi::RADIX_STATUS_OK
216 }
217}
218
219unsafe extern "C" fn charge_work(handle: u64, units: u32) -> abi::RadixAbiStatusV1 {
220 let state = unsafe { state(handle) };
221 let Some(total) = state.work_units.checked_add(units) else {
222 return abi::RADIX_STATUS_LIMIT_EXCEEDED;
223 };
224 if total > DEFAULT_WORK_UNITS {
225 abi::RADIX_STATUS_LIMIT_EXCEEDED
226 } else {
227 state.work_units = total;
228 abi::RADIX_STATUS_OK
229 }
230}
231
232unsafe extern "C" fn write_diagnostic(
233 handle: u64,
234 diagnostic: *const abi::RadixAbiDiagnosticV1,
235) -> abi::RadixAbiStatusV1 {
236 let Some(diagnostic) = (unsafe { diagnostic.as_ref() }) else {
237 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
238 };
239 if abi::validate_diagnostic(diagnostic).is_err() {
240 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
241 }
242 let bytes = if diagnostic.detail.len == 0 {
243 &[][..]
244 } else {
245 unsafe { slice::from_raw_parts(diagnostic.detail.ptr, diagnostic.detail.len as usize) }
247 };
248 let Ok(detail) = std::str::from_utf8(bytes) else {
249 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
250 };
251 unsafe { state(handle) }.diagnostic = detail.to_owned();
252 abi::RADIX_STATUS_OK
253}
254
255unsafe extern "C" fn write_result(
256 handle: u64,
257 flags: u32,
258 reserved: u32,
259 bytes: abi::RadixAbiSliceV1,
260) -> abi::RadixAbiStatusV1 {
261 let state = unsafe { state(handle) };
262 if state.finished
263 || state.staged.len() >= state.max_items as usize
264 || abi::validate_result_item(flags, reserved, bytes, state.max_output_bytes).is_err()
265 {
266 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
267 }
268 let next = match state.output_bytes.checked_add(bytes.len as usize) {
269 Some(next) if next <= state.max_output_bytes as usize => next,
270 _ => return abi::RADIX_STATUS_LIMIT_EXCEEDED,
271 };
272 let copied = if bytes.len == 0 {
273 Vec::new()
274 } else {
275 unsafe { slice::from_raw_parts(bytes.ptr, bytes.len as usize) }.to_vec()
277 };
278 state.output_bytes = next;
279 state.staged.push(copied);
280 state
281 .staged_nulls
282 .push(flags & abi::RADIX_RESULT_ITEM_FLAG_NULL != 0);
283 abi::RADIX_STATUS_OK
284}
285
286unsafe extern "C" fn finish_result(handle: u64) -> abi::RadixAbiStatusV1 {
287 let state = unsafe { state(handle) };
288 if state.finished {
289 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
290 }
291 state.finished = true;
292 abi::RADIX_STATUS_OK
293}
294
295unsafe extern "C" fn append_hash(
296 handle: u64,
297 kind: u16,
298 reserved: u16,
299 bytes: abi::RadixAbiSliceV1,
300) -> abi::RadixAbiStatusV1 {
301 let state = unsafe { state(handle) };
302 let remaining = abi::RADIX_MAX_HASH_BYTES.saturating_sub(state.hash_bytes as u32);
303 if state.hash_components.len() >= abi::RADIX_MAX_HASH_COMPONENTS as usize
304 || abi::validate_hash_component(kind, reserved, bytes, remaining).is_err()
305 {
306 return abi::RADIX_STATUS_CONTRACT_VIOLATION;
307 }
308 let copied = if bytes.len == 0 {
309 Vec::new()
310 } else {
311 unsafe { slice::from_raw_parts(bytes.ptr, bytes.len as usize) }.to_vec()
313 };
314 state.hash_bytes += copied.len();
315 state.hash_components.push(HashComponent {
316 kind,
317 bytes: copied,
318 });
319 abi::RADIX_STATUS_OK
320}
321
322fn external<'r, 'v>(
323 registry: &'r PluginRegistry,
324 value: &'v Value,
325) -> Result<
326 (
327 &'r RegisteredExternalType,
328 radixdb_core::ExternalValueRef<'v>,
329 ),
330 PluginInvocationError,
331> {
332 let view = value
333 .as_external()
334 .ok_or(PluginInvocationError::Contract("value is not external"))?;
335 let registered = registry
336 .external_type(&view.type_ref().type_object_id())
337 .ok_or(PluginInvocationError::UnknownType(
338 view.type_ref().type_object_id(),
339 ))?;
340 if view.type_ref().codec_version() != registered.codec_version {
341 return Err(PluginInvocationError::CodecVersion {
342 expected: registered.codec_version,
343 actual: view.type_ref().codec_version(),
344 });
345 }
346 if view.payload().len() > registered.max_bytes as usize {
347 return Err(PluginInvocationError::PayloadLimit);
348 }
349 Ok((registered, view))
350}
351
352fn raw_value(view: radixdb_core::ExternalValueRef<'_>) -> abi::RadixAbiValueV1 {
353 abi::RadixAbiValueV1 {
354 type_ref: abi::RadixAbiTypeRefV1::external(
355 view.type_ref().type_object_id(),
356 view.type_ref().codec_version(),
357 ),
358 flags: 0,
359 reserved: 0,
360 inline_bytes: [0; 16],
361 borrowed_bytes: abi::RadixAbiSliceV1 {
362 ptr: view.payload().as_ptr(),
363 len: view.payload().len() as u32,
364 reserved: 0,
365 },
366 }
367}
368
369fn run_parse(
370 callback: abi::RadixAbiParseFnV1,
371 input: &[u8],
372 maximum: u32,
373) -> Result<Vec<u8>, PluginInvocationError> {
374 if input.len() > u32::MAX as usize {
375 return Err(PluginInvocationError::PayloadLimit);
376 }
377 let mut state = CallState::new(maximum, 1);
378 let diagnostics = state.diagnostic_sink();
379 let context = state.context(&diagnostics);
380 let output = state.result_builder();
381 let input = abi::RadixAbiSliceV1 {
382 ptr: input.as_ptr(),
383 len: input.len() as u32,
384 reserved: 0,
385 };
386 let status = unsafe { callback(&context, input, &output) };
388 finish_single_output(state, status)
389}
390
391fn run_codec(
392 callback: abi::RadixAbiCodecFnV1,
393 input: &abi::RadixAbiValueV1,
394 maximum: u32,
395) -> Result<Vec<u8>, PluginInvocationError> {
396 let mut state = CallState::new(maximum, 1);
397 let diagnostics = state.diagnostic_sink();
398 let context = state.context(&diagnostics);
399 let output = state.result_builder();
400 let status = unsafe { callback(&context, input, &output) };
402 finish_single_output(state, status)
403}
404
405fn finish_single_output(
406 mut state: CallState,
407 status: abi::RadixAbiStatusV1,
408) -> Result<Vec<u8>, PluginInvocationError> {
409 if status != abi::RADIX_STATUS_OK {
410 return Err(state.callback_error(status));
411 }
412 if !state.finished || state.staged.len() != 1 {
413 return Err(PluginInvocationError::Contract(
414 "callback did not finish exactly one result",
415 ));
416 }
417 if state.staged_nulls != [false] {
418 return Err(PluginInvocationError::Contract(
419 "codec callback returned NULL",
420 ));
421 }
422 Ok(state.staged.pop().expect("one checked result"))
423}
424
425#[derive(Debug, Clone, Copy)]
426pub struct InvocationLimits {
427 pub cancel_check: Option<fn() -> bool>,
428 pub deadline_unix_ns: u64,
429}
430
431impl Default for InvocationLimits {
432 fn default() -> Self {
433 Self {
434 cancel_check: None,
435 deadline_unix_ns: u64::MAX,
436 }
437 }
438}
439
440impl InvocationLimits {
441 fn cancelled_or_expired(self) -> bool {
442 self.cancel_check.is_some_and(|check| check())
443 || (self.deadline_unix_ns != u64::MAX
444 && SystemTime::now()
445 .duration_since(UNIX_EPOCH)
446 .map_or(true, |now| {
447 now.as_nanos() >= u128::from(self.deadline_unix_ns)
448 }))
449 }
450}
451
452fn matching_pair<'a>(
453 registry: &'a PluginRegistry,
454 left: &Value,
455 right: &Value,
456) -> Result<
457 (
458 &'a RegisteredExternalType,
459 abi::RadixAbiValueV1,
460 abi::RadixAbiValueV1,
461 ),
462 PluginInvocationError,
463> {
464 let (left_type, left_view) = external(registry, left)?;
465 let (_, right_view) = external(registry, right)?;
466 if left_view.type_ref() != right_view.type_ref() {
467 return Err(PluginInvocationError::Contract(
468 "external comparison requires identical type identity and codec version",
469 ));
470 }
471 Ok((left_type, raw_value(left_view), raw_value(right_view)))
472}
473
474fn abi_type_ref(value: RegisteredTypeRef) -> abi::RadixAbiTypeRefV1 {
475 match value {
476 RegisteredTypeRef::Builtin(tag) => abi::RadixAbiTypeRefV1::builtin(tag),
477 RegisteredTypeRef::External {
478 object_id,
479 codec_version,
480 } => abi::RadixAbiTypeRefV1::external(object_id, codec_version),
481 }
482}
483
484fn raw_argument(
485 value: &Value,
486 expected: RegisteredTypeRef,
487 backing: &mut Vec<Vec<u8>>,
488) -> Result<abi::RadixAbiValueV1, PluginInvocationError> {
489 let type_ref = abi_type_ref(expected);
490 if value.is_null() {
491 return Ok(abi::RadixAbiValueV1 {
492 type_ref,
493 flags: abi::RADIX_VALUE_FLAG_NULL,
494 reserved: 0,
495 inline_bytes: [0; 16],
496 borrowed_bytes: abi::RadixAbiSliceV1::EMPTY,
497 });
498 }
499 let mut inline_bytes = [0; 16];
500 let variable = match expected {
501 RegisteredTypeRef::External {
502 object_id,
503 codec_version,
504 } => {
505 let external = value.as_external().ok_or(PluginInvocationError::Contract(
506 "native function argument type differs from descriptor",
507 ))?;
508 if external.type_ref().type_object_id() != object_id
509 || external.type_ref().codec_version() != codec_version
510 {
511 return Err(PluginInvocationError::Contract(
512 "native function external argument identity differs from descriptor",
513 ));
514 }
515 Some(external.payload().to_vec())
516 }
517 RegisteredTypeRef::Builtin(tag) => {
518 let actual = u16::from(value.data_type().as_u8());
519 if actual != tag {
520 return Err(PluginInvocationError::Contract(
521 "native function built-in argument type differs from descriptor",
522 ));
523 }
524 match tag {
525 abi::RADIX_BUILTIN_INTEGER => {
526 inline_bytes[..8].copy_from_slice(
527 &value
528 .as_int64()
529 .expect("exact INTEGER type was checked")
530 .to_le_bytes(),
531 );
532 None
533 }
534 abi::RADIX_BUILTIN_FLOAT => {
535 inline_bytes[..8].copy_from_slice(
536 &value
537 .as_float64()
538 .expect("exact FLOAT type was checked")
539 .to_bits()
540 .to_le_bytes(),
541 );
542 None
543 }
544 abi::RADIX_BUILTIN_BOOLEAN => {
545 inline_bytes[0] =
546 u8::from(value.as_boolean().expect("exact BOOLEAN type was checked"));
547 None
548 }
549 abi::RADIX_BUILTIN_TIMESTAMP => {
550 inline_bytes[..8].copy_from_slice(
551 &value
552 .artifact_timestamp_nanos()
553 .ok_or(PluginInvocationError::Contract(
554 "timestamp is outside the ABI nanosecond range",
555 ))?
556 .to_le_bytes(),
557 );
558 None
559 }
560 abi::RADIX_BUILTIN_UUID => {
561 inline_bytes.copy_from_slice(
562 &value.as_uuid_bytes().expect("exact UUID type was checked"),
563 );
564 None
565 }
566 abi::RADIX_BUILTIN_DATE => {
567 inline_bytes[..4].copy_from_slice(
568 &value
569 .as_date_days()
570 .expect("exact DATE type was checked")
571 .to_le_bytes(),
572 );
573 None
574 }
575 abi::RADIX_BUILTIN_TEXT => Some(
576 value
577 .as_str()
578 .expect("exact TEXT type was checked")
579 .as_bytes()
580 .to_vec(),
581 ),
582 abi::RADIX_BUILTIN_JSON => Some(
583 value
584 .as_json()
585 .expect("exact JSON type was checked")
586 .as_bytes()
587 .to_vec(),
588 ),
589 abi::RADIX_BUILTIN_VECTOR => Some(
590 value
591 .as_vector_f32()
592 .expect("exact VECTOR type was checked")
593 .into_iter()
594 .flat_map(f32::to_le_bytes)
595 .collect(),
596 ),
597 abi::RADIX_BUILTIN_DECIMAL => {
598 let (unscaled, precision, scale) = value
599 .as_decimal_parts()
600 .expect("exact DECIMAL type was checked");
601 let mut bytes = unscaled.to_le_bytes().to_vec();
602 bytes.extend([precision, scale]);
603 Some(bytes)
604 }
605 abi::RADIX_BUILTIN_BYTES => Some(
606 value
607 .as_bytes_value()
608 .expect("exact BYTES type was checked")
609 .to_vec(),
610 ),
611 _ => {
612 return Err(PluginInvocationError::Contract(
613 "native function descriptor contains an unknown built-in type",
614 ))
615 }
616 }
617 }
618 };
619 let borrowed_bytes = if let Some(bytes) = variable {
620 backing.push(bytes);
621 let bytes = backing.last().expect("just pushed ABI backing bytes");
622 abi::RadixAbiSliceV1 {
623 ptr: bytes.as_ptr(),
624 len: bytes.len() as u32,
625 reserved: 0,
626 }
627 } else {
628 abi::RadixAbiSliceV1::EMPTY
629 };
630 let raw = abi::RadixAbiValueV1 {
631 type_ref,
632 flags: 0,
633 reserved: 0,
634 inline_bytes,
635 borrowed_bytes,
636 };
637 unsafe { abi::validate_value_contents(&raw) }
639 .map_err(|_| PluginInvocationError::Contract("host encoded an invalid ABI argument"))?;
640 Ok(raw)
641}
642
643fn encode_normalized_predicate(
644 target_function_id: ObjectId,
645 operator_class_id: ObjectId,
646 indexed_argument: usize,
647 arguments: &[NormalizedPredicateArgument<'_>],
648 expected: &[RegisteredTypeRef],
649) -> Result<Vec<u8>, PluginInvocationError> {
650 let indexed_argument = u16::try_from(indexed_argument).map_err(|_| {
651 PluginInvocationError::Contract("normalized predicate argument index exceeds u16")
652 })?;
653 let argument_count = u16::try_from(arguments.len()).map_err(|_| {
654 PluginInvocationError::Contract("normalized predicate argument count exceeds u16")
655 })?;
656 let mut output = Vec::new();
657 output.extend_from_slice(b"RPN1");
658 output.extend_from_slice(&target_function_id);
659 output.extend_from_slice(&operator_class_id);
660 output.extend_from_slice(&indexed_argument.to_le_bytes());
661 output.extend_from_slice(&argument_count.to_le_bytes());
662 let mut backing = Vec::with_capacity(arguments.len());
663 for (argument, expected) in arguments.iter().zip(expected.iter().copied()) {
664 let (kind, flags, bytes) = match argument {
665 NormalizedPredicateArgument::IndexedColumn => (1_u8, 0_u8, Vec::new()),
666 NormalizedPredicateArgument::Constant(value) => {
667 let raw = raw_argument(value, expected, &mut backing)?;
668 let bytes = if raw.borrowed_bytes.len != 0 {
669 unsafe {
671 slice::from_raw_parts(
672 raw.borrowed_bytes.ptr,
673 raw.borrowed_bytes.len as usize,
674 )
675 }
676 .to_vec()
677 } else if raw.flags & abi::RADIX_VALUE_FLAG_NULL != 0 {
678 Vec::new()
679 } else {
680 let width = normalized_fixed_width(expected)?;
681 raw.inline_bytes[..width].to_vec()
682 };
683 (
684 2_u8,
685 u8::from(raw.flags & abi::RADIX_VALUE_FLAG_NULL != 0),
686 bytes,
687 )
688 }
689 };
690 let type_ref = abi_type_ref(expected);
691 let len = u32::try_from(bytes.len()).map_err(|_| PluginInvocationError::PayloadLimit)?;
692 output.push(kind);
693 output.push(flags);
694 output.extend_from_slice(&0_u16.to_le_bytes());
695 output.extend_from_slice(&type_ref.kind.to_le_bytes());
696 output.extend_from_slice(&type_ref.builtin_tag.to_le_bytes());
697 output.extend_from_slice(&type_ref.object_id);
698 output.extend_from_slice(&type_ref.codec_version.to_le_bytes());
699 output.extend_from_slice(&len.to_le_bytes());
700 output.extend_from_slice(&bytes);
701 }
702 if output.len() > MAX_NORMALIZED_PREDICATE_BYTES {
703 return Err(PluginInvocationError::PayloadLimit);
704 }
705 Ok(output)
706}
707
708fn normalized_fixed_width(expected: RegisteredTypeRef) -> Result<usize, PluginInvocationError> {
709 match expected {
710 RegisteredTypeRef::External { .. } => Ok(0),
711 RegisteredTypeRef::Builtin(tag) => match tag {
712 abi::RADIX_BUILTIN_INTEGER
713 | abi::RADIX_BUILTIN_FLOAT
714 | abi::RADIX_BUILTIN_TIMESTAMP => Ok(8),
715 abi::RADIX_BUILTIN_BOOLEAN => Ok(1),
716 abi::RADIX_BUILTIN_UUID => Ok(16),
717 abi::RADIX_BUILTIN_DATE => Ok(4),
718 abi::RADIX_BUILTIN_TEXT
719 | abi::RADIX_BUILTIN_JSON
720 | abi::RADIX_BUILTIN_VECTOR
721 | abi::RADIX_BUILTIN_DECIMAL
722 | abi::RADIX_BUILTIN_BYTES => Ok(0),
723 _ => Err(PluginInvocationError::Contract(
724 "normalized predicate contains an unknown built-in type",
725 )),
726 },
727 }
728}
729
730fn decode_function_result(
731 registry: &PluginRegistry,
732 expected: RegisteredTypeRef,
733 is_null: bool,
734 bytes: Vec<u8>,
735) -> Result<Value, PluginInvocationError> {
736 if is_null {
737 return Ok(Value::Null(match expected {
738 RegisteredTypeRef::Builtin(tag) => u8::try_from(tag)
739 .ok()
740 .and_then(DataType::from_u8)
741 .unwrap_or(DataType::Null),
742 RegisteredTypeRef::External { .. } => DataType::Null,
743 }));
744 }
745 let value = match expected {
746 RegisteredTypeRef::External {
747 object_id,
748 codec_version,
749 } => Value::try_external(
750 ExternalTypeRef::new(object_id, codec_version)
751 .map_err(|_| PluginInvocationError::Contract("invalid result type identity"))?,
752 bytes,
753 )
754 .map_err(|_| PluginInvocationError::PayloadLimit)?,
755 RegisteredTypeRef::Builtin(tag) => match tag {
756 abi::RADIX_BUILTIN_INTEGER if bytes.len() == 8 => Value::Integer(i64::from_le_bytes(
757 bytes.as_slice().try_into().expect("length checked"),
758 )),
759 abi::RADIX_BUILTIN_FLOAT if bytes.len() == 8 => Value::Float(f64::from_bits(
760 u64::from_le_bytes(bytes.as_slice().try_into().expect("length checked")),
761 )),
762 abi::RADIX_BUILTIN_BOOLEAN if bytes.as_slice() == [0] => Value::Boolean(false),
763 abi::RADIX_BUILTIN_BOOLEAN if bytes.as_slice() == [1] => Value::Boolean(true),
764 abi::RADIX_BUILTIN_TIMESTAMP if bytes.len() == 8 => Value::Integer(i64::from_le_bytes(
765 bytes.as_slice().try_into().expect("length checked"),
766 ))
767 .try_coerce_to_type(DataType::Timestamp)
768 .map_err(|_| PluginInvocationError::Contract("invalid TIMESTAMP result"))?,
769 abi::RADIX_BUILTIN_TEXT => Value::text(
770 String::from_utf8(bytes)
771 .map_err(|_| PluginInvocationError::Contract("TEXT result is not UTF-8"))?,
772 ),
773 abi::RADIX_BUILTIN_JSON => Value::try_json(
774 String::from_utf8(bytes)
775 .map_err(|_| PluginInvocationError::Contract("JSON result is not UTF-8"))?,
776 )
777 .map_err(|_| PluginInvocationError::Contract("JSON result is invalid"))?,
778 abi::RADIX_BUILTIN_VECTOR => Value::try_vector_from_bytes(bytes.into())
779 .map_err(|_| PluginInvocationError::Contract("VECTOR result is invalid"))?,
780 abi::RADIX_BUILTIN_UUID if bytes.len() == 16 => {
781 Value::uuid(bytes.as_slice().try_into().expect("length checked"))
782 }
783 abi::RADIX_BUILTIN_DECIMAL if bytes.len() == 18 => Value::try_decimal(
784 i128::from_le_bytes(bytes[..16].try_into().expect("length checked")),
785 bytes[16],
786 bytes[17],
787 )
788 .map_err(|_| PluginInvocationError::Contract("DECIMAL result is invalid"))?,
789 abi::RADIX_BUILTIN_DATE if bytes.len() == 4 => Value::date(i32::from_le_bytes(
790 bytes.as_slice().try_into().expect("length checked"),
791 )),
792 abi::RADIX_BUILTIN_BYTES => Value::bytes(bytes),
793 _ => {
794 return Err(PluginInvocationError::Contract(
795 "native function returned bytes incompatible with its result type",
796 ))
797 }
798 },
799 };
800 if matches!(expected, RegisteredTypeRef::External { .. }) {
801 registry.validate_external_value(&value)?;
802 } else {
803 value
804 .validate_shape()
805 .map_err(|_| PluginInvocationError::Contract("native function result is malformed"))?;
806 }
807 Ok(value)
808}
809
810fn finish_function_output(
811 registry: &PluginRegistry,
812 function: &RegisteredFunction,
813 mut state: CallState,
814 status: abi::RadixAbiStatusV1,
815) -> Result<Value, PluginInvocationError> {
816 if status != abi::RADIX_STATUS_OK {
817 return Err(state.callback_error(status));
818 }
819 if !state.finished || state.staged.len() != 1 || state.staged_nulls.len() != 1 {
820 return Err(PluginInvocationError::Contract(
821 "native function did not finish exactly one result",
822 ));
823 }
824 let bytes = state.staged.pop().expect("one checked result");
825 let is_null = state.staged_nulls.pop().expect("one checked null flag");
826 decode_function_result(registry, function.result, is_null, bytes)
827}
828
829enum BatchColumnStorage {
830 Aligned(Vec<u64>),
831 Bytes(Vec<u8>),
832}
833
834impl BatchColumnStorage {
835 fn bytes(&self) -> (*const u8, usize) {
836 match self {
837 Self::Aligned(words) => (words.as_ptr().cast(), words.len() * 8),
838 Self::Bytes(bytes) => (bytes.as_ptr(), bytes.len()),
839 }
840 }
841}
842
843struct OwnedBatchColumn {
844 type_ref: abi::RadixAbiTypeRefV1,
845 layout: u16,
846 element_width: u16,
847 alignment: u16,
848 stride: u32,
849 null_bitmap: Vec<u8>,
850 storage: BatchColumnStorage,
851 offsets: Vec<u32>,
852}
853
854impl OwnedBatchColumn {
855 fn new(
856 registry: &PluginRegistry,
857 expected: RegisteredTypeRef,
858 values: &[&Value],
859 ) -> Result<Self, PluginInvocationError> {
860 let type_ref = abi_type_ref(expected);
861 let mut backing = Vec::with_capacity(values.len());
862 let raw = values
863 .iter()
864 .map(|value| raw_argument(value, expected, &mut backing))
865 .collect::<Result<Vec<_>, _>>()?;
866 let mut null_bitmap = vec![0_u8; values.len().div_ceil(8)];
867 for (index, value) in values.iter().enumerate() {
868 if value.is_null() {
869 null_bitmap[index / 8] |= 1 << (index % 8);
870 }
871 }
872 if null_bitmap.iter().all(|byte| *byte == 0) {
873 null_bitmap.clear();
874 }
875
876 if let Some(width) = batch_fixed_width(registry, expected)? {
877 let stride = width.div_ceil(8) * 8;
878 let total = stride
879 .checked_mul(values.len())
880 .ok_or(PluginInvocationError::PayloadLimit)?;
881 let mut words = vec![0_u64; total.div_ceil(8)];
882 let bytes = unsafe {
885 slice::from_raw_parts_mut(words.as_mut_ptr().cast::<u8>(), words.len() * 8)
886 };
887 for (row, value) in raw.iter().enumerate() {
888 if value.flags & abi::RADIX_VALUE_FLAG_NULL != 0 {
889 continue;
890 }
891 let source = if value.borrowed_bytes.len == 0 {
892 &value.inline_bytes[..width]
893 } else {
894 unsafe {
897 slice::from_raw_parts(
898 value.borrowed_bytes.ptr,
899 value.borrowed_bytes.len as usize,
900 )
901 }
902 };
903 if source.len() != width {
904 return Err(PluginInvocationError::Contract(
905 "native batch fixed-width argument has the wrong width",
906 ));
907 }
908 let start = row * stride;
909 bytes[start..start + width].copy_from_slice(source);
910 }
911 return Ok(Self {
912 type_ref,
913 layout: abi::RADIX_COLUMN_LAYOUT_FIXED,
914 element_width: u16::try_from(width)
915 .map_err(|_| PluginInvocationError::PayloadLimit)?,
916 alignment: 8,
917 stride: u32::try_from(stride).map_err(|_| PluginInvocationError::PayloadLimit)?,
918 null_bitmap,
919 storage: BatchColumnStorage::Aligned(words),
920 offsets: Vec::new(),
921 });
922 }
923
924 let mut bytes = Vec::new();
925 let mut offsets = Vec::with_capacity(values.len() + 1);
926 offsets.push(0);
927 for value in &raw {
928 if value.flags & abi::RADIX_VALUE_FLAG_NULL == 0 {
929 bytes.extend_from_slice(unsafe {
932 slice::from_raw_parts(
933 value.borrowed_bytes.ptr,
934 value.borrowed_bytes.len as usize,
935 )
936 });
937 }
938 offsets
939 .push(u32::try_from(bytes.len()).map_err(|_| PluginInvocationError::PayloadLimit)?);
940 }
941 Ok(Self {
942 type_ref,
943 layout: abi::RADIX_COLUMN_LAYOUT_VARIABLE,
944 element_width: 0,
945 alignment: 1,
946 stride: 0,
947 null_bitmap,
948 storage: BatchColumnStorage::Bytes(bytes),
949 offsets,
950 })
951 }
952
953 fn as_abi(&self, row_count: u32) -> abi::RadixAbiColumnViewV1 {
954 let (data, data_len) = self.storage.bytes();
955 abi::RadixAbiColumnViewV1 {
956 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiColumnViewV1>(0),
957 type_ref: self.type_ref,
958 row_count,
959 layout: self.layout,
960 element_width: self.element_width,
961 alignment: self.alignment,
962 reserved_u16: 0,
963 stride: self.stride,
964 null_bitmap: abi::RadixAbiSliceV1 {
965 ptr: if self.null_bitmap.is_empty() {
966 std::ptr::null()
967 } else {
968 self.null_bitmap.as_ptr()
969 },
970 len: self.null_bitmap.len() as u32,
971 reserved: 0,
972 },
973 data: abi::RadixAbiSliceV1 {
974 ptr: if data_len == 0 {
975 std::ptr::null()
976 } else {
977 data
978 },
979 len: data_len as u32,
980 reserved: 0,
981 },
982 offsets: abi::RadixAbiU32SliceV1 {
983 ptr: if self.offsets.is_empty() {
984 std::ptr::null()
985 } else {
986 self.offsets.as_ptr()
987 },
988 len: self.offsets.len() as u32,
989 reserved: 0,
990 },
991 }
992 }
993}
994
995fn batch_fixed_width(
996 registry: &PluginRegistry,
997 expected: RegisteredTypeRef,
998) -> Result<Option<usize>, PluginInvocationError> {
999 match expected {
1000 RegisteredTypeRef::Builtin(tag) => Ok(match tag {
1001 abi::RADIX_BUILTIN_INTEGER
1002 | abi::RADIX_BUILTIN_FLOAT
1003 | abi::RADIX_BUILTIN_TIMESTAMP => Some(8),
1004 abi::RADIX_BUILTIN_BOOLEAN => Some(1),
1005 abi::RADIX_BUILTIN_UUID => Some(16),
1006 abi::RADIX_BUILTIN_DECIMAL => Some(18),
1007 abi::RADIX_BUILTIN_DATE => Some(4),
1008 abi::RADIX_BUILTIN_TEXT
1009 | abi::RADIX_BUILTIN_JSON
1010 | abi::RADIX_BUILTIN_VECTOR
1011 | abi::RADIX_BUILTIN_BYTES => None,
1012 _ => {
1013 return Err(PluginInvocationError::Contract(
1014 "native function descriptor contains an unknown built-in type",
1015 ))
1016 }
1017 }),
1018 RegisteredTypeRef::External { object_id, .. } => {
1019 let registered = registry
1020 .external_type(&object_id)
1021 .ok_or(PluginInvocationError::UnknownType(object_id))?;
1022 Ok(
1023 (registered.storage_kind == abi::RADIX_EXTERNAL_STORAGE_FIXED)
1024 .then_some(registered.fixed_bytes as usize),
1025 )
1026 }
1027 }
1028}
1029
1030fn finish_batch_output(
1031 registry: &PluginRegistry,
1032 function: &RegisteredFunction,
1033 state: CallState,
1034 status: abi::RadixAbiStatusV1,
1035 expected_rows: usize,
1036) -> Result<Vec<Value>, PluginInvocationError> {
1037 if status != abi::RADIX_STATUS_OK {
1038 return Err(state.callback_error(status));
1039 }
1040 if !state.finished
1041 || state.staged.len() != expected_rows
1042 || state.staged_nulls.len() != expected_rows
1043 {
1044 return Err(PluginInvocationError::Contract(
1045 "native batch did not finish exactly one result per row",
1046 ));
1047 }
1048 state
1049 .staged
1050 .into_iter()
1051 .zip(state.staged_nulls)
1052 .map(|(bytes, is_null)| decode_function_result(registry, function.result, is_null, bytes))
1053 .collect()
1054}
1055
1056impl PluginRegistry {
1057 pub fn invoke_planner_support(
1061 &self,
1062 support_id: ObjectId,
1063 arguments: &[NormalizedPredicateArgument<'_>],
1064 limits: InvocationLimits,
1065 ) -> Result<PlannerSupportOutcome, PluginInvocationError> {
1066 let support = self
1067 .planner_support(&support_id)
1068 .ok_or(PluginInvocationError::UnknownPlannerSupport(support_id))?;
1069 let target_function_id =
1070 support
1071 .target_function_id
1072 .ok_or(PluginInvocationError::Contract(
1073 "planner support has no target function",
1074 ))?;
1075 let operator_class_id =
1076 support
1077 .target_operator_class_id
1078 .ok_or(PluginInvocationError::Contract(
1079 "planner support has no target operator class",
1080 ))?;
1081 let function = self
1082 .function(&target_function_id)
1083 .ok_or(PluginInvocationError::UnknownFunction(target_function_id))?;
1084 let operator_class = self.operator_class(&operator_class_id).ok_or(
1085 PluginInvocationError::UnknownOperatorClass(operator_class_id),
1086 )?;
1087 if arguments.len() != function.arguments.len() {
1088 return Err(PluginInvocationError::Contract(
1089 "normalized predicate arity differs from target function",
1090 ));
1091 }
1092 let indexed = arguments
1093 .iter()
1094 .position(|argument| matches!(argument, NormalizedPredicateArgument::IndexedColumn))
1095 .ok_or(PluginInvocationError::Contract(
1096 "normalized predicate has no indexed argument",
1097 ))?;
1098 if arguments
1099 .iter()
1100 .skip(indexed + 1)
1101 .any(|argument| matches!(argument, NormalizedPredicateArgument::IndexedColumn))
1102 || function.arguments[indexed] != operator_class.input_type
1103 {
1104 return Err(PluginInvocationError::Contract(
1105 "normalized predicate has an invalid indexed argument",
1106 ));
1107 }
1108 if limits.cancelled_or_expired() {
1109 return Ok(PlannerSupportOutcome::Fallback {
1110 reason: "planner support was cancelled before dispatch".to_owned(),
1111 });
1112 }
1113
1114 let predicate = encode_normalized_predicate(
1115 target_function_id,
1116 operator_class_id,
1117 indexed,
1118 arguments,
1119 &function.arguments,
1120 )?;
1121 let mut state = CallState::new(
1122 support.max_output_bytes,
1123 support.max_spans.saturating_add(1),
1124 )
1125 .with_limits(limits);
1126 let diagnostics = state.diagnostic_sink();
1127 let context = state.context(&diagnostics);
1128 let output = state.result_builder();
1129 let input = abi::RadixAbiSliceV1 {
1130 ptr: predicate.as_ptr(),
1131 len: predicate.len() as u32,
1132 reserved: 0,
1133 };
1134 let status = unsafe { (support.callback)(&context, input, &output) };
1137 if status != abi::RADIX_STATUS_OK {
1138 if matches!(
1139 status,
1140 abi::RADIX_STATUS_INVALID_ARGUMENT
1141 | abi::RADIX_STATUS_DOMAIN_ERROR
1142 | abi::RADIX_STATUS_LIMIT_EXCEEDED
1143 | abi::RADIX_STATUS_CANCELLED
1144 ) {
1145 return Ok(PlannerSupportOutcome::Fallback {
1146 reason: state.callback_error(status).to_string(),
1147 });
1148 }
1149 return Err(state.callback_error(status));
1150 }
1151 if !state.finished || state.staged.len() != state.staged_nulls.len() {
1152 return Err(PluginInvocationError::Contract(
1153 "planner support did not finish its candidate plan",
1154 ));
1155 }
1156 if state.staged_nulls.iter().any(|value| *value) {
1157 return Err(PluginInvocationError::Contract(
1158 "planner support emitted a NULL plan item",
1159 ));
1160 }
1161
1162 let mut estimate = None;
1163 let mut spans = Vec::new();
1164 for item in state.staged {
1165 match item.first().copied() {
1166 Some(1) => {
1167 if item.len() < 12 || item[..4] != [1, 0, 0, 0] {
1168 return Err(PluginInvocationError::Contract(
1169 "planner support emitted a malformed span header",
1170 ));
1171 }
1172 let start_len =
1173 u32::from_le_bytes(item[4..8].try_into().expect("fixed width")) as usize;
1174 let end_len =
1175 u32::from_le_bytes(item[8..12].try_into().expect("fixed width")) as usize;
1176 let split =
1177 12usize
1178 .checked_add(start_len)
1179 .ok_or(PluginInvocationError::Contract(
1180 "planner support span length overflow",
1181 ))?;
1182 let end = split
1183 .checked_add(end_len)
1184 .ok_or(PluginInvocationError::Contract(
1185 "planner support span length overflow",
1186 ))?;
1187 if end != item.len() {
1188 return Err(PluginInvocationError::Contract(
1189 "planner support span lengths are invalid",
1190 ));
1191 }
1192 let start = decode_function_result(
1193 self,
1194 operator_class.key_type,
1195 false,
1196 item[12..split].to_vec(),
1197 )?;
1198 let end_value = decode_function_result(
1199 self,
1200 operator_class.key_type,
1201 false,
1202 item[split..end].to_vec(),
1203 )?;
1204 if start > end_value {
1205 return Err(PluginInvocationError::Contract(
1206 "planner support span start exceeds its end",
1207 ));
1208 }
1209 spans.push(CandidateKeySpan {
1210 start,
1211 end: end_value,
1212 });
1213 }
1214 Some(2) => {
1215 if item.len() != 20 || item[..4] != [2, 0, 0, 0] || estimate.is_some() {
1216 return Err(PluginInvocationError::Contract(
1217 "planner support emitted malformed or duplicate estimate metadata",
1218 ));
1219 }
1220 let rows = u64::from_le_bytes(item[4..12].try_into().expect("fixed width"));
1221 let cost = u32::from_le_bytes(item[12..16].try_into().expect("fixed width"));
1222 if item[16..20] != [0, 0, 0, 0] || cost > 1_000_000 {
1223 return Err(PluginInvocationError::Contract(
1224 "planner support estimate metadata is outside its bounds",
1225 ));
1226 }
1227 estimate = Some((rows, cost));
1228 }
1229 _ => {
1230 return Err(PluginInvocationError::Contract(
1231 "planner support emitted an unknown plan item",
1232 ))
1233 }
1234 }
1235 }
1236 let Some((estimated_rows, cost_hint)) = estimate else {
1237 return Ok(PlannerSupportOutcome::Fallback {
1238 reason: "planner support returned an unknown estimate".to_owned(),
1239 });
1240 };
1241 if spans.len() > support.max_spans as usize {
1242 return Err(PluginInvocationError::Contract(
1243 "planner support exceeded its admitted span bound",
1244 ));
1245 }
1246 spans.sort_by(|left, right| {
1247 left.start
1248 .cmp(&right.start)
1249 .then_with(|| left.end.cmp(&right.end))
1250 });
1251 let mut canonical: Vec<CandidateKeySpan> = Vec::with_capacity(spans.len());
1252 for span in spans {
1253 if let Some(previous) = canonical.last_mut() {
1254 if span.start <= previous.end {
1255 if span.end > previous.end {
1256 previous.end = span.end;
1257 }
1258 continue;
1259 }
1260 }
1261 canonical.push(span);
1262 }
1263 Ok(PlannerSupportOutcome::Plan(CandidatePlan {
1264 spans: canonical,
1265 requires_recheck: support.recheck_policy == abi::RADIX_RECHECK_ALWAYS,
1266 estimated_rows,
1267 cost_hint,
1268 identity: PlannerPlanIdentity {
1269 registry_generation: self.generation(),
1270 support_id,
1271 support_semantic_revision: support.semantic_revision,
1272 support_fingerprint: support.fingerprint,
1273 target_function_id,
1274 function_semantic_revision: function.semantic_revision,
1275 operator_class_id,
1276 operator_class_semantic_revision: operator_class.semantic_revision,
1277 key_codec_revision: operator_class.key_codec_revision,
1278 operator_class_fingerprint: operator_class.fingerprint,
1279 },
1280 }))
1281 }
1282
1283 pub fn encode_operator_class_key(
1289 &self,
1290 object_id: ObjectId,
1291 value: &Value,
1292 limits: InvocationLimits,
1293 ) -> Result<Value, PluginInvocationError> {
1294 let operator_class = self
1295 .operator_class(&object_id)
1296 .ok_or(PluginInvocationError::UnknownOperatorClass(object_id))?;
1297 let mut backing = Vec::with_capacity(1);
1298 let raw = raw_argument(value, operator_class.input_type, &mut backing)?;
1299 if operator_class.access_method == abi::RADIX_ACCESS_METHOD_HASH {
1300 if operator_class.key_type != RegisteredTypeRef::Builtin(abi::RADIX_BUILTIN_BYTES) {
1301 return Err(PluginInvocationError::Contract(
1302 "hash operator class physical key type is not BYTES",
1303 ));
1304 }
1305 let components = self.external_hash_components(value)?;
1306 let mut bytes = Vec::new();
1307 bytes.extend_from_slice(&(components.len() as u32).to_le_bytes());
1308 for component in components {
1309 bytes.extend_from_slice(&component.kind.to_le_bytes());
1310 bytes.extend_from_slice(&(component.bytes.len() as u32).to_le_bytes());
1311 bytes.extend_from_slice(&component.bytes);
1312 }
1313 return Ok(Value::bytes(bytes));
1314 }
1315 if limits.cancelled_or_expired() {
1316 return Err(PluginInvocationError::Callback {
1317 status: abi::RADIX_STATUS_CANCELLED,
1318 diagnostic: "operator-class key encoding was cancelled before dispatch".to_owned(),
1319 });
1320 }
1321 let mut state = CallState::new(abi::RADIX_MAX_EXTERNAL_VALUE_BYTES, 1).with_limits(limits);
1322 let diagnostics = state.diagnostic_sink();
1323 let context = state.context(&diagnostics);
1324 let output = state.result_builder();
1325 let status = unsafe { (operator_class.encode_key)(&context, &raw, &output) };
1328 let bytes = finish_single_output(state, status)?;
1329 decode_function_result(self, operator_class.key_type, false, bytes)
1330 }
1331
1332 pub fn invoke_scalar_function(
1333 &self,
1334 object_id: ObjectId,
1335 arguments: &[Value],
1336 limits: InvocationLimits,
1337 ) -> Result<Value, PluginInvocationError> {
1338 let function = self
1339 .function(&object_id)
1340 .ok_or(PluginInvocationError::UnknownFunction(object_id))?;
1341 if arguments.len() != function.arguments.len() {
1342 return Err(PluginInvocationError::Contract(
1343 "native function argument count differs from descriptor",
1344 ));
1345 }
1346 if function.strict && arguments.iter().any(Value::is_null) {
1347 return decode_function_result(self, function.result, true, Vec::new());
1348 }
1349 if limits.cancelled_or_expired() {
1350 return Err(PluginInvocationError::Callback {
1351 status: abi::RADIX_STATUS_CANCELLED,
1352 diagnostic: "native function call was cancelled before dispatch".to_owned(),
1353 });
1354 }
1355 let mut backing = Vec::with_capacity(arguments.len());
1356 let raw = arguments
1357 .iter()
1358 .zip(&function.arguments)
1359 .map(|(value, expected)| raw_argument(value, *expected, &mut backing))
1360 .collect::<Result<Vec<_>, _>>()?;
1361 let mut state = CallState::new(function.max_output_bytes, 1).with_limits(limits);
1362 let diagnostics = state.diagnostic_sink();
1363 let context = state.context(&diagnostics);
1364 let output = state.result_builder();
1365 let status =
1368 unsafe { (function.scalar)(&context, raw.as_ptr(), raw.len() as u32, &output) };
1369 finish_function_output(self, function, state, status)
1370 }
1371
1372 pub fn invoke_function_batch(
1373 &self,
1374 object_id: ObjectId,
1375 rows: &[Vec<Value>],
1376 limits: InvocationLimits,
1377 ) -> Result<Vec<Value>, PluginInvocationError> {
1378 let function = self
1379 .function(&object_id)
1380 .ok_or(PluginInvocationError::UnknownFunction(object_id))?;
1381 if rows.len() > u32::MAX as usize
1382 || rows.iter().any(|row| row.len() != function.arguments.len())
1383 {
1384 return Err(PluginInvocationError::Contract(
1385 "native function batch shape differs from descriptor",
1386 ));
1387 }
1388 if rows.is_empty() {
1389 return Ok(Vec::new());
1390 }
1391 let Some(batch_callback) = function
1395 .batch
1396 .filter(|_| !function.strict || !rows.iter().flatten().any(Value::is_null))
1397 else {
1398 return rows
1399 .iter()
1400 .map(|row| self.invoke_scalar_function(object_id, row, limits))
1401 .collect();
1402 };
1403 if limits.cancelled_or_expired() {
1404 return Err(PluginInvocationError::Callback {
1405 status: abi::RADIX_STATUS_CANCELLED,
1406 diagnostic: "native function batch was cancelled before dispatch".to_owned(),
1407 });
1408 }
1409 let columns = function
1410 .arguments
1411 .iter()
1412 .enumerate()
1413 .map(|(index, expected)| {
1414 let values = rows.iter().map(|row| &row[index]).collect::<Vec<_>>();
1415 OwnedBatchColumn::new(self, *expected, &values)
1416 })
1417 .collect::<Result<Vec<_>, _>>()?;
1418 let row_count = rows.len() as u32;
1419 let raw_columns = columns
1420 .iter()
1421 .map(|column| column.as_abi(row_count))
1422 .collect::<Vec<_>>();
1423 let batch = abi::RadixAbiBatchViewV1 {
1424 header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiBatchViewV1>(0),
1425 row_count,
1426 column_count: raw_columns.len() as u32,
1427 columns: if raw_columns.is_empty() {
1428 std::ptr::null()
1429 } else {
1430 raw_columns.as_ptr()
1431 },
1432 };
1433 unsafe { abi::validate_batch_columns(&batch) }.map_err(|_| {
1436 PluginInvocationError::Contract("host constructed an invalid native batch")
1437 })?;
1438 let maximum = function.max_output_bytes.saturating_mul(row_count);
1439 let mut state = CallState::new(maximum, row_count).with_limits(limits);
1440 let diagnostics = state.diagnostic_sink();
1441 let context = state.context(&diagnostics);
1442 let output = state.result_builder();
1443 let status = unsafe { batch_callback(&context, &batch, &output) };
1446 finish_batch_output(self, function, state, status, rows.len())
1447 }
1448
1449 pub fn validate_external_value(&self, value: &Value) -> Result<(), PluginInvocationError> {
1450 let (registered, view) = external(self, value)?;
1451 let output = run_parse(registered.decode, view.payload(), registered.max_bytes)?;
1452 if output != view.payload() {
1453 return Err(PluginInvocationError::Contract(
1454 "canonical codec decode/encode did not preserve payload bytes",
1455 ));
1456 }
1457 Ok(())
1458 }
1459
1460 pub fn parse_external_text(
1461 &self,
1462 type_ref: ExternalTypeRef,
1463 text: &str,
1464 ) -> Result<Value, PluginInvocationError> {
1465 let registered = self.external_type(&type_ref.type_object_id()).ok_or(
1466 PluginInvocationError::UnknownType(type_ref.type_object_id()),
1467 )?;
1468 if registered.codec_version != type_ref.codec_version() {
1469 return Err(PluginInvocationError::CodecVersion {
1470 expected: registered.codec_version,
1471 actual: type_ref.codec_version(),
1472 });
1473 }
1474 let callback = registered
1475 .text_input
1476 .ok_or(PluginInvocationError::MissingCapability("text input"))?;
1477 let payload = run_parse(callback, text.as_bytes(), registered.max_bytes)?;
1478 let value = Value::try_external(type_ref, payload)
1479 .map_err(|_| PluginInvocationError::PayloadLimit)?;
1480 self.validate_external_value(&value)?;
1481 Ok(value)
1482 }
1483
1484 pub fn format_external_text(&self, value: &Value) -> Result<String, PluginInvocationError> {
1485 let (registered, view) = external(self, value)?;
1486 let callback = registered
1487 .text_output
1488 .ok_or(PluginInvocationError::MissingCapability("text output"))?;
1489 let bytes = run_codec(
1490 callback,
1491 &raw_value(view),
1492 abi::RADIX_MAX_EXTERNAL_VALUE_BYTES,
1493 )?;
1494 String::from_utf8(bytes)
1495 .map_err(|_| PluginInvocationError::Contract("text output is not UTF-8"))
1496 }
1497
1498 pub fn parse_external_binary(
1499 &self,
1500 type_ref: ExternalTypeRef,
1501 bytes: &[u8],
1502 ) -> Result<Value, PluginInvocationError> {
1503 let registered = self.external_type(&type_ref.type_object_id()).ok_or(
1504 PluginInvocationError::UnknownType(type_ref.type_object_id()),
1505 )?;
1506 if registered.codec_version != type_ref.codec_version() {
1507 return Err(PluginInvocationError::CodecVersion {
1508 expected: registered.codec_version,
1509 actual: type_ref.codec_version(),
1510 });
1511 }
1512 let callback = registered.binary_input.unwrap_or(registered.decode);
1513 let payload = run_parse(callback, bytes, registered.max_bytes)?;
1514 let value = Value::try_external(type_ref, payload)
1515 .map_err(|_| PluginInvocationError::PayloadLimit)?;
1516 self.validate_external_value(&value)?;
1517 Ok(value)
1518 }
1519
1520 pub fn format_external_binary(&self, value: &Value) -> Result<Vec<u8>, PluginInvocationError> {
1521 let (registered, view) = external(self, value)?;
1522 let callback = registered.binary_output.unwrap_or(registered.encode);
1523 run_codec(callback, &raw_value(view), registered.max_bytes)
1524 }
1525
1526 pub fn external_equal(
1527 &self,
1528 left: &Value,
1529 right: &Value,
1530 ) -> Result<bool, PluginInvocationError> {
1531 let (registered, left, right) = matching_pair(self, left, right)?;
1532 let callback = registered
1533 .equality
1534 .ok_or(PluginInvocationError::MissingCapability("equality"))?;
1535 let mut state = CallState::new(1, 1);
1536 let diagnostics = state.diagnostic_sink();
1537 let context = state.context(&diagnostics);
1538 let mut output = u8::MAX;
1539 let status = unsafe { callback(&context, &left, &right, &mut output) };
1541 if status != abi::RADIX_STATUS_OK {
1542 return Err(state.callback_error(status));
1543 }
1544 match output {
1545 0 => Ok(false),
1546 1 => Ok(true),
1547 _ => Err(PluginInvocationError::Contract(
1548 "equality callback returned a non-boolean byte",
1549 )),
1550 }
1551 }
1552
1553 pub fn external_compare(
1554 &self,
1555 left: &Value,
1556 right: &Value,
1557 ) -> Result<Ordering, PluginInvocationError> {
1558 let (registered, left, right) = matching_pair(self, left, right)?;
1559 let callback = registered
1560 .ordering
1561 .ok_or(PluginInvocationError::MissingCapability("ordering"))?;
1562 let mut state = CallState::new(1, 1);
1563 let diagnostics = state.diagnostic_sink();
1564 let context = state.context(&diagnostics);
1565 let mut output = i8::MIN;
1566 let status = unsafe { callback(&context, &left, &right, &mut output) };
1568 if status != abi::RADIX_STATUS_OK {
1569 return Err(state.callback_error(status));
1570 }
1571 match output {
1572 -1 => Ok(Ordering::Less),
1573 0 => Ok(Ordering::Equal),
1574 1 => Ok(Ordering::Greater),
1575 _ => Err(PluginInvocationError::Contract(
1576 "ordering callback returned a value outside -1..=1",
1577 )),
1578 }
1579 }
1580
1581 pub fn external_hash_components(
1582 &self,
1583 value: &Value,
1584 ) -> Result<Vec<HashComponent>, PluginInvocationError> {
1585 let (registered, view) = external(self, value)?;
1586 let callback = registered
1587 .hash
1588 .ok_or(PluginInvocationError::MissingCapability("hash"))?;
1589 let raw = raw_value(view);
1590 let mut state = CallState::new(1, 1);
1591 let diagnostics = state.diagnostic_sink();
1592 let context = state.context(&diagnostics);
1593 let sink = state.hash_sink();
1594 let status = unsafe { callback(&context, &raw, &sink) };
1596 if status != abi::RADIX_STATUS_OK {
1597 return Err(state.callback_error(status));
1598 }
1599 if state.hash_components.is_empty() {
1600 return Err(PluginInvocationError::Contract(
1601 "hash callback emitted no semantic components",
1602 ));
1603 }
1604 Ok(state.hash_components)
1605 }
1606}
1607
1608#[cfg(test)]
1609mod tests {
1610 use super::*;
1611 use crate::{
1612 RegisteredExternalType, RegisteredFunction, RegisteredOperatorClass, RegisteredPackage,
1613 RegisteredPlannerSupport,
1614 };
1615 use std::ptr;
1616
1617 unsafe extern "C" fn parse_echo(
1618 _context: *const abi::RadixAbiCallContextV1,
1619 input: abi::RadixAbiSliceV1,
1620 output: *const abi::RadixAbiResultBuilderV1,
1621 ) -> abi::RadixAbiStatusV1 {
1622 let Some(output) = (unsafe { output.as_ref() }) else {
1623 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1624 };
1625 let Some(write) = output.write else {
1626 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1627 };
1628 let Some(finish) = output.finish else {
1629 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1630 };
1631 let status = unsafe { write(output.handle, 0, 0, input) };
1632 if status != abi::RADIX_STATUS_OK {
1633 return status;
1634 }
1635 unsafe { finish(output.handle) }
1636 }
1637
1638 unsafe extern "C" fn codec_echo(
1639 _context: *const abi::RadixAbiCallContextV1,
1640 input: *const abi::RadixAbiValueV1,
1641 output: *const abi::RadixAbiResultBuilderV1,
1642 ) -> abi::RadixAbiStatusV1 {
1643 let Some(input) = (unsafe { input.as_ref() }) else {
1644 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1645 };
1646 unsafe { parse_echo(ptr::null(), input.borrowed_bytes, output) }
1647 }
1648
1649 unsafe extern "C" fn equal(
1650 _context: *const abi::RadixAbiCallContextV1,
1651 left: *const abi::RadixAbiValueV1,
1652 right: *const abi::RadixAbiValueV1,
1653 output: *mut u8,
1654 ) -> abi::RadixAbiStatusV1 {
1655 let (Some(left), Some(right), Some(output)) = (
1656 unsafe { left.as_ref() },
1657 unsafe { right.as_ref() },
1658 unsafe { output.as_mut() },
1659 ) else {
1660 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1661 };
1662 let left = unsafe {
1663 slice::from_raw_parts(left.borrowed_bytes.ptr, left.borrowed_bytes.len as usize)
1664 };
1665 let right = unsafe {
1666 slice::from_raw_parts(right.borrowed_bytes.ptr, right.borrowed_bytes.len as usize)
1667 };
1668 *output = u8::from(left == right);
1669 abi::RADIX_STATUS_OK
1670 }
1671
1672 unsafe extern "C" fn compare(
1673 _context: *const abi::RadixAbiCallContextV1,
1674 left: *const abi::RadixAbiValueV1,
1675 right: *const abi::RadixAbiValueV1,
1676 output: *mut i8,
1677 ) -> abi::RadixAbiStatusV1 {
1678 let (Some(left), Some(right), Some(output)) = (
1679 unsafe { left.as_ref() },
1680 unsafe { right.as_ref() },
1681 unsafe { output.as_mut() },
1682 ) else {
1683 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1684 };
1685 let left = unsafe {
1686 slice::from_raw_parts(left.borrowed_bytes.ptr, left.borrowed_bytes.len as usize)
1687 };
1688 let right = unsafe {
1689 slice::from_raw_parts(right.borrowed_bytes.ptr, right.borrowed_bytes.len as usize)
1690 };
1691 *output = match left.cmp(right) {
1692 Ordering::Less => -1,
1693 Ordering::Equal => 0,
1694 Ordering::Greater => 1,
1695 };
1696 abi::RADIX_STATUS_OK
1697 }
1698
1699 unsafe extern "C" fn hash(
1700 _context: *const abi::RadixAbiCallContextV1,
1701 value: *const abi::RadixAbiValueV1,
1702 sink: *const abi::RadixAbiHashSinkV1,
1703 ) -> abi::RadixAbiStatusV1 {
1704 let (Some(value), Some(sink)) = (unsafe { value.as_ref() }, unsafe { sink.as_ref() })
1705 else {
1706 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1707 };
1708 let Some(append) = sink.append else {
1709 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1710 };
1711 unsafe {
1712 append(
1713 sink.handle,
1714 abi::RADIX_HASH_COMPONENT_BYTES,
1715 0,
1716 value.borrowed_bytes,
1717 )
1718 }
1719 }
1720
1721 unsafe extern "C" fn increment_scalar(
1722 context: *const abi::RadixAbiCallContextV1,
1723 arguments: *const abi::RadixAbiValueV1,
1724 argument_count: u32,
1725 output: *const abi::RadixAbiResultBuilderV1,
1726 ) -> abi::RadixAbiStatusV1 {
1727 let (Some(context), Some(output)) =
1728 (unsafe { context.as_ref() }, unsafe { output.as_ref() })
1729 else {
1730 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1731 };
1732 if argument_count != 1 || arguments.is_null() {
1733 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1734 }
1735 if let Some(check) = context.check_cancelled {
1736 let status = unsafe { check(context.handle) };
1737 if status != abi::RADIX_STATUS_OK {
1738 return status;
1739 }
1740 }
1741 let argument = unsafe { &*arguments };
1742 let value = i64::from_le_bytes(argument.inline_bytes[..8].try_into().unwrap()) + 1;
1743 let bytes = value.to_le_bytes();
1744 let Some(write) = output.write else {
1745 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1746 };
1747 let status = unsafe {
1748 write(
1749 output.handle,
1750 0,
1751 0,
1752 abi::RadixAbiSliceV1 {
1753 ptr: bytes.as_ptr(),
1754 len: bytes.len() as u32,
1755 reserved: 0,
1756 },
1757 )
1758 };
1759 if status != abi::RADIX_STATUS_OK {
1760 return status;
1761 }
1762 unsafe { output.finish.unwrap()(output.handle) }
1763 }
1764
1765 unsafe extern "C" fn unused_key_encoder(
1766 _context: *const abi::RadixAbiCallContextV1,
1767 _value: *const abi::RadixAbiValueV1,
1768 _output: *const abi::RadixAbiResultBuilderV1,
1769 ) -> abi::RadixAbiStatusV1 {
1770 abi::RADIX_STATUS_INTERNAL_ERROR
1771 }
1772
1773 unsafe fn write_plan_item(
1774 output: &abi::RadixAbiResultBuilderV1,
1775 bytes: &[u8],
1776 ) -> abi::RadixAbiStatusV1 {
1777 unsafe {
1778 output.write.unwrap()(
1779 output.handle,
1780 0,
1781 0,
1782 abi::RadixAbiSliceV1 {
1783 ptr: bytes.as_ptr(),
1784 len: bytes.len() as u32,
1785 reserved: 0,
1786 },
1787 )
1788 }
1789 }
1790
1791 unsafe fn write_integer_span(
1792 output: &abi::RadixAbiResultBuilderV1,
1793 start: i64,
1794 end: i64,
1795 ) -> abi::RadixAbiStatusV1 {
1796 let mut span = Vec::with_capacity(28);
1797 span.extend_from_slice(&[1, 0, 0, 0]);
1798 span.extend_from_slice(&8_u32.to_le_bytes());
1799 span.extend_from_slice(&8_u32.to_le_bytes());
1800 span.extend_from_slice(&start.to_le_bytes());
1801 span.extend_from_slice(&end.to_le_bytes());
1802 unsafe { write_plan_item(output, &span) }
1803 }
1804
1805 unsafe fn write_estimate(
1806 output: &abi::RadixAbiResultBuilderV1,
1807 rows: u64,
1808 ) -> abi::RadixAbiStatusV1 {
1809 let mut estimate = Vec::with_capacity(20);
1810 estimate.extend_from_slice(&[2, 0, 0, 0]);
1811 estimate.extend_from_slice(&rows.to_le_bytes());
1812 estimate.extend_from_slice(&7_u32.to_le_bytes());
1813 estimate.extend_from_slice(&0_u32.to_le_bytes());
1814 unsafe { write_plan_item(output, &estimate) }
1815 }
1816
1817 unsafe extern "C" fn overlapping_plan(
1818 _context: *const abi::RadixAbiCallContextV1,
1819 _predicate: abi::RadixAbiSliceV1,
1820 output: *const abi::RadixAbiResultBuilderV1,
1821 ) -> abi::RadixAbiStatusV1 {
1822 let Some(output) = (unsafe { output.as_ref() }) else {
1823 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1824 };
1825 for (start, end) in [(3, 7), (1, 5), (8, 9)] {
1826 let status = unsafe { write_integer_span(output, start, end) };
1827 if status != abi::RADIX_STATUS_OK {
1828 return status;
1829 }
1830 }
1831 let status = unsafe { write_estimate(output, 4) };
1832 if status != abi::RADIX_STATUS_OK {
1833 return status;
1834 }
1835 unsafe { output.finish.unwrap()(output.handle) }
1836 }
1837
1838 unsafe extern "C" fn unknown_estimate_plan(
1839 _context: *const abi::RadixAbiCallContextV1,
1840 _predicate: abi::RadixAbiSliceV1,
1841 output: *const abi::RadixAbiResultBuilderV1,
1842 ) -> abi::RadixAbiStatusV1 {
1843 let Some(output) = (unsafe { output.as_ref() }) else {
1844 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1845 };
1846 let status = unsafe { write_integer_span(output, 1, 2) };
1847 if status != abi::RADIX_STATUS_OK {
1848 return status;
1849 }
1850 unsafe { output.finish.unwrap()(output.handle) }
1851 }
1852
1853 unsafe extern "C" fn malformed_plan(
1854 _context: *const abi::RadixAbiCallContextV1,
1855 _predicate: abi::RadixAbiSliceV1,
1856 output: *const abi::RadixAbiResultBuilderV1,
1857 ) -> abi::RadixAbiStatusV1 {
1858 let Some(output) = (unsafe { output.as_ref() }) else {
1859 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1860 };
1861 let status = unsafe { write_plan_item(output, &[99, 0, 0, 0]) };
1862 if status != abi::RADIX_STATUS_OK {
1863 return status;
1864 }
1865 unsafe { output.finish.unwrap()(output.handle) }
1866 }
1867
1868 unsafe extern "C" fn exploding_plan(
1869 _context: *const abi::RadixAbiCallContextV1,
1870 _predicate: abi::RadixAbiSliceV1,
1871 output: *const abi::RadixAbiResultBuilderV1,
1872 ) -> abi::RadixAbiStatusV1 {
1873 let Some(output) = (unsafe { output.as_ref() }) else {
1874 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1875 };
1876 for value in 0..=2 {
1877 let status = unsafe { write_integer_span(output, value, value) };
1878 if status != abi::RADIX_STATUS_OK {
1879 return status;
1880 }
1881 }
1882 unsafe { output.finish.unwrap()(output.handle) }
1883 }
1884
1885 unsafe extern "C" fn unsupported_plan(
1886 _context: *const abi::RadixAbiCallContextV1,
1887 _predicate: abi::RadixAbiSliceV1,
1888 _output: *const abi::RadixAbiResultBuilderV1,
1889 ) -> abi::RadixAbiStatusV1 {
1890 abi::RADIX_STATUS_DOMAIN_ERROR
1891 }
1892
1893 fn planner_registry(
1894 callback: abi::RadixAbiPlannerSupportFnV1,
1895 max_spans: u32,
1896 ) -> (PluginRegistry, ObjectId) {
1897 let package_id = [0x71; 16];
1898 let function_id = [0x72; 16];
1899 let class_id = [0x73; 16];
1900 let support_id = [0x74; 16];
1901 let integer = RegisteredTypeRef::Builtin(abi::RADIX_BUILTIN_INTEGER);
1902 let function = RegisteredFunction {
1903 package_id,
1904 object_id: function_id,
1905 local_id: "predicate".to_owned(),
1906 display_name: "predicate".to_owned(),
1907 semantic_revision: 2,
1908 arguments: vec![integer, integer],
1909 result: RegisteredTypeRef::Builtin(abi::RADIX_BUILTIN_BOOLEAN),
1910 volatility: abi::RADIX_VOLATILITY_IMMUTABLE,
1911 cancellation: abi::RADIX_CANCELLATION_BOUNDED,
1912 strict: true,
1913 parallel_safe: true,
1914 cost: 1,
1915 max_output_bytes: 1,
1916 scalar: increment_scalar,
1917 batch: None,
1918 };
1919 let operator_class = RegisteredOperatorClass {
1920 package_id,
1921 object_id: class_id,
1922 local_id: "integer_btree".to_owned(),
1923 semantic_revision: 3,
1924 access_method: abi::RADIX_ACCESS_METHOD_BTREE,
1925 input_type: integer,
1926 key_type: integer,
1927 key_codec_revision: 4,
1928 strategies: vec![],
1929 supports: vec![],
1930 fingerprint: [0x75; 32],
1931 encode_key: unused_key_encoder,
1932 };
1933 let support = RegisteredPlannerSupport {
1934 package_id,
1935 object_id: support_id,
1936 local_id: "predicate_support".to_owned(),
1937 semantic_revision: 5,
1938 max_spans,
1939 max_output_bytes: 4096,
1940 recheck_policy: abi::RADIX_RECHECK_ALWAYS,
1941 target_function_id: Some(function_id),
1942 target_operator_class_id: Some(class_id),
1943 fingerprint: [0x76; 32],
1944 callback,
1945 };
1946 let package = RegisteredPackage::for_test(package_id, "planner", "1.0.0", [0x77; 32]);
1947 (
1948 PluginRegistry::from_test_objects(
1949 [package],
1950 [],
1951 [function],
1952 [],
1953 [operator_class],
1954 [support],
1955 ),
1956 support_id,
1957 )
1958 }
1959
1960 unsafe extern "C" fn increment_batch(
1961 _context: *const abi::RadixAbiCallContextV1,
1962 input: *const abi::RadixAbiBatchViewV1,
1963 output: *const abi::RadixAbiResultBuilderV1,
1964 ) -> abi::RadixAbiStatusV1 {
1965 let (Some(input), Some(output)) = (unsafe { input.as_ref() }, unsafe { output.as_ref() })
1966 else {
1967 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1968 };
1969 if input.column_count != 1 || input.columns.is_null() {
1970 return abi::RADIX_STATUS_INVALID_ARGUMENT;
1971 }
1972 let column = unsafe { &*input.columns };
1973 let data = unsafe { slice::from_raw_parts(column.data.ptr, column.data.len as usize) };
1974 for row in 0..input.row_count as usize {
1975 let start = row * column.stride as usize;
1976 let value = i64::from_le_bytes(data[start..start + 8].try_into().unwrap()) + 1;
1977 let bytes = value.to_le_bytes();
1978 let status = unsafe {
1979 output.write.unwrap()(
1980 output.handle,
1981 0,
1982 0,
1983 abi::RadixAbiSliceV1 {
1984 ptr: bytes.as_ptr(),
1985 len: bytes.len() as u32,
1986 reserved: 0,
1987 },
1988 )
1989 };
1990 if status != abi::RADIX_STATUS_OK {
1991 return status;
1992 }
1993 }
1994 unsafe { output.finish.unwrap()(output.handle) }
1995 }
1996
1997 unsafe extern "C" fn partial_batch_error(
1998 _context: *const abi::RadixAbiCallContextV1,
1999 _input: *const abi::RadixAbiBatchViewV1,
2000 output: *const abi::RadixAbiResultBuilderV1,
2001 ) -> abi::RadixAbiStatusV1 {
2002 let Some(output) = (unsafe { output.as_ref() }) else {
2003 return abi::RADIX_STATUS_INVALID_ARGUMENT;
2004 };
2005 let bytes = 42_i64.to_le_bytes();
2006 let _ = unsafe {
2007 output.write.unwrap()(
2008 output.handle,
2009 0,
2010 0,
2011 abi::RadixAbiSliceV1 {
2012 ptr: bytes.as_ptr(),
2013 len: bytes.len() as u32,
2014 reserved: 0,
2015 },
2016 )
2017 };
2018 abi::RADIX_STATUS_INTERNAL_ERROR
2019 }
2020
2021 fn registry() -> (PluginRegistry, ExternalTypeRef) {
2022 let package_id = [0x51; 16];
2023 let object_id = [0x52; 16];
2024 let external = RegisteredExternalType {
2025 package_id,
2026 object_id,
2027 local_id: "sample".to_owned(),
2028 display_name: "sample".to_owned(),
2029 codec_version: 3,
2030 semantic_revision: 1,
2031 storage_kind: abi::RADIX_EXTERNAL_STORAGE_VARIABLE,
2032 fixed_bytes: 0,
2033 max_bytes: 1024,
2034 capabilities: abi::RADIX_TYPE_CAP_EQUALITY
2035 | abi::RADIX_TYPE_CAP_HASH
2036 | abi::RADIX_TYPE_CAP_ORDERING
2037 | abi::RADIX_TYPE_CAP_TEXT_INPUT
2038 | abi::RADIX_TYPE_CAP_TEXT_OUTPUT
2039 | abi::RADIX_TYPE_CAP_BINARY_INPUT
2040 | abi::RADIX_TYPE_CAP_BINARY_OUTPUT,
2041 codec_fingerprint: [7; 32],
2042 encode: codec_echo,
2043 decode: parse_echo,
2044 equality: Some(equal),
2045 hash: Some(hash),
2046 ordering: Some(compare),
2047 text_input: Some(parse_echo),
2048 text_output: Some(codec_echo),
2049 binary_input: Some(parse_echo),
2050 binary_output: Some(codec_echo),
2051 };
2052 let package = RegisteredPackage::for_test(package_id, "fixture", "1.0.0", [8; 32]);
2053 (
2054 PluginRegistry::from_test_objects([package], [external], [], [], [], []),
2055 ExternalTypeRef::new(object_id, 3).unwrap(),
2056 )
2057 }
2058
2059 fn function_registry(
2060 strict: bool,
2061 batch: Option<abi::RadixAbiBatchFnV1>,
2062 ) -> (PluginRegistry, ObjectId) {
2063 let package_id = [0x61; 16];
2064 let object_id = [0x62; 16];
2065 let function = RegisteredFunction {
2066 package_id,
2067 object_id,
2068 local_id: "increment".to_owned(),
2069 display_name: "increment".to_owned(),
2070 semantic_revision: 1,
2071 arguments: vec![RegisteredTypeRef::Builtin(abi::RADIX_BUILTIN_INTEGER)],
2072 result: RegisteredTypeRef::Builtin(abi::RADIX_BUILTIN_INTEGER),
2073 volatility: abi::RADIX_VOLATILITY_IMMUTABLE,
2074 cancellation: abi::RADIX_CANCELLATION_BOUNDED,
2075 strict,
2076 parallel_safe: true,
2077 cost: 1,
2078 max_output_bytes: 8,
2079 scalar: increment_scalar,
2080 batch,
2081 };
2082 let package = RegisteredPackage::for_test(package_id, "functions", "1.0.0", [9; 32]);
2083 (
2084 PluginRegistry::from_test_objects([package], [], [function], [], [], []),
2085 object_id,
2086 )
2087 }
2088
2089 #[test]
2090 fn external_callbacks_own_semantics_and_io() {
2091 let (registry, type_ref) = registry();
2092 let left = registry.parse_external_text(type_ref, "alpha").unwrap();
2093 let same = registry.parse_external_binary(type_ref, b"alpha").unwrap();
2094 let greater = Value::try_external(type_ref, b"beta").unwrap();
2095
2096 registry.validate_external_value(&left).unwrap();
2097 assert_eq!(registry.format_external_text(&left).unwrap(), "alpha");
2098 assert_eq!(registry.format_external_binary(&left).unwrap(), b"alpha");
2099 assert!(registry.external_equal(&left, &same).unwrap());
2100 assert_eq!(
2101 registry.external_compare(&left, &greater).unwrap(),
2102 Ordering::Less
2103 );
2104 assert_eq!(
2105 registry.external_hash_components(&left).unwrap(),
2106 vec![HashComponent {
2107 kind: abi::RADIX_HASH_COMPONENT_BYTES,
2108 bytes: b"alpha".to_vec(),
2109 }]
2110 );
2111 }
2112
2113 #[test]
2114 fn external_callbacks_fail_closed_on_identity_and_codec_mismatch() {
2115 let (registry, type_ref) = registry();
2116 let valid = Value::try_external(type_ref, b"alpha").unwrap();
2117 let other =
2118 Value::try_external(ExternalTypeRef::new([0x53; 16], 3).unwrap(), b"alpha").unwrap();
2119 let stale =
2120 Value::try_external(ExternalTypeRef::new([0x52; 16], 2).unwrap(), b"alpha").unwrap();
2121
2122 assert!(matches!(
2123 registry.external_equal(&valid, &other),
2124 Err(PluginInvocationError::UnknownType(_))
2125 ));
2126 assert!(matches!(
2127 registry.validate_external_value(&stale),
2128 Err(PluginInvocationError::CodecVersion { .. })
2129 ));
2130 }
2131
2132 #[test]
2133 fn scalar_and_batch_function_paths_have_identical_results() {
2134 let (registry, object_id) = function_registry(false, Some(increment_batch));
2135 let rows = vec![
2136 vec![Value::Integer(1)],
2137 vec![Value::Integer(41)],
2138 vec![Value::Integer(-2)],
2139 ];
2140 let scalar = rows
2141 .iter()
2142 .map(|row| registry.invoke_scalar_function(object_id, row, InvocationLimits::default()))
2143 .collect::<Result<Vec<_>, _>>()
2144 .unwrap();
2145 let batch = registry
2146 .invoke_function_batch(object_id, &rows, InvocationLimits::default())
2147 .unwrap();
2148 assert_eq!(batch, scalar);
2149 assert_eq!(
2150 batch,
2151 vec![Value::Integer(2), Value::Integer(42), Value::Integer(-1)]
2152 );
2153 }
2154
2155 #[test]
2156 fn strict_null_uses_scalar_fallback_and_short_circuits_callback() {
2157 let (registry, object_id) = function_registry(true, Some(increment_batch));
2158 let rows = vec![
2159 vec![Value::Integer(1)],
2160 vec![Value::Null(DataType::Integer)],
2161 ];
2162 assert_eq!(
2163 registry
2164 .invoke_function_batch(object_id, &rows, InvocationLimits::default())
2165 .unwrap(),
2166 vec![Value::Integer(2), Value::Null(DataType::Integer)]
2167 );
2168 }
2169
2170 #[test]
2171 fn cancellation_and_batch_failure_publish_no_partial_results() {
2172 fn cancelled() -> bool {
2173 true
2174 }
2175 let (registry, object_id) = function_registry(false, Some(increment_batch));
2176 assert!(matches!(
2177 registry.invoke_function_batch(
2178 object_id,
2179 &[vec![Value::Integer(1)]],
2180 InvocationLimits {
2181 cancel_check: Some(cancelled),
2182 deadline_unix_ns: u64::MAX,
2183 },
2184 ),
2185 Err(PluginInvocationError::Callback {
2186 status: abi::RADIX_STATUS_CANCELLED,
2187 ..
2188 })
2189 ));
2190
2191 let (registry, object_id) = function_registry(false, Some(partial_batch_error));
2192 assert!(matches!(
2193 registry.invoke_function_batch(
2194 object_id,
2195 &[vec![Value::Integer(1)], vec![Value::Integer(2)]],
2196 InvocationLimits::default(),
2197 ),
2198 Err(PluginInvocationError::Callback {
2199 status: abi::RADIX_STATUS_INTERNAL_ERROR,
2200 ..
2201 })
2202 ));
2203 }
2204
2205 #[test]
2206 fn planner_support_canonicalizes_overlaps_and_carries_dependency_identity() {
2207 let (registry, support_id) = planner_registry(overlapping_plan, 8);
2208 let outcome = registry
2209 .invoke_planner_support(
2210 support_id,
2211 &[
2212 NormalizedPredicateArgument::IndexedColumn,
2213 NormalizedPredicateArgument::Constant(&Value::Integer(5)),
2214 ],
2215 InvocationLimits::default(),
2216 )
2217 .unwrap();
2218 let PlannerSupportOutcome::Plan(plan) = outcome else {
2219 panic!("valid candidate plan fell back")
2220 };
2221 assert_eq!(
2222 plan.spans,
2223 vec![
2224 CandidateKeySpan {
2225 start: Value::Integer(1),
2226 end: Value::Integer(7),
2227 },
2228 CandidateKeySpan {
2229 start: Value::Integer(8),
2230 end: Value::Integer(9),
2231 },
2232 ]
2233 );
2234 assert!(plan.requires_recheck);
2235 assert_eq!(plan.estimated_rows, 4);
2236 assert_eq!(plan.identity.registry_generation, registry.generation());
2237 assert_eq!(plan.identity.support_semantic_revision, 5);
2238 assert_eq!(plan.identity.function_semantic_revision, 2);
2239 assert_eq!(plan.identity.operator_class_semantic_revision, 3);
2240 assert_eq!(plan.identity.key_codec_revision, 4);
2241 }
2242
2243 #[test]
2244 fn planner_support_falls_back_for_unknown_estimate_and_range_explosion() {
2245 for (callback, max_spans) in [
2246 (unknown_estimate_plan as abi::RadixAbiPlannerSupportFnV1, 2),
2247 (exploding_plan as abi::RadixAbiPlannerSupportFnV1, 2),
2248 (unsupported_plan as abi::RadixAbiPlannerSupportFnV1, 2),
2249 ] {
2250 let (registry, support_id) = planner_registry(callback, max_spans);
2251 assert!(matches!(
2252 registry
2253 .invoke_planner_support(
2254 support_id,
2255 &[
2256 NormalizedPredicateArgument::IndexedColumn,
2257 NormalizedPredicateArgument::Constant(&Value::Integer(5)),
2258 ],
2259 InvocationLimits::default(),
2260 )
2261 .unwrap(),
2262 PlannerSupportOutcome::Fallback { .. }
2263 ));
2264 }
2265 }
2266
2267 #[test]
2268 fn malformed_planner_output_fails_closed() {
2269 let (registry, support_id) = planner_registry(malformed_plan, 2);
2270 assert!(matches!(
2271 registry.invoke_planner_support(
2272 support_id,
2273 &[
2274 NormalizedPredicateArgument::IndexedColumn,
2275 NormalizedPredicateArgument::Constant(&Value::Integer(5)),
2276 ],
2277 InvocationLimits::default(),
2278 ),
2279 Err(PluginInvocationError::Contract(_))
2280 ));
2281 }
2282}