use crate::{
db::{
PrimaryKeyValue,
cursor::{ContinuationKeyRef, ContinuationRuntime, IndexScanContinuationInput},
data::{DecodedDataStoreKey, RawDataStoreKey},
direction::Direction,
executor::{
LoweredIndexPrefixSpec, LoweredIndexRangeSpec, LoweredKey,
lowered_index_prefix_exact_cardinalities,
},
index::{
IndexEntryExistenceWitness, IndexEntryRowWitness, IndexEntryValue, IndexKey,
RawIndexStoreKey, decode_index_key_frame,
predicate::{
IndexPredicateExecution, eval_index_program_on_frame,
eval_index_program_on_prefix_components,
},
resume_prefix_bounds_for_continuation,
},
registry::StoreHandle,
resource::budget::{
ExecutionConstructionBudget, charge_current_execution_budget,
charge_current_execution_budget_pair,
},
},
error::InternalError,
types::EntityTag,
};
use icydb_diagnostic_code::DiagnosticExecutionBudgetResource;
use std::{cmp::Ordering, mem::size_of, ops::Bound, sync::Arc};
pub(in crate::db::executor) type IndexComponentValues = Arc<[Vec<u8>]>;
pub(in crate::db::executor) type IndexComponentRow = (
DecodedDataStoreKey,
IndexEntryExistenceWitness,
IndexComponentValues,
);
pub(in crate::db::executor) type IndexComponentRows = Vec<IndexComponentRow>;
struct ExactIntersectionPrimaryKey {
value: PrimaryKeyValue,
}
struct MergedPrimaryKeyOrder {
value: PrimaryKeyValue,
bytes_len: usize,
}
impl Eq for MergedPrimaryKeyOrder {}
impl PartialEq for MergedPrimaryKeyOrder {
fn eq(&self, other: &Self) -> bool {
self.value == other.value
}
}
impl Ord for MergedPrimaryKeyOrder {
fn cmp(&self, other: &Self) -> Ordering {
self.value.cmp(&other.value)
}
}
impl PartialOrd for MergedPrimaryKeyOrder {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
fn charge_merged_range_structural_bytes(bytes: usize) -> Result<(), InternalError> {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::TemporaryBytes,
u64::try_from(bytes).unwrap_or(u64::MAX),
)
}
pub(in crate::db::executor) const ACCESS_SCAN_CHUNK_ENTRIES: usize = 64;
const PREFIX_STREAM_SMALL_CHUNK_ENTRIES: usize = 2;
const PREFIX_STREAM_MAX_CHUNK_ENTRIES: usize = 64;
const fn prefix_stream_chunk_entries(fetch_hint: Option<usize>, prefix_count: usize) -> usize {
let Some(fetch_hint) = fetch_hint else {
return ACCESS_SCAN_CHUNK_ENTRIES;
};
if fetch_hint <= PREFIX_STREAM_SMALL_CHUNK_ENTRIES.saturating_mul(2) {
return PREFIX_STREAM_SMALL_CHUNK_ENTRIES;
}
let prefix_count = if prefix_count == 0 { 1 } else { prefix_count };
let fair_prefix_window = fetch_hint.div_ceil(prefix_count);
if fair_prefix_window < PREFIX_STREAM_SMALL_CHUNK_ENTRIES {
PREFIX_STREAM_SMALL_CHUNK_ENTRIES
} else if fair_prefix_window > PREFIX_STREAM_MAX_CHUNK_ENTRIES {
PREFIX_STREAM_MAX_CHUNK_ENTRIES
} else {
fair_prefix_window
}
}
pub(in crate::db::executor) const fn branch_stream_chunk_entries(
index_fetch_hint: Option<usize>,
active_branch_count: usize,
) -> usize {
prefix_stream_chunk_entries(index_fetch_hint, active_branch_count)
}
pub(in crate::db::executor) const fn index_stream_chunk_entries_for_remaining(
chunk_entries: usize,
remaining: Option<usize>,
) -> usize {
let chunk_entries = if chunk_entries == 0 {
ACCESS_SCAN_CHUNK_ENTRIES
} else {
chunk_entries
};
match remaining {
Some(remaining) if remaining < chunk_entries => remaining,
Some(_) | None => chunk_entries,
}
}
pub(in crate::db::executor) const fn index_stream_output_limit_for_chunk(
remaining: Option<usize>,
chunk_entries: usize,
) -> Option<usize> {
match remaining {
Some(remaining) if remaining < chunk_entries => Some(remaining),
Some(_) => Some(chunk_entries),
None => None,
}
}
pub(in crate::db::executor) fn apply_index_scan_chunk_progress(
anchor: &mut Option<RawIndexStoreKey>,
remaining: &mut Option<usize>,
exhausted: &mut bool,
emitted: usize,
last_raw_key: Option<RawIndexStoreKey>,
) {
if let Some(raw_key) = last_raw_key {
*anchor = Some(raw_key);
} else {
*exhausted = true;
}
if let Some(remaining) = remaining.as_mut() {
*remaining = remaining.saturating_sub(emitted);
if *remaining == 0 {
*exhausted = true;
}
}
}
pub(in crate::db::executor) fn index_predicate_rejects_prefix_components(
prefix_components: &[Vec<u8>],
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> bool {
predicate_execution
.and_then(|execution| eval_index_program_on_prefix_components(prefix_components, execution))
.is_some_and(|passed| !passed)
}
pub(in crate::db::executor) fn active_lowered_index_prefix_specs<'a>(
empty_proof_store: Option<StoreHandle>,
index_prefix_specs: &'a [LoweredIndexPrefixSpec],
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Vec<&'a LoweredIndexPrefixSpec> {
let mut active_specs = Vec::with_capacity(index_prefix_specs.len());
if let Some(store) = empty_proof_store {
let exact_cardinalities =
lowered_index_prefix_exact_cardinalities(store, index_prefix_specs);
for (position, spec) in index_prefix_specs.iter().enumerate() {
if exact_cardinalities
.as_ref()
.and_then(|counts| counts.get(position))
== Some(&0)
{
continue;
}
if index_predicate_rejects_prefix_components(
spec.prefix_components(),
predicate_execution,
) {
continue;
}
active_specs.push(spec);
}
} else {
for spec in index_prefix_specs {
if index_predicate_rejects_prefix_components(
spec.prefix_components(),
predicate_execution,
) {
continue;
}
active_specs.push(spec);
}
}
active_specs
}
pub(in crate::db::executor) struct PrimaryScan;
impl PrimaryScan {
pub(in crate::db::executor) fn decode_data_key(
raw: &RawDataStoreKey,
) -> Result<DecodedDataStoreKey, InternalError> {
DecodedDataStoreKey::try_from_raw(raw).map_err(|_err| InternalError::identity_corruption())
}
}
pub(in crate::db::executor) struct IndexScan;
pub(in crate::db::executor) struct IndexDecodedKeyScanChunk {
keys: Vec<DecodedDataStoreKey>,
last_raw_key: Option<RawIndexStoreKey>,
}
impl IndexDecodedKeyScanChunk {
#[must_use]
const fn new(keys: Vec<DecodedDataStoreKey>, last_raw_key: Option<RawIndexStoreKey>) -> Self {
Self { keys, last_raw_key }
}
#[must_use]
pub(in crate::db::executor) fn into_decoded_keys_and_resume_anchor(
self,
) -> (Vec<DecodedDataStoreKey>, Option<RawIndexStoreKey>) {
(self.keys, self.last_raw_key)
}
}
pub(in crate::db::executor) struct IndexComponentScanChunk {
rows: IndexComponentRows,
last_raw_key: Option<RawIndexStoreKey>,
}
impl IndexComponentScanChunk {
#[must_use]
const fn new(rows: IndexComponentRows, last_raw_key: Option<RawIndexStoreKey>) -> Self {
Self { rows, last_raw_key }
}
#[must_use]
pub(in crate::db::executor) fn into_component_rows_and_resume_anchor(
self,
) -> (IndexComponentRows, Option<RawIndexStoreKey>) {
(self.rows, self.last_raw_key)
}
}
impl IndexScan {
const LIMITED_SCAN_PREALLOC_CAP: usize = 32;
fn charge_exact_intersection_structural_work(
child_cardinalities: &[u64],
) -> Result<u64, InternalError> {
let Some((&first, remaining)) = child_cardinalities.split_first() else {
return Err(InternalError::executor_invariant());
};
if first == 0 {
return Err(InternalError::executor_invariant());
}
let mut primary_key_slots = first;
let mut overlap_upper_bound = first;
let mut comparison_upper_bound = 0u64;
for cardinality in remaining {
if *cardinality == 0 {
return Err(InternalError::executor_invariant());
}
primary_key_slots = primary_key_slots
.checked_add(*cardinality)
.ok_or_else(InternalError::executor_invariant)?;
comparison_upper_bound = comparison_upper_bound
.checked_add(
overlap_upper_bound
.checked_add(*cardinality)
.and_then(|comparisons| comparisons.checked_sub(1))
.ok_or_else(InternalError::executor_invariant)?,
)
.ok_or_else(InternalError::executor_invariant)?;
overlap_upper_bound = overlap_upper_bound.min(*cardinality);
primary_key_slots = primary_key_slots
.checked_add(overlap_upper_bound)
.ok_or_else(InternalError::executor_invariant)?;
}
let primary_key_bytes = primary_key_slots
.checked_mul(
u64::try_from(size_of::<ExactIntersectionPrimaryKey>()).unwrap_or(u64::MAX),
)
.ok_or_else(InternalError::executor_invariant)?;
let result_bytes = overlap_upper_bound
.checked_mul(u64::try_from(size_of::<DecodedDataStoreKey>()).unwrap_or(u64::MAX))
.ok_or_else(InternalError::executor_invariant)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::TemporaryBytes,
primary_key_bytes
.checked_add(result_bytes)
.ok_or_else(InternalError::executor_invariant)?,
)?;
Ok(comparison_upper_bound)
}
fn collect_exact_intersection_child(
store: StoreHandle,
spec: &LoweredIndexPrefixSpec,
expected_cardinality: u64,
additional_cursor_steps: u64,
direction: Direction,
) -> Result<Vec<ExactIntersectionPrimaryKey>, InternalError> {
let bounds = spec.raw_bounds(&ExecutionConstructionBudget)?;
Self::collect_exact_intersection_child_in_bounds(
store,
bounds,
expected_cardinality,
additional_cursor_steps,
direction,
)
}
fn collect_exact_intersection_child_in_bounds(
store: StoreHandle,
bounds: (&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>),
expected_cardinality: u64,
additional_cursor_steps: u64,
direction: Direction,
) -> Result<Vec<ExactIntersectionPrimaryKey>, InternalError> {
let expected = usize::try_from(expected_cardinality)
.map_err(|_| InternalError::executor_invariant())?;
let cursor_steps = expected_cardinality
.checked_add(additional_cursor_steps)
.ok_or_else(InternalError::executor_invariant)?;
charge_current_execution_budget_pair(
(
DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
expected_cardinality,
),
(DiagnosticExecutionBudgetResource::CursorSteps, cursor_steps),
)?;
let mut keys = Vec::new();
keys.try_reserve_exact(expected)
.map_err(|_| InternalError::executor_internal())?;
let extra_capacity = keys.capacity().saturating_sub(expected);
let extra_capacity_bytes = extra_capacity
.checked_mul(size_of::<ExactIntersectionPrimaryKey>())
.ok_or_else(InternalError::executor_invariant)?;
if extra_capacity_bytes != 0 {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::TemporaryBytes,
u64::try_from(extra_capacity_bytes).unwrap_or(u64::MAX),
)?;
}
let mut raw_bytes_read = 0u64;
let scan_result = store.with_index(|index_store| {
index_store.visit_raw_entries_in_range(
(bounds.0.as_ref(), bounds.1.as_ref()),
direction,
|raw_key, entry| {
let exceeds_cardinality_proof = keys.len() >= expected;
if exceeds_cardinality_proof {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
1,
)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::CursorSteps,
1,
)?;
}
let raw_bytes = u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX);
raw_bytes_read = raw_bytes_read
.checked_add(raw_bytes)
.ok_or_else(InternalError::executor_invariant)?;
let (primary_key, _) = IndexKey::primary_key_value_and_bytes_from_raw(raw_key)
.map_err(|_| InternalError::index_corruption())?;
let existence_witness = entry
.decode_existence_witness()
.map_err(|_| InternalError::index_entry_decode_failed())?;
if matches!(existence_witness, IndexEntryExistenceWitness::Missing) {
return Err(InternalError::store_corruption());
}
if exceeds_cardinality_proof {
return Err(InternalError::store_corruption());
}
keys.push(ExactIntersectionPrimaryKey { value: primary_key });
Ok(false)
},
)
});
charge_current_execution_budget_pair(
(
DiagnosticExecutionBudgetResource::StoredBytesRead,
raw_bytes_read,
),
(
DiagnosticExecutionBudgetResource::DecodedBytes,
raw_bytes_read,
),
)?;
scan_result?;
if keys.len() != expected {
return Err(InternalError::store_corruption());
}
Ok(keys)
}
fn intersect_exact_primary_keys(
mut overlap: Vec<ExactIntersectionPrimaryKey>,
keys: &[ExactIntersectionPrimaryKey],
direction: Direction,
) -> Vec<ExactIntersectionPrimaryKey> {
let mut right = 0;
overlap.retain(|left_key| {
while let Some(right_key) = keys.get(right) {
let order = match direction {
Direction::Asc => left_key.value.cmp(&right_key.value),
Direction::Desc => right_key.value.cmp(&left_key.value),
};
match order {
std::cmp::Ordering::Less => return false,
std::cmp::Ordering::Greater => {
right += 1;
}
std::cmp::Ordering::Equal => {
right += 1;
return true;
}
}
}
false
});
overlap
}
pub(in crate::db::executor) fn exact_prefix_intersection_structural(
store: StoreHandle,
entity: EntityTag,
specs: &[&LoweredIndexPrefixSpec],
child_cardinalities: &[u64],
direction: Direction,
) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
if specs.len() != child_cardinalities.len() || specs.is_empty() {
return Err(InternalError::executor_invariant());
}
let comparison_cursor_steps =
Self::charge_exact_intersection_structural_work(child_cardinalities)?;
let mut overlap = None;
for (spec, expected_cardinality) in specs.iter().zip(child_cardinalities.iter().copied()) {
let additional_cursor_steps = if overlap.is_none() {
comparison_cursor_steps
} else {
0
};
let keys = Self::collect_exact_intersection_child(
store,
spec,
expected_cardinality,
additional_cursor_steps,
direction,
)?;
overlap = Some(match overlap {
None => keys,
Some(current) => Self::intersect_exact_primary_keys(current, &keys, direction),
});
}
let overlap = overlap.ok_or_else(InternalError::executor_invariant)?;
let mut result = Vec::new();
result
.try_reserve_exact(overlap.len())
.map_err(|_| InternalError::executor_internal())?;
let extra_capacity = result.capacity().saturating_sub(overlap.len());
let extra_capacity_bytes = extra_capacity
.checked_mul(size_of::<DecodedDataStoreKey>())
.ok_or_else(InternalError::executor_invariant)?;
if extra_capacity_bytes != 0 {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::TemporaryBytes,
u64::try_from(extra_capacity_bytes).unwrap_or(u64::MAX),
)?;
}
for key in overlap {
result.push(DecodedDataStoreKey::new_primary_key_value(
entity, &key.value,
));
}
Ok(result)
}
pub(in crate::db::executor) fn merged_components_without_index_values(
store: StoreHandle,
entity: EntityTag,
bounds: &[(&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>)],
direction: Direction,
limit: usize,
) -> Result<Option<IndexComponentRows>, InternalError> {
let mut rows = Vec::with_capacity(limit.min(Self::LIMITED_SCAN_PREALLOC_CAP));
let mut entries_visited = 0u64;
let mut raw_bytes_read = 0u64;
let scan_result = store.with_index(|index_store| {
index_store.visit_raw_entries_in_merged_ranges(
bounds,
direction,
charge_merged_range_structural_bytes,
|raw_key| {
entries_visited = entries_visited
.checked_add(1)
.ok_or_else(InternalError::executor_invariant)?;
let raw_key_bytes = u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX);
raw_bytes_read = raw_bytes_read
.checked_add(raw_key_bytes)
.ok_or_else(InternalError::executor_invariant)?;
let (primary_key_value, primary_key_bytes) =
IndexKey::primary_key_value_and_bytes_from_raw(raw_key)
.map_err(|_| InternalError::index_corruption())?;
Ok(MergedPrimaryKeyOrder {
value: primary_key_value,
bytes_len: primary_key_bytes.len(),
})
},
|order_key, raw_key, value| {
let primary_key_value = order_key.value;
let row_witness = value
.decode_row_witness_from_primary_key_value(&primary_key_value)
.map_err(|_| InternalError::index_entry_decode_failed())?;
let bytes_start = raw_key
.as_bytes()
.len()
.checked_sub(order_key.bytes_len)
.ok_or_else(InternalError::executor_invariant)?;
let primary_key_bytes = raw_key
.as_bytes()
.get(bytes_start..)
.ok_or_else(InternalError::executor_invariant)?;
let data_key = DecodedDataStoreKey::new_with_raw_primary_key_value(
entity,
&primary_key_value,
RawDataStoreKey::from_entity_and_primary_key_bytes(
entity,
primary_key_bytes,
),
);
rows.push((data_key, row_witness.existence_witness(), Arc::default()));
Ok(rows.len() == limit)
},
)
});
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
entries_visited,
)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::StoredBytesRead,
raw_bytes_read,
)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::CursorSteps,
entries_visited,
)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::DecodedBytes,
raw_bytes_read,
)?;
let supported = scan_result?;
Ok(supported.then_some(rows))
}
pub(in crate::db::executor) fn prefix_structural(
store: StoreHandle,
entity_tag: EntityTag,
spec: &LoweredIndexPrefixSpec,
continuation: IndexScanContinuationInput<'_>,
limit: usize,
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
let (lower, upper) = spec.raw_bounds(&ExecutionConstructionBudget)?;
let (lower, upper) = resume_prefix_bounds_for_continuation(
continuation.direction(),
continuation.anchor(),
lower,
upper,
)?;
Self::resolve_data_values_in_raw_range_limited(
store,
entity_tag,
&lower,
&upper,
IndexScanContinuationInput::new(None, continuation.direction()),
limit,
predicate_execution,
)
}
#[expect(clippy::too_many_arguments)]
pub(in crate::db::executor) fn components_structural(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
limit: usize,
component_indices: &[usize],
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<IndexComponentRows, InternalError> {
if limit == 0 {
return Ok(Vec::new());
}
let continuation = ContinuationRuntime::new(continuation);
let bounds = continuation.scan_bounds((lower, upper))?;
let mut out = Vec::with_capacity(limit.min(Self::LIMITED_SCAN_PREALLOC_CAP));
store.with_index(|index_store| {
index_store.visit_raw_entries_in_range(
bounds,
continuation.direction(),
|raw_key, value| {
Self::accept_scan_key(&continuation, raw_key)?;
Self::decode_index_entry_and_push_with_components(
entity_tag,
raw_key,
value,
&mut out,
Some(limit),
component_indices,
predicate_execution,
)
},
)
})?;
Ok(out)
}
pub(in crate::db::executor) fn range_structural(
store: StoreHandle,
entity_tag: EntityTag,
spec: &LoweredIndexRangeSpec,
continuation: IndexScanContinuationInput<'_>,
limit: usize,
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
if index_predicate_rejects_prefix_components(spec.prefix_components(), predicate_execution)
{
return Ok(Vec::new());
}
Self::resolve_data_values_in_raw_range_limited(
store,
entity_tag,
spec.lower(),
spec.upper(),
continuation,
limit,
predicate_execution,
)
}
pub(in crate::db::executor) fn chunk_structural(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
max_entries: usize,
output_limit: Option<usize>,
) -> Result<IndexDecodedKeyScanChunk, InternalError> {
Self::resolve_chunk(
store,
entity_tag,
lower,
upper,
continuation,
max_entries,
output_limit,
)
}
#[expect(clippy::too_many_arguments)]
pub(in crate::db::executor) fn components_chunk_structural(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
max_entries: usize,
output_limit: Option<usize>,
component_indices: &[usize],
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<IndexComponentScanChunk, InternalError> {
Self::resolve_component_chunk(
store,
entity_tag,
lower,
upper,
continuation,
max_entries,
output_limit,
component_indices,
predicate_execution,
)
}
fn resolve_data_values_in_raw_range_limited(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
limit: usize,
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
if limit == 0 {
return Ok(Vec::new());
}
let continuation = ContinuationRuntime::new(continuation);
let bounds = continuation.scan_bounds((lower, upper))?;
let mut keys = Vec::with_capacity(limit.min(Self::LIMITED_SCAN_PREALLOC_CAP));
store.with_index(|index_store| {
index_store.visit_raw_entries_in_range(
bounds,
continuation.direction(),
|raw_key, value| {
Self::accept_scan_key(&continuation, raw_key)?;
Self::decode_index_entry_and_push(
entity_tag,
raw_key,
value,
&mut keys,
Some(limit),
predicate_execution,
)
},
)
})?;
Ok(keys)
}
fn resolve_chunk(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
max_entries: usize,
output_limit: Option<usize>,
) -> Result<IndexDecodedKeyScanChunk, InternalError> {
if max_entries == 0 || matches!(output_limit, Some(0)) {
return Ok(IndexDecodedKeyScanChunk::new(Vec::new(), None));
}
let continuation = ContinuationRuntime::new(continuation);
let bounds = continuation.scan_bounds((lower, upper))?;
let mut keys = Vec::with_capacity(max_entries.min(Self::LIMITED_SCAN_PREALLOC_CAP));
let mut last_raw_key = None;
let mut scanned_entries = 0usize;
store.with_index(|index_store| {
index_store.visit_raw_entries_in_range(
bounds,
continuation.direction(),
|raw_key, value| {
Self::accept_scan_key(&continuation, raw_key)?;
scanned_entries = scanned_entries.saturating_add(1);
let output_full = Self::decode_index_entry_and_push(
entity_tag,
raw_key,
value,
&mut keys,
output_limit,
None,
)?;
let stop = output_full || scanned_entries == max_entries;
if stop {
last_raw_key = Some(raw_key.clone());
}
Ok::<_, InternalError>(stop)
},
)
})?;
let chunk = IndexDecodedKeyScanChunk::new(keys, last_raw_key);
Ok(chunk)
}
#[expect(clippy::too_many_arguments)]
fn resolve_component_chunk(
store: StoreHandle,
entity_tag: EntityTag,
lower: &Bound<LoweredKey>,
upper: &Bound<LoweredKey>,
continuation: IndexScanContinuationInput<'_>,
max_entries: usize,
output_limit: Option<usize>,
component_indices: &[usize],
predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<IndexComponentScanChunk, InternalError> {
if max_entries == 0 || matches!(output_limit, Some(0)) {
return Ok(IndexComponentScanChunk::new(Vec::new(), None));
}
let continuation = ContinuationRuntime::new(continuation);
let bounds = continuation.scan_bounds((lower, upper))?;
let mut rows = Vec::with_capacity(max_entries.min(Self::LIMITED_SCAN_PREALLOC_CAP));
let mut last_raw_key = None;
let mut scanned_entries = 0usize;
store.with_index(|index_store| {
index_store.visit_raw_entries_in_range(
bounds,
continuation.direction(),
|raw_key, value| {
Self::accept_scan_key(&continuation, raw_key)?;
scanned_entries = scanned_entries.saturating_add(1);
let output_full = Self::decode_index_entry_and_push_with_components(
entity_tag,
raw_key,
value,
&mut rows,
output_limit,
component_indices,
predicate_execution,
)?;
let stop = output_full || scanned_entries == max_entries;
if stop {
last_raw_key = Some(raw_key.clone());
}
Ok::<_, InternalError>(stop)
},
)
})?;
Ok(IndexComponentScanChunk::new(rows, last_raw_key))
}
fn accept_scan_key(
continuation: &ContinuationRuntime<'_>,
raw_key: &RawIndexStoreKey,
) -> Result<(), InternalError> {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
1,
)?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::StoredBytesRead,
u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX),
)?;
charge_current_execution_budget(DiagnosticExecutionBudgetResource::CursorSteps, 1)?;
continuation.accept_key(ContinuationKeyRef::scan(raw_key))
}
fn decode_index_entry_and_push(
entity: EntityTag,
raw_key: &RawIndexStoreKey,
value: &IndexEntryValue,
out: &mut Vec<DecodedDataStoreKey>,
limit: Option<usize>,
index_predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<bool, InternalError> {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::DecodedBytes,
u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX),
)?;
let (primary_key_value, primary_key_bytes) =
if let Some(execution) = index_predicate_execution {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::PredicateExpressionSteps,
1,
)?;
let frame = decode_index_key_frame(raw_key)
.map_err(|_| InternalError::index_corruption())?;
if !eval_index_program_on_frame(&frame, execution)? {
return Ok(false);
}
frame
.primary_key_value_and_bytes()
.map_err(|_| InternalError::index_entry_decode_failed())?
} else {
IndexKey::primary_key_value_and_bytes_from_raw(raw_key)
.map_err(|_| InternalError::index_corruption())?
};
let row_witness = value
.decode_row_witness_from_primary_key_value(&primary_key_value)
.map_err(|_| InternalError::index_entry_decode_failed())?;
out.push(Self::data_key_from_row_witness_with_primary_key_bytes(
entity,
&row_witness,
primary_key_bytes,
));
if let Some(limit) = limit
&& out.len() == limit
{
return Ok(true);
}
Ok(false)
}
fn decode_index_entry_and_push_with_components(
entity: EntityTag,
raw_key: &RawIndexStoreKey,
value: &IndexEntryValue,
out: &mut IndexComponentRows,
limit: Option<usize>,
component_indices: &[usize],
index_predicate_execution: Option<IndexPredicateExecution<'_>>,
) -> Result<bool, InternalError> {
if component_indices.is_empty() && index_predicate_execution.is_none() {
return Self::decode_index_entry_and_push_without_components(
entity, raw_key, value, out, limit,
);
}
let frame =
decode_index_key_frame(raw_key).map_err(|_| InternalError::index_corruption())?;
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::DecodedBytes,
u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX),
)?;
if component_indices
.iter()
.any(|index| *index >= frame.components().len())
{
return Err(InternalError::index_invariant());
}
if let Some(execution) = index_predicate_execution {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::PredicateExpressionSteps,
1,
)?;
if !eval_index_program_on_frame(&frame, execution)? {
return Ok(false);
}
}
let (primary_key_value, primary_key_bytes) = frame
.primary_key_value_and_bytes()
.map_err(|_| InternalError::index_entry_decode_failed())?;
let row_witness = value
.decode_row_witness_from_primary_key_value(&primary_key_value)
.map_err(|_| InternalError::index_entry_decode_failed())?;
let mut components = Vec::with_capacity(component_indices.len());
for component_index in component_indices {
let component = frame
.components()
.get(*component_index)
.ok_or_else(InternalError::index_invariant)?;
components.push(component.to_vec());
}
let components: Arc<[Vec<u8>]> = Arc::from(components);
out.push((
Self::data_key_from_row_witness_with_primary_key_bytes(
entity,
&row_witness,
primary_key_bytes,
),
row_witness.existence_witness(),
components,
));
if let Some(limit) = limit
&& out.len() == limit
{
return Ok(true);
}
Ok(false)
}
fn decode_index_entry_and_push_without_components(
entity: EntityTag,
raw_key: &RawIndexStoreKey,
value: &IndexEntryValue,
out: &mut IndexComponentRows,
limit: Option<usize>,
) -> Result<bool, InternalError> {
charge_current_execution_budget(
DiagnosticExecutionBudgetResource::DecodedBytes,
u64::try_from(raw_key.as_bytes().len()).unwrap_or(u64::MAX),
)?;
let (primary_key_value, primary_key_bytes) =
IndexKey::primary_key_value_and_bytes_from_raw(raw_key)
.map_err(|_| InternalError::index_corruption())?;
let row_witness = value
.decode_row_witness_from_primary_key_value(&primary_key_value)
.map_err(|_| InternalError::index_entry_decode_failed())?;
out.push((
Self::data_key_from_row_witness_with_primary_key_bytes(
entity,
&row_witness,
primary_key_bytes,
),
row_witness.existence_witness(),
Arc::default(),
));
Ok(limit.is_some_and(|limit| out.len() == limit))
}
fn data_key_from_row_witness_with_primary_key_bytes(
entity: EntityTag,
row_witness: &IndexEntryRowWitness,
primary_key_bytes: &[u8],
) -> DecodedDataStoreKey {
DecodedDataStoreKey::new_with_raw_primary_key_value(
entity,
row_witness.primary_key_value(),
RawDataStoreKey::from_entity_and_primary_key_bytes(entity, primary_key_bytes),
)
}
}
#[cfg(test)]
mod tests;