use crate::protocol::sql::statement_is_plsql;
use crate::protocol::thin::BindValue;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ExecutemanyManagerError {
ZeroBatchSize,
RowCountOverflow,
}
impl std::fmt::Display for ExecutemanyManagerError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ZeroBatchSize => f.write_str("batch_size must be greater than zero"),
Self::RowCountOverflow => f.write_str("executemany row count exceeds u32"),
}
}
}
impl std::error::Error for ExecutemanyManagerError {}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutemanyManager {
total_rows: u32,
batch_size: u32,
num_rows: u32,
message_offset: u32,
chunk_ends: Vec<u32>,
}
impl ExecutemanyManager {
pub fn new(total_rows: usize, batch_size: u32) -> Result<Self, ExecutemanyManagerError> {
Self::with_chunks(total_rows, batch_size, Vec::new())
}
pub fn with_chunks(
total_rows: usize,
batch_size: u32,
chunk_lengths: Vec<usize>,
) -> Result<Self, ExecutemanyManagerError> {
let total_rows =
u32::try_from(total_rows).map_err(|_| ExecutemanyManagerError::RowCountOverflow)?;
if batch_size == 0 {
return Err(ExecutemanyManagerError::ZeroBatchSize);
}
let mut chunk_ends = Vec::with_capacity(chunk_lengths.len());
let mut acc: u32 = 0;
for len in chunk_lengths {
acc = acc.saturating_add(
u32::try_from(len).map_err(|_| ExecutemanyManagerError::RowCountOverflow)?,
);
chunk_ends.push(acc);
}
let mut manager = Self {
total_rows,
batch_size,
num_rows: 0,
message_offset: 0,
chunk_ends,
};
manager.num_rows = manager.batch_len_from(0);
Ok(manager)
}
fn batch_len_from(&self, offset: u32) -> u32 {
let remaining = self.total_rows.saturating_sub(offset);
let mut len = remaining.min(self.batch_size);
if let Some(next_end) = self.chunk_ends.iter().find(|&&end| end > offset) {
len = len.min(next_end.saturating_sub(offset));
}
len
}
pub fn num_rows(&self) -> u32 {
self.num_rows
}
pub fn message_offset(&self) -> u32 {
self.message_offset
}
pub fn next_batch(&mut self) {
self.message_offset = self.message_offset.saturating_add(self.num_rows);
self.num_rows = self.batch_len_from(self.message_offset);
}
}
fn bind_value_is_output(value: &BindValue) -> bool {
matches!(
value,
BindValue::Output { .. } | BindValue::ReturnOutput { .. } | BindValue::ObjectOutput { .. }
)
}
pub fn bind_rows_need_iterative_plsql(statement: &str, bind_rows: &[Vec<BindValue>]) -> bool {
statement_is_plsql(statement)
&& bind_rows
.iter()
.any(|row| row.iter().any(bind_value_is_output))
}
#[cfg(test)]
mod tests {
use super::*;
fn output_bind() -> BindValue {
BindValue::Output {
ora_type_num: 1,
csfrm: 1,
buffer_size: 32,
}
}
fn input_bind() -> BindValue {
BindValue::Number("1".to_string())
}
#[test]
fn plsql_with_output_bind_is_iterative() {
let rows = vec![vec![input_bind(), output_bind()]];
assert!(bind_rows_need_iterative_plsql(
"begin :res := pkg.fn(:arg); end;",
&rows
));
}
#[test]
fn plsql_with_output_bind_in_any_row_is_iterative() {
let rows = vec![
vec![input_bind(), input_bind()],
vec![input_bind(), output_bind()],
];
assert!(bind_rows_need_iterative_plsql(
"begin proc(:a, :b); end;",
&rows
));
}
#[test]
fn plsql_without_output_binds_is_batched() {
let rows = vec![vec![input_bind()], vec![input_bind()]];
assert!(!bind_rows_need_iterative_plsql(
"begin proc(:a); end;",
&rows
));
}
#[test]
fn dml_with_output_binds_stays_batched() {
let rows = vec![vec![input_bind(), output_bind()]];
assert!(!bind_rows_need_iterative_plsql(
"insert into t (c) values (:c) returning id into :id",
&rows
));
assert!(!bind_rows_need_iterative_plsql(
"update t set c = :c returning id into :id",
&rows
));
}
#[test]
fn empty_rows_are_batched() {
assert!(!bind_rows_need_iterative_plsql("begin null; end;", &[]));
}
#[test]
fn return_and_object_output_binds_count_as_outputs() {
let return_rows = vec![vec![BindValue::ReturnOutput {
ora_type_num: 1,
csfrm: 1,
buffer_size: 32,
}]];
assert!(bind_rows_need_iterative_plsql(
"declare begin :x := 1; end;",
&return_rows
));
}
fn drain(mut manager: ExecutemanyManager) -> Vec<(u32, u32)> {
let mut batches = Vec::new();
while manager.num_rows() > 0 {
batches.push((manager.message_offset(), manager.num_rows()));
manager.next_batch();
}
batches
}
#[test]
fn batches_split_evenly() {
let manager = ExecutemanyManager::new(6, 2).unwrap();
assert_eq!(drain(manager), vec![(0, 2), (2, 2), (4, 2)]);
}
#[test]
fn final_batch_carries_the_remainder() {
let manager = ExecutemanyManager::new(7, 3).unwrap();
assert_eq!(drain(manager), vec![(0, 3), (3, 3), (6, 1)]);
}
#[test]
fn single_batch_when_batch_size_covers_all_rows() {
let manager = ExecutemanyManager::new(4, 10).unwrap();
assert_eq!(manager.num_rows(), 4);
assert_eq!(manager.message_offset(), 0);
assert_eq!(drain(manager), vec![(0, 4)]);
}
#[test]
fn zero_total_rows_yields_no_batches() {
let manager = ExecutemanyManager::new(0, 5).unwrap();
assert_eq!(manager.num_rows(), 0);
assert!(drain(manager).is_empty());
}
#[test]
fn zero_batch_size_is_rejected() {
assert_eq!(
ExecutemanyManager::new(3, 0),
Err(ExecutemanyManagerError::ZeroBatchSize)
);
}
#[test]
fn row_count_overflow_is_rejected() {
let too_many = (u32::MAX as usize) + 1;
assert_eq!(
ExecutemanyManager::new(too_many, 1),
Err(ExecutemanyManagerError::RowCountOverflow)
);
}
#[test]
fn batches_never_cross_a_chunk_boundary() {
let manager = ExecutemanyManager::with_chunks(6, 2, vec![3, 3]).unwrap();
assert_eq!(drain(manager), vec![(0, 2), (2, 1), (3, 2), (5, 1)]);
}
#[test]
fn chunk_boundary_aligned_with_batch_size_does_not_split() {
let manager = ExecutemanyManager::with_chunks(4, 2, vec![2, 2]).unwrap();
assert_eq!(drain(manager), vec![(0, 2), (2, 2)]);
}
}