mod allocation;
pub mod context;
mod persistence;
mod resolution;
use super::{
session::{NontransactionalSequenceValue, SessionSequenceValue},
SequenceState,
};
use context::SequenceValueContext;
use resolution::{BoundSequenceValue, ValueAccess};
use uqa_core::RelationIdentity;
use uqa_sql::catalog::sequence_functions::value_error::SequenceValueError;
struct NextvalTarget {
name: String,
relation: RelationIdentity,
object_id: [u8; 16],
state: SequenceState,
temporary: bool,
}
impl SequenceValueContext<'_> {
pub fn nextval(&self, name: &str) -> Result<i64, SequenceValueError> {
self.transactions
.with_value_transaction(Box::new(|| self.nextval_locked(name)))
}
fn nextval_locked(&self, name: &str) -> Result<i64, SequenceValueError> {
let bound = self.bind_sequence_value_reference(name)?;
loop {
self.runtime.cancellation().check()?;
let target = self.lock_sequence_value_target(&bound, ValueAccess::Next)?;
if self.runtime.current_transaction_is_read_only() && !target.temporary {
return Err(SequenceValueError::ReadOnly("nextval"));
}
let mut caches = self.runtime.caches();
if let Some((current, autonomous)) = Self::take_cached_nextval(&target, &mut caches)? {
drop(caches);
self.complete_nextval(
&target.relation,
target.object_id,
current,
target.state,
autonomous,
);
return Ok(current);
}
let Some((reservation, autonomous)) = self.reserve_nextval_block(&target)? else {
drop(caches);
continue;
};
let physical =
self.install_nextval_reservation(&target, reservation, autonomous, &mut caches)?;
drop(caches);
self.complete_nextval(
&target.relation,
target.object_id,
reservation.first_value,
physical,
autonomous,
);
return Ok(reservation.first_value);
}
}
pub fn currval(&self, name: &str) -> Result<i64, SequenceValueError> {
self.transactions
.with_value_transaction(Box::new(|| self.currval_locked(name)))
}
fn currval_locked(&self, name: &str) -> Result<i64, SequenceValueError> {
let bound = self.bind_sequence_value_reference(name)?;
let target = self.lock_sequence_value_target(&bound, ValueAccess::Current)?;
self.runtime
.session_read()
.currvals()
.values()
.find(|current| current.object_id == target.object_id)
.map(|current| current.value)
.ok_or(SequenceValueError::CurrvalUndefined(target.relation.name))
}
pub fn lastval(&self) -> Result<i64, SequenceValueError> {
self.transactions
.with_value_transaction(Box::new(|| self.lastval_locked()))
}
fn lastval_locked(&self) -> Result<i64, SequenceValueError> {
let snapshot = self.read_snapshot()?;
let session = self.runtime.session_read();
let last = session.last().ok_or(SequenceValueError::LastvalUndefined)?;
let object_id = last.object_id;
let value = session
.currvals()
.values()
.find(|current| current.object_id == object_id)
.map(|current| current.value)
.ok_or(SequenceValueError::LastvalUndefined)?;
drop(session);
let relation = snapshot
.object_ids
.iter()
.find_map(|(relation, candidate)| (*candidate == object_id).then(|| relation.clone()))
.ok_or(SequenceValueError::LastvalUndefined)?;
self.lock_sequence_value_target(
&BoundSequenceValue {
name: relation.qualified_name(),
object_id,
},
ValueAccess::Current,
)?;
Ok(value)
}
pub fn setval(
&self,
name: &str,
value: i64,
is_called: bool,
) -> Result<i64, SequenceValueError> {
self.transactions
.with_value_transaction(Box::new(|| self.setval_locked(name, value, is_called)))
}
fn setval_locked(
&self,
name: &str,
value: i64,
is_called: bool,
) -> Result<i64, SequenceValueError> {
let bound = self.bind_sequence_value_reference(name)?;
loop {
self.runtime.cancellation().check()?;
let target = self.lock_sequence_value_target(&bound, ValueAccess::Set)?;
if self.runtime.current_transaction_is_read_only() && !target.temporary {
return Err(SequenceValueError::ReadOnly("setval"));
}
let (min, max) = (target.state.min_value, target.state.max_value);
if !(min..=max).contains(&value) {
return Err(SequenceValueError::SetvalOutOfBounds {
name: target.name,
value,
min,
max,
});
}
if let Some(value) = self.setval_target(target, value, is_called)? {
return Ok(value);
}
}
}
fn setval_target(
&self,
target: NextvalTarget,
value: i64,
is_called: bool,
) -> Result<Option<i64>, SequenceValueError> {
let NextvalTarget {
name,
relation,
object_id,
state: previous,
temporary,
} = target;
let persisted = self.mutate_persistent_value(
temporary,
&relation,
object_id,
"persist sequence value",
|catalog| {
catalog.set_sequence_value(
&name,
object_id,
previous.definition_generation,
value,
is_called,
0,
)
},
)?;
let autonomous = match persisted {
Some((uqa_storage::SequenceSetValueResult::Set(_), autonomous)) => Some(autonomous),
Some((uqa_storage::SequenceSetValueResult::DefinitionChanged, _)) => return Ok(None),
Some((uqa_storage::SequenceSetValueResult::Missing, _)) => {
return Err(SequenceValueError::Undefined(name))
}
None => None,
};
let mut seqs = self.runtime.states_write();
match seqs.get_mut(&relation) {
Some(seq) if seq.definition_generation == previous.definition_generation => {
seq.current = value;
seq.called = is_called;
seq.log_count = 0;
}
_ if autonomous.is_some() => {}
Some(_) => return Ok(None),
None => return Err(SequenceValueError::Undefined(name)),
}
drop(seqs);
self.runtime
.caches()
.retain(|_, cache| cache.object_id != object_id);
if is_called {
let mut session = self.runtime.session_write();
session
.currvals_mut()
.retain(|_, current| current.object_id != object_id);
session
.currvals_mut()
.insert(relation.clone(), SessionSequenceValue { object_id, value });
}
self.runtime.record_nontransactional_sequence_value(
previous.definition_generation,
NontransactionalSequenceValue {
object_id,
current: value,
called: is_called,
log_count: 0,
autonomous: autonomous.unwrap_or(false),
},
false,
);
Ok(Some(value))
}
}