use std::{collections::HashSet, sync::Arc};
use reifydb_codec::row::bytes::EncodedBytes;
use reifydb_core::{
error::diagnostic::{
catalog::{namespace_not_found, ringbuffer_not_found},
engine,
},
interface::{
catalog::{
config::{ConfigKey, GetConfig},
namespace::Namespace,
policy::{DataOp, PolicyTargetType},
ringbuffer::{RingBuffer, RingBufferMetadata},
},
resolved::{ResolvedNamespace, ResolvedObject, ResolvedRingBuffer},
},
key::{
any::TaggedKey,
row::{PartitionedRowKey, RowKey},
},
value::column::columns::Columns,
};
use reifydb_evaluate::stack::SymbolTable;
use reifydb_rql::{nodes::DeleteRingBufferNode, query::QueryPlan};
use reifydb_transaction::{multi::RangeScope, transaction::Transaction};
use reifydb_value::{
fragment::Fragment,
params::Params,
return_error,
value::{Value, partition::Partition, row_number::RowNumber},
};
use super::{
context::{RingBufferTarget, WriteExecCtx},
returning::{decode_returning_dictionaries, decode_rows_to_columns, evaluate_returning, with_pre_image},
shape::get_or_create_ringbuffer_shape,
};
use crate::{
Result,
policy::PolicyEvaluator,
transaction::operation::ringbuffer::{RingBufferOperations, apply_ringbuffer_partition_metadata_after_delete},
vm::{
services::Services,
volcano::{
compile::compile,
query::{QueryContext, QueryNode, query_budget},
},
},
};
pub(crate) fn delete_ringbuffer(
services: &Arc<Services>,
txn: &mut Transaction<'_>,
plan: DeleteRingBufferNode,
params: Params,
symbols: &SymbolTable,
) -> Result<Columns> {
let DeleteRingBufferNode {
input,
target,
returning,
} = plan;
let (namespace, ringbuffer) = resolve_delete_ringbuffer_target(services, txn, &target)?;
let resolved_source = build_delete_ringbuffer_resolved_source(&namespace, &ringbuffer);
let target_data = RingBufferTarget {
namespace: &namespace,
ringbuffer: &ringbuffer,
};
let partition_col_indices = compute_partition_col_indices(&ringbuffer);
let shape = get_or_create_ringbuffer_shape(&services.catalog, &ringbuffer, txn)?;
let exec = WriteExecCtx {
services,
symbols,
};
let row_numbers_filter = if let Some(input_plan) = input {
Some(collect_row_numbers_for_ringbuffer_delete(
&exec,
txn,
*input_plan,
&target_data,
&resolved_source,
¶ms,
)?)
} else {
None
};
let (deleted_count, returned_rows) = delete_ringbuffer_partitions(
services,
txn,
&target_data,
&partition_col_indices,
row_numbers_filter.as_ref(),
returning.is_some(),
)?;
if let Some(returning_exprs) = &returning {
let mut columns = decode_rows_to_columns(&shape, &returned_rows);
decode_returning_dictionaries(services, txn, &ringbuffer.columns, &mut columns)?;
let columns = with_pre_image(columns.clone(), &columns);
return evaluate_returning(services, symbols, returning_exprs, columns, txn.identity());
}
Ok(delete_ringbuffer_result(namespace.name(), &ringbuffer.name, deleted_count))
}
#[inline]
fn resolve_delete_ringbuffer_target(
services: &Arc<Services>,
txn: &mut Transaction<'_>,
target: &ResolvedRingBuffer,
) -> Result<(Namespace, RingBuffer)> {
let namespace_name = target.namespace().name();
let Some(namespace) = services.catalog.find_namespace_by_name(txn, namespace_name)? else {
return_error!(namespace_not_found(Fragment::internal(namespace_name), namespace_name));
};
let ringbuffer_name = target.name();
let Some(ringbuffer) = services.catalog.find_ringbuffer_by_name(txn, namespace.id(), ringbuffer_name)? else {
let fragment = Fragment::internal(target.name());
return_error!(ringbuffer_not_found(fragment.clone(), namespace_name, ringbuffer_name));
};
Ok((namespace, ringbuffer))
}
#[inline]
fn build_delete_ringbuffer_resolved_source(namespace: &Namespace, ringbuffer: &RingBuffer) -> Option<ResolvedObject> {
let namespace_ident = Fragment::internal(namespace.name());
let resolved_namespace = ResolvedNamespace::new(namespace_ident, namespace.clone());
let rb_ident = Fragment::internal(ringbuffer.name.clone());
let resolved_rb = ResolvedRingBuffer::new(rb_ident, resolved_namespace, ringbuffer.clone());
Some(ResolvedObject::RingBuffer(resolved_rb))
}
#[inline]
fn compute_partition_col_indices(ringbuffer: &RingBuffer) -> Vec<usize> {
ringbuffer
.partition_by
.iter()
.map(|pb_col| ringbuffer.columns.iter().position(|c| c.name == *pb_col).unwrap())
.collect()
}
fn collect_row_numbers_for_ringbuffer_delete(
exec: &WriteExecCtx<'_>,
txn: &mut Transaction<'_>,
input_plan: QueryPlan,
target: &RingBufferTarget<'_>,
resolved_source: &Option<ResolvedObject>,
params: &Params,
) -> Result<HashSet<RowNumber>> {
let mut row_numbers_to_delete = HashSet::new();
let batch_size = exec.services.catalog.get_config_uint2(ConfigKey::QueryRowBatchSize) as u64;
let mut input_node = compile(
input_plan,
txn,
Arc::new(QueryContext {
services: exec.services.clone(),
source: resolved_source.clone(),
batch_size,
params: params.clone(),
symbols: exec.symbols.clone(),
identity: txn.identity(),
memory: query_budget(exec.services),
}),
);
let context = QueryContext {
services: exec.services.clone(),
source: None,
batch_size,
params: params.clone(),
symbols: exec.symbols.clone(),
identity: txn.identity(),
memory: query_budget(exec.services),
};
input_node.initialize(txn, &context)?;
let mut mutable_context = context.clone();
while let Some(columns) = input_node.next(txn, &mut mutable_context)? {
PolicyEvaluator::new(exec.services, exec.symbols).enforce_write_policies(
txn,
target.namespace.name(),
&target.ringbuffer.name,
DataOp::Delete,
&columns,
PolicyTargetType::RingBuffer,
)?;
if columns.row_numbers().is_empty() {
return_error!(engine::missing_row_number_column());
}
row_numbers_to_delete.extend(columns.row_numbers().iter().copied());
}
Ok(row_numbers_to_delete)
}
fn delete_ringbuffer_partitions(
services: &Arc<Services>,
txn: &mut Transaction<'_>,
target: &RingBufferTarget<'_>,
partition_col_indices: &[usize],
row_numbers_filter: Option<&HashSet<RowNumber>>,
has_returning: bool,
) -> Result<(u64, Vec<(RowNumber, EncodedBytes)>)> {
let ringbuffer = target.ringbuffer;
let mut deleted_count = 0u64;
let mut returned_rows: Vec<(RowNumber, EncodedBytes)> = Vec::new();
let partitions = services.catalog.list_ringbuffer_partitions(txn, ringbuffer)?;
for partition_info in partitions {
let partition_key = partition_info.partition_values.clone();
let partition = partition_info.metadata;
let mut min_remaining_row: Option<u64> = None;
let mut partition_deleted = 0u64;
let partition_hash = if partition_col_indices.is_empty() {
None
} else {
Some(Partition::of(&partition_key))
};
for row_num in collect_partition_row_numbers(txn, ringbuffer, partition_hash, &partition)? {
let should_delete = match row_numbers_filter {
Some(filter) => filter.contains(&row_num),
None => true,
};
if should_delete {
let deleted_values = txn.remove_from_ringbuffer(ringbuffer, partition_hash, row_num)?;
if has_returning {
returned_rows.push((row_num, deleted_values));
}
partition_deleted += 1;
deleted_count += 1;
} else {
min_remaining_row =
Some(min_remaining_row.map_or(row_num.0, |m: u64| m.min(row_num.0)));
}
}
if row_numbers_filter.is_some() {
apply_ringbuffer_partition_metadata_after_delete(
&services.catalog,
txn,
ringbuffer,
&partition_key,
partition,
partition_deleted,
min_remaining_row,
)?;
} else {
services.catalog.remove_partition_metadata(txn, ringbuffer, &partition_key)?;
}
}
Ok((deleted_count, returned_rows))
}
#[inline]
fn delete_ringbuffer_result(namespace: &str, ringbuffer: &str, deleted: u64) -> Columns {
Columns::single_row([
("namespace", Value::Utf8(namespace.to_string())),
("ringbuffer", Value::Utf8(ringbuffer.to_string())),
("deleted", Value::Uint8(deleted)),
])
}
fn collect_partition_row_numbers(
txn: &mut Transaction<'_>,
ringbuffer: &RingBuffer,
partition: Option<Partition>,
metadata: &RingBufferMetadata,
) -> Result<Vec<RowNumber>> {
let Some(partition) = partition else {
let mut out = Vec::new();
for row_num_value in metadata.head..metadata.tail {
let row_num = RowNumber(row_num_value);
if txn.get(&RowKey::new(ringbuffer.id, row_num))?.is_some() {
out.push(row_num);
}
}
return Ok(out);
};
let mut out = Vec::new();
let mut last_key = None;
loop {
let batch: Vec<_> = txn
.range(
PartitionedRowKey::partition_scan_range(ringbuffer.id, partition, last_key.as_ref()),
RangeScope::All,
1024,
)?
.collect::<Result<Vec<_>>>()?;
if batch.is_empty() {
break;
}
let n = batch.len();
for entry in batch {
if let TaggedKey::PartitionedRow(pk) = &entry.key {
out.push(pk.row);
}
last_key = Some(entry.key.clone());
}
if n < 1024 {
break;
}
}
out.sort_by_key(|rn| rn.0);
Ok(out)
}