use std::{collections::HashMap, marker::PhantomData, sync::Arc};
use arrow_array::ArrayRef;
use arrow_schema::FieldRef;
use reifydb_catalog::{
catalog::Catalog,
error::{CatalogError, CatalogObjectKind},
};
use reifydb_codec::row::{
bytes::{EncodedBytes, RowBuilder, SourceRowBuilder},
pod::EncodedPodRow,
shape::RowShape,
};
use reifydb_core::{
common::TimeSource,
delta::Delta,
durability::Durability,
interface::catalog::{
column::Column,
namespace::Namespace,
object::ObjectId,
policy::{DataOp, PolicyTargetType},
ringbuffer::RingBuffer,
series::{Series, SeriesPartitionMetadata, decode_series_partition_metadata},
storage::StorageId,
table::Table,
},
internal_error,
key::{any::TaggedKey, series::SeriesPartitionMetadataKey},
value::batch::batch,
};
use reifydb_evaluate::stack::SymbolTable;
use reifydb_transaction::{
multi::transaction::write::Rewrite,
transaction::{Transaction, command::CommandTransaction},
};
use reifydb_value::{
error::Error,
fragment::Fragment,
value::{
Value, column_view::ColumnView, constraint::Constraint, datetime::DateTime, identity::IdentityId,
partition::Partition, value_type::ValueType,
},
};
use super::{
BulkInsertResult, RingBufferInsertResult, SeriesInsertResult, TableInsertResult,
coerce::{RowCoercer, with_row_note},
validation::{CoercedTags, coerce_columns},
};
use crate::{
Result,
bulk_insert::storage::{
ringbuffer::{PendingRingBufferInsert, RingBufferInsertBuilder},
series::{PendingSeriesInsert, SeriesInsertBuilder},
table::{PendingTableInsert, TableInsertBuilder},
},
engine::StandardEngine,
error::EngineError,
flow_sync::{load_flow_diff_need, sync_transactional_views},
partition::resolve_partitions,
policy::PolicyEvaluator,
transaction::operation::series::{SeriesRows, write_series_rows},
vm::{
instruction::dml::{
coerce::key_out_of_range,
columns::{CastColumns, fill_sequences, intern_dictionary_columns},
partition::{ringbuffer_batch_partitions, save_all_partition_metadata},
primary_key::{Violation, get_primary_key},
ringbuffer_insert::insert_ringbuffer_chunks,
series_insert::{assign_series_keys, series_batch_partitions, write_inserted_series_metadata},
shape::{
get_or_create_ringbuffer_shape, get_or_create_series_shape, get_or_create_table_shape,
},
table_insert::{EncodedTableRows, table_batch_partitions, write_table_rows},
time::{RowTime, populator_index},
},
services::Services,
},
};
pub trait ValidationMode: sealed::Sealed + 'static {
const VALIDATED: bool;
fn run<F, R>(txn: &mut CommandTransaction, total_rows: usize, body: F) -> Result<R>
where
F: FnOnce(&mut CommandTransaction) -> Result<R>;
}
pub struct Validated;
impl ValidationMode for Validated {
const VALIDATED: bool = true;
fn run<F, R>(txn: &mut CommandTransaction, total_rows: usize, body: F) -> Result<R>
where
F: FnOnce(&mut CommandTransaction) -> Result<R>,
{
run_checked(txn, total_rows, body)
}
}
pub struct Unchecked;
impl ValidationMode for Unchecked {
const VALIDATED: bool = false;
fn run<F, R>(txn: &mut CommandTransaction, _total_rows: usize, body: F) -> Result<R>
where
F: FnOnce(&mut CommandTransaction) -> Result<R>,
{
txn.execute_bulk_unchecked(body)
}
}
fn run_checked<F, R>(txn: &mut CommandTransaction, total_rows: usize, body: F) -> Result<R>
where
F: FnOnce(&mut CommandTransaction) -> Result<R>,
{
if total_rows > 0 {
txn.reserve_writes(total_rows.saturating_mul(2))?;
}
let r = body(txn)?;
txn.commit()?;
Ok(r)
}
pub mod sealed {
use super::{Unchecked, Validated};
pub trait Sealed {}
impl Sealed for Validated {}
impl Sealed for Unchecked {}
}
pub struct BulkInsertBuilder<'e, V: ValidationMode = Validated> {
engine: &'e StandardEngine,
identity: IdentityId,
pending_tables: Vec<PendingTableInsert>,
pending_ringbuffers: Vec<PendingRingBufferInsert>,
pending_series: Vec<PendingSeriesInsert>,
_validation: PhantomData<V>,
}
impl<'e> BulkInsertBuilder<'e, Validated> {
pub(crate) fn new(engine: &'e StandardEngine, identity: IdentityId) -> Self {
Self {
engine,
identity,
pending_tables: Vec::new(),
pending_ringbuffers: Vec::new(),
pending_series: Vec::new(),
_validation: PhantomData,
}
}
}
impl<'e> BulkInsertBuilder<'e, Unchecked> {
pub(crate) fn new_unchecked(engine: &'e StandardEngine, identity: IdentityId) -> Self {
Self {
engine,
identity,
pending_tables: Vec::new(),
pending_ringbuffers: Vec::new(),
pending_series: Vec::new(),
_validation: PhantomData,
}
}
}
impl<'e, V: ValidationMode> BulkInsertBuilder<'e, V> {
pub fn table<'a>(&'a mut self, qualified_name: &str) -> TableInsertBuilder<'a, 'e, V> {
let (namespace, table) = parse_qualified_name(qualified_name);
TableInsertBuilder::new(self, namespace, table)
}
pub fn ringbuffer<'a>(&'a mut self, qualified_name: &str) -> RingBufferInsertBuilder<'a, 'e, V> {
let (namespace, ringbuffer) = parse_qualified_name(qualified_name);
RingBufferInsertBuilder::new(self, namespace, ringbuffer)
}
pub fn series<'a>(&'a mut self, qualified_name: &str) -> SeriesInsertBuilder<'a, 'e, V> {
let (namespace, series) = parse_qualified_name(qualified_name);
SeriesInsertBuilder::new(self, namespace, series)
}
pub(super) fn add_table_insert(&mut self, pending: PendingTableInsert) {
self.pending_tables.push(pending);
}
pub(super) fn add_ringbuffer_insert(&mut self, pending: PendingRingBufferInsert) {
self.pending_ringbuffers.push(pending);
}
pub(super) fn add_series_insert(&mut self, pending: PendingSeriesInsert) {
self.pending_series.push(pending);
}
pub fn execute(self) -> Result<BulkInsertResult> {
self.run()
}
fn run(self) -> Result<BulkInsertResult> {
self.engine.reject_if_read_only()?;
let mut txn = self.engine.begin_command(self.identity, Durability::Async)?;
let catalog = self.engine.catalog();
let services = self.engine.services();
let clock = self.engine.clock();
let total_rows = self.total_pending_rows();
let pending_tables = self.pending_tables;
let pending_ringbuffers = self.pending_ringbuffers;
let pending_series = self.pending_series;
let identity = self.identity;
let checked_commit =
!pending_ringbuffers.is_empty() || has_keyed_table(&catalog, &mut txn, &pending_tables)?;
let body = move |txn: &mut CommandTransaction| {
let writer = Writer {
catalog: &catalog,
services: &services,
now: clock.now(),
coercer: RowCoercer::new(identity),
};
load_flow_diff_need(writer.catalog, &mut Transaction::Command(txn))?;
let result = run_all_pending::<V>(
&writer,
txn,
pending_tables,
pending_ringbuffers,
pending_series,
)?;
sync_transactional_views(writer.services, Transaction::Command(txn))?;
Ok(result)
};
if checked_commit {
run_checked(&mut txn, total_rows, body)
} else {
V::run(&mut txn, total_rows, body)
}
}
#[inline]
fn total_pending_rows(&self) -> usize {
self.pending_tables.iter().map(|p| p.row_count()).sum::<usize>()
+ self.pending_ringbuffers.iter().map(|p| p.row_count()).sum::<usize>()
+ self.pending_series.iter().map(|p| p.row_count()).sum::<usize>()
}
}
struct Writer<'a> {
catalog: &'a Catalog,
services: &'a Arc<Services>,
now: DateTime,
coercer: RowCoercer,
}
#[inline]
fn run_all_pending<V: ValidationMode>(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
pending_tables: Vec<PendingTableInsert>,
pending_ringbuffers: Vec<PendingRingBufferInsert>,
pending_series: Vec<PendingSeriesInsert>,
) -> Result<BulkInsertResult> {
let mut result = BulkInsertResult::default();
for pending in pending_tables {
result.tables.push(execute_table_insert::<V>(writer, txn, &pending)?);
}
for pending in pending_ringbuffers {
result.ringbuffers.push(execute_ringbuffer_insert::<V>(writer, txn, &pending)?);
}
for pending in pending_series {
result.series.push(execute_series_insert::<V>(writer, txn, &pending)?);
}
Ok(result)
}
struct Checks<'a> {
name: &'a str,
columns: &'a [Column],
filled: &'a [Vec<usize>],
series: Option<&'a Series>,
time: &'a TimeSource,
}
struct Checked {
keys: Vec<u64>,
tags: Vec<Option<u8>>,
}
fn check_rows<V: ValidationMode>(
checks: &Checks<'_>,
columns: &[(FieldRef, ArrayRef)],
tags: Option<CoercedTags>,
rows: usize,
) -> Result<Checked> {
let views = columns.iter().map(ColumnView::try_from).collect::<Result<Vec<_>>>()?;
let mut failure: Option<(usize, Error)> = None;
let limit = |failure: &Option<(usize, Error)>| failure.as_ref().map_or(rows, |(row, _)| *row);
if V::VALIDATED {
for ((column, view), filled) in checks.columns.iter().zip(&views).zip(checks.filled) {
if let Some(row) = first_constraint_failure(column, view, filled, limit(&failure)) {
failure = Some((row, constraint_error(column, view.get_value(row), checks.name, row)));
}
}
}
let mut keys = Vec::new();
if let Some(series) = checks.series {
let index = key_column_index(series)?;
let (column, view) = (&checks.columns[index], &views[index]);
let given = series.key.keys_to_u64(view);
let dictionary_key = column.dictionary_id.is_some();
for (row, key) in given.iter().enumerate().take(limit(&failure)) {
if view.none_at(row) {
failure = Some((row, constraint_error(column, view.get_value(row), checks.name, row)));
break;
}
if dictionary_key || key.is_none() {
failure = Some((row, key_out_of_range(&series.key, &view.get_value(row))));
break;
}
}
if failure.is_none() {
keys = given.into_iter().collect::<Option<Vec<u64>>>().ok_or_else(|| {
internal_error!("bulk insert into series {} kept a row without a key", series.name)
})?;
}
}
let mut variant_tags = vec![None; rows];
if let Some((tags, tag_failure)) = tags {
variant_tags = tags;
if let Some((row, error)) = tag_failure
&& row < limit(&failure)
{
failure = Some((row, error));
}
}
if let TimeSource::Event {
ts,
} = checks.time && let Some(index) = checks.columns.iter().position(|column| &column.name == ts)
&& checks.columns[index].dictionary_id.is_none()
{
let view = &views[index];
if let Some(row) = (0..limit(&failure)).find(|&row| view.none_at(row)) {
failure = Some((
row,
EngineError::TimePopulatorNotDateTime {
object: checks.name.to_string(),
column: ts.to_string(),
found: format!("{:?}", Value::none()),
}
.into(),
));
}
}
if let Some((_, error)) = failure {
return Err(error);
}
Ok(Checked {
keys,
tags: variant_tags,
})
}
fn first_constraint_failure(column: &Column, view: &ColumnView<'_>, filled: &[usize], limit: usize) -> Option<usize> {
let target = column.constraint.get_type();
let max_bytes = match column.constraint.constraint() {
Some(Constraint::MaxBytes(max)) => Some(usize::from(*max)),
_ => None,
};
(0..limit).find(|&row| {
if view.none_at(row) {
return !target.is_option() && filled.binary_search(&row).is_err();
}
match (target.inner_type(), max_bytes) {
(ValueType::Utf8, Some(max)) => view.get_str(row).is_some_and(|text| text.len() > max),
(ValueType::Blob, Some(max)) => view.get_bytes(row).is_some_and(|bytes| bytes.len() > max),
_ => false,
}
})
}
fn constraint_error(column: &Column, mut value: Value, name: &str, row: usize) -> Error {
let mut error = match column.constraint.coerce(&mut value) {
Err(error) => error,
Ok(()) => internal_error!("bulk column {} passed the constraint its scan refused", column.name),
};
error.0.fragment = Fragment::internal(&column.name);
with_row_note(error, name, row)
}
fn key_column_index(series: &Series) -> Result<usize> {
let key_col_name = series.key.column();
series.columns
.iter()
.position(|c| c.name == key_col_name)
.ok_or_else(|| internal_error!("series {} key column {} not found", series.name, key_col_name))
}
fn enforce_write_policies(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
namespace: &Namespace,
object: &str,
target: PolicyTargetType,
columns: &[(FieldRef, ArrayRef)],
) -> Result<()> {
let symbols = SymbolTable::new();
PolicyEvaluator::new(writer.services, &symbols).enforce_write_policies(
&mut Transaction::Command(txn),
namespace.name(),
object,
DataOp::Insert,
&batch(columns.to_vec())?,
target,
)
}
fn stamp<B: SourceRowBuilder>(row: &mut B, times: &RowTime<'_>, index: usize, now: DateTime) -> Result<()> {
row.set_timestamps(now, now);
if let Some(time) = times.inserted(index, row.as_slice(), now)? {
row.set_time(time);
}
Ok(())
}
fn execute_table_insert<V: ValidationMode>(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
pending: &PendingTableInsert,
) -> Result<TableInsertResult> {
let catalog = writer.catalog;
let (namespace, table) = resolve_table(catalog, txn, pending)?;
let shape = get_or_create_table_shape(catalog, &table, &mut Transaction::Command(txn))?;
let (columns, _) =
coerce_columns(&pending.rows, &pending.batches, &table.columns, None, &table.name, &writer.coercer)?;
let rows = pending.row_count();
if rows == 0 {
return Ok(empty_table_result(pending));
}
enforce_write_policies(writer, txn, &namespace, &table.name, PolicyTargetType::Table, &columns)?;
let filled = auto_increment_rows(&table.columns, &columns)?;
check_rows::<V>(
&Checks {
name: &table.name,
columns: &table.columns,
filled: &filled,
series: None,
time: &table.time,
},
&columns,
None,
rows,
)?;
let fills = filled.into_iter().enumerate().filter(|(_, rows)| !rows.is_empty()).collect();
let mut cast = [CastColumns::with_fills(columns, fills, rows)];
intern_dictionary_columns(catalog, &mut Transaction::Command(txn), &table.columns, None, &mut cast)?;
fill_sequences(catalog, &mut Transaction::Command(txn), &table.columns, table.id.into(), &mut cast)?;
let mut encoded = EncodedTableRows {
rows: (0..rows).map(|_| shape.allocate_table()).collect(),
..EncodedTableRows::default()
};
cast[0].write(&shape, &mut encoded.rows)?;
let now = writer.now;
let times = RowTime::new(
&table.name,
&table.time,
&shape,
populator_index(&table.time, &shape).map(|index| cast[0].view(index)).transpose()?,
);
for (index, row) in encoded.rows.iter_mut().enumerate() {
stamp(row, ×, index, now)?;
}
table_batch_partitions(&table, &cast[0], &mut encoded)?;
write_table_rows(
catalog,
&mut Transaction::Command(txn),
&table,
&shape,
&mut encoded,
&Violation::BulkRow(&table.name),
)?;
Ok(TableInsertResult {
namespace: pending.namespace.clone(),
table: pending.table.clone(),
inserted: rows as u64,
})
}
fn auto_increment_rows(columns: &[Column], cast: &[(FieldRef, ArrayRef)]) -> Result<Vec<Vec<usize>>> {
columns.iter()
.zip(cast)
.map(|(column, cast)| {
if !column.auto_increment {
return Ok(Vec::new());
}
let view = ColumnView::try_from(cast)?;
Ok((0..view.len()).filter(|&row| view.none_at(row)).collect())
})
.collect()
}
#[inline]
fn empty_table_result(pending: &PendingTableInsert) -> TableInsertResult {
TableInsertResult {
namespace: pending.namespace.clone(),
table: pending.table.clone(),
inserted: 0,
}
}
fn resolve_namespace(catalog: &Catalog, txn: &mut CommandTransaction, name: &str) -> Result<Namespace> {
catalog.find_namespace_by_name(&mut Transaction::Command(txn), name)?.ok_or_else(|| {
CatalogError::NotFound {
kind: CatalogObjectKind::Namespace,
namespace: name.to_string(),
name: String::new(),
fragment: Fragment::None,
}
.into()
})
}
fn has_keyed_table(
catalog: &Catalog,
txn: &mut CommandTransaction,
pending_tables: &[PendingTableInsert],
) -> Result<bool> {
for pending in pending_tables {
let (_, table) = resolve_table(catalog, txn, pending)?;
if get_primary_key(catalog, &mut Transaction::Command(txn), &table)?.is_some() {
return Ok(true);
}
}
Ok(false)
}
fn resolve_table(
catalog: &Catalog,
txn: &mut CommandTransaction,
pending: &PendingTableInsert,
) -> Result<(Namespace, Table)> {
let namespace = resolve_namespace(catalog, txn, &pending.namespace)?;
let table = catalog
.find_table_by_name(&mut Transaction::Command(txn), namespace.id(), &pending.table)?
.ok_or_else(|| -> Error {
CatalogError::NotFound {
kind: CatalogObjectKind::Table,
namespace: pending.namespace.to_string(),
name: pending.table.to_string(),
fragment: Fragment::None,
}
.into()
})?;
Ok((namespace, table))
}
fn execute_ringbuffer_insert<V: ValidationMode>(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
pending: &PendingRingBufferInsert,
) -> Result<RingBufferInsertResult> {
let catalog = writer.catalog;
let (namespace, ringbuffer) = resolve_ringbuffer(catalog, txn, pending)?;
let shape = get_or_create_ringbuffer_shape(catalog, &ringbuffer, &mut Transaction::Command(txn))?;
let (columns, _) = coerce_columns(
&pending.rows,
&pending.batches,
&ringbuffer.columns,
None,
&ringbuffer.name,
&writer.coercer,
)?;
let rows = pending.row_count();
let inserted = if rows == 0 {
0
} else {
insert_ringbuffer_rows::<V>(writer, txn, &namespace, &ringbuffer, &shape, columns, rows)?
};
Ok(RingBufferInsertResult {
namespace: pending.namespace.clone(),
ringbuffer: pending.ringbuffer.clone(),
inserted,
})
}
#[inline]
fn resolve_ringbuffer(
catalog: &Catalog,
txn: &mut CommandTransaction,
pending: &PendingRingBufferInsert,
) -> Result<(Namespace, RingBuffer)> {
let namespace = resolve_namespace(catalog, txn, &pending.namespace)?;
let ringbuffer = catalog
.find_ringbuffer_by_name(&mut Transaction::Command(txn), namespace.id(), &pending.ringbuffer)?
.ok_or_else(|| -> Error {
CatalogError::NotFound {
kind: CatalogObjectKind::RingBuffer,
namespace: pending.namespace.to_string(),
name: pending.ringbuffer.to_string(),
fragment: Fragment::None,
}
.into()
})?;
Ok((namespace, ringbuffer))
}
fn insert_ringbuffer_rows<V: ValidationMode>(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
namespace: &Namespace,
ringbuffer: &RingBuffer,
shape: &RowShape,
columns: Vec<(FieldRef, ArrayRef)>,
rows: usize,
) -> Result<u64> {
let catalog = writer.catalog;
enforce_write_policies(writer, txn, namespace, &ringbuffer.name, PolicyTargetType::RingBuffer, &columns)?;
check_rows::<V>(
&Checks {
name: &ringbuffer.name,
columns: &ringbuffer.columns,
filled: &vec![Vec::new(); ringbuffer.columns.len()],
series: None,
time: &ringbuffer.time,
},
&columns,
None,
rows,
)?;
let mut cast = [CastColumns::new(columns, rows)];
intern_dictionary_columns(catalog, &mut Transaction::Command(txn), &ringbuffer.columns, None, &mut cast)?;
let mut built: Vec<_> = (0..rows).map(|_| shape.allocate_ringbuffer()).collect();
cast[0].write(shape, &mut built)?;
let batch = ringbuffer_batch_partitions(ringbuffer, &cast[0])?;
let mut verified = HashMap::new();
resolve_partitions(
&mut Transaction::Command(txn),
ObjectId::ringbuffer(ringbuffer.id),
&batch.distinct,
&mut verified,
)?;
let now = writer.now;
let times = RowTime::new(
&ringbuffer.name,
&ringbuffer.time,
shape,
populator_index(&ringbuffer.time, shape).map(|index| cast[0].view(index)).transpose()?,
);
let mut entries = Vec::with_capacity(rows);
for (row, mut builder) in built.into_iter().enumerate() {
stamp(&mut builder, ×, row, now)?;
entries.push(builder.freeze_bytes());
}
let mut cache = HashMap::new();
let inserted = insert_ringbuffer_chunks(
catalog,
&mut Transaction::Command(txn),
ringbuffer,
shape,
&entries,
&batch.rows,
&verified,
&mut cache,
None,
)?;
save_all_partition_metadata(catalog, &mut Transaction::Command(txn), ringbuffer, &cache)?;
Ok(inserted)
}
fn execute_series_insert<V: ValidationMode>(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
pending: &PendingSeriesInsert,
) -> Result<SeriesInsertResult> {
let catalog = writer.catalog;
let (namespace, series) = resolve_series(catalog, txn, pending)?;
let mut metadata_by_partition: HashMap<Partition, SeriesPartitionMetadata> = HashMap::new();
let mut inserted_keys: HashMap<Partition, Vec<u64>> = HashMap::new();
let shape = get_or_create_series_shape(catalog, &series, &mut Transaction::Command(txn))?;
let sumtype = series.tag.map(|id| catalog.get_sumtype(&mut Transaction::Command(txn), id)).transpose()?;
let (columns, tags) = coerce_columns(
&pending.rows,
&pending.batches,
&series.columns,
sumtype.as_ref(),
&series.name,
&writer.coercer,
)?;
let rows = pending.row_count();
let inserted = if rows == 0 {
0
} else {
enforce_write_policies(writer, txn, &namespace, &series.name, PolicyTargetType::Series, &columns)?;
let checked = check_rows::<V>(
&Checks {
name: &series.name,
columns: &series.columns,
filled: &vec![Vec::new(); series.columns.len()],
series: Some(&series),
time: &series.time,
},
&columns,
tags,
rows,
)?;
insert_series_rows(
writer,
txn,
&series,
&shape,
columns,
checked,
&mut metadata_by_partition,
&mut inserted_keys,
)?
};
let now = writer.now;
for (partition, metadata) in metadata_by_partition {
let keys = inserted_keys.remove(&partition).unwrap_or_default();
write_inserted_series_metadata(
catalog,
&mut Transaction::Command(txn),
&series,
partition,
metadata,
keys,
now,
V::VALIDATED,
)?;
if !V::VALIDATED {
txn.rewrite_on(
&SeriesPartitionMetadataKey::new(series.id, partition).into(),
renumber_series_rows(StorageId::series(series.id), partition),
)?;
}
}
Ok(SeriesInsertResult {
namespace: pending.namespace.clone(),
series: pending.series.clone(),
inserted,
})
}
#[inline]
fn resolve_series(
catalog: &Catalog,
txn: &mut CommandTransaction,
pending: &PendingSeriesInsert,
) -> Result<(Namespace, Series)> {
let namespace = resolve_namespace(catalog, txn, &pending.namespace)?;
let series = catalog
.find_series_by_name(&mut Transaction::Command(txn), namespace.id(), &pending.series)?
.ok_or_else(|| -> Error {
CatalogError::NotFound {
kind: CatalogObjectKind::Series,
namespace: pending.namespace.to_string(),
name: pending.series.to_string(),
fragment: Fragment::None,
}
.into()
})?;
Ok((namespace, series))
}
fn insert_series_rows(
writer: &Writer<'_>,
txn: &mut CommandTransaction,
series: &Series,
shape: &RowShape,
columns: Vec<(FieldRef, ArrayRef)>,
checked: Checked,
metadata_by_partition: &mut HashMap<Partition, SeriesPartitionMetadata>,
inserted_keys: &mut HashMap<Partition, Vec<u64>>,
) -> Result<u64> {
let catalog = writer.catalog;
let rows = checked.keys.len();
let key_index = key_column_index(series)?;
let mut cast = [CastColumns::new(columns, rows)];
intern_dictionary_columns(
catalog,
&mut Transaction::Command(txn),
&series.columns,
Some(&series.key),
&mut cast,
)?;
let [cast] = cast;
let key_column = series.key_column_data(checked.keys.clone());
let mut built: Vec<_> = (0..rows).map(|_| shape.allocate_series()).collect();
let mut views = Vec::with_capacity(series.columns.len());
views.push(ColumnView::try_from(&key_column)?);
for index in 0..series.columns.len() {
if index != key_index {
views.push(cast.view(index)?);
}
}
shape.write_columns(&mut built, &views)?;
let now = writer.now;
let times = RowTime::new(
&series.name,
&series.time,
shape,
populator_index(&series.time, shape).map(|index| views[index].clone()),
);
let batch = series_batch_partitions(series, &cast, Some(&views[0]))?;
resolve_partitions(
&mut Transaction::Command(txn),
ObjectId::Series(series.id),
&batch.distinct,
&mut HashMap::new(),
)?;
let assigned = assign_series_keys(
catalog,
&mut Transaction::Command(txn),
series,
&batch.rows,
checked.keys.into_iter().map(Some).collect(),
&checked.tags,
metadata_by_partition,
|_| Err(internal_error!("bulk insert into series {} kept a row without a key", series.name)),
)?;
for (row, &key) in assigned.keys.iter().enumerate() {
inserted_keys.entry(batch.rows.get(row).copied().unwrap_or_default()).or_default().push(key);
}
for (row, builder) in built.iter_mut().enumerate() {
stamp(builder, ×, row, now)?;
}
let written = write_series_rows(
&mut Transaction::Command(txn),
series,
shape,
SeriesRows {
keys: assigned.storage,
partitions: &batch.rows,
rows: built,
},
)?;
Ok(written.len() as u64)
}
fn renumber_series_rows(storage: StorageId, partition: Partition) -> Rewrite {
Arc::new(move |deltas, at_base, at_latest| {
let counter = |bytes: Option<&EncodedBytes>| -> Result<u64> {
Ok(match bytes {
Some(bytes) => {
decode_series_partition_metadata(EncodedPodRow::view(bytes))?.sequence_counter
}
None => 0,
})
};
let base = counter(at_base)?;
let gap = counter(at_latest)?.checked_sub(base).ok_or_else(|| {
internal_error!(
"series {:?} partition counter fell below the bulk's snapshot counter {}",
storage,
base
)
})?;
if gap == 0 {
return Ok(());
}
for delta in deltas.make_mut() {
if let Delta::Set {
key,
..
} = delta
{
match key {
TaggedKey::SeriesRow(row) if row.storage == storage && row.sequence > base => {
row.sequence += gap;
}
TaggedKey::PartitionedSeriesRow(row)
if row.storage == storage
&& row.partition == partition && row.sequence > base =>
{
row.sequence += gap;
}
_ => {}
}
}
}
Ok(())
})
}
fn parse_qualified_name(qualified_name: &str) -> (String, String) {
if let Some((ns, name)) = qualified_name.rsplit_once("::") {
(ns.to_string(), name.to_string())
} else {
("default".to_string(), qualified_name.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_qualified_name_simple() {
assert_eq!(parse_qualified_name("table"), ("default".to_string(), "table".to_string()));
}
#[test]
fn parse_qualified_name_single_namespace() {
assert_eq!(parse_qualified_name("ns::table"), ("ns".to_string(), "table".to_string()));
}
#[test]
fn parse_qualified_name_nested_namespace() {
assert_eq!(parse_qualified_name("a::b::table"), ("a::b".to_string(), "table".to_string()));
}
#[test]
fn parse_qualified_name_deeply_nested_namespace() {
assert_eq!(parse_qualified_name("a::b::c::table"), ("a::b::c".to_string(), "table".to_string()));
}
#[test]
fn parse_qualified_name_empty_string() {
assert_eq!(parse_qualified_name(""), ("default".to_string(), "".to_string()));
}
}