use std::collections::BTreeMap;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::OnceLock;
use std::time::{Duration, Instant};
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use serde::de::{self, MapAccess, SeqAccess, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
mod build_identity;
mod derived_overload;
mod identity;
mod lifecycle_completion;
mod local_development;
mod mutation_lifecycle;
mod operation_locator;
mod performance;
mod replayable_source;
mod resource_spec;
mod shared_root_history;
mod state_outcomes;
mod workload_class;
pub use build_identity::*;
pub use derived_overload::*;
pub use identity::{
BackendAuthorityId, BackendAuthorityPair, ClusterIdentity, ClusterStorageId, CollectionId, CollectionIndexRef,
FenceToken, IdentityError, IndexAddress, IndexId, IndexName, NodeIncarnation, ObjectIncarnation, PrimaryTerm,
ProtectionSnapshotId, RecoveryAttemptId, RecoveryEra, RecoveryGeneration, ShardKey,
};
pub use lifecycle_completion::*;
pub use local_development::{LocalDevelopmentInstanceId, LocalDevelopmentInstanceIdError};
pub use mutation_lifecycle::*;
pub use operation_locator::{
ARTIFACT_CHUNK_ALIGNMENT, ARTIFACT_LOCATOR_FORMAT_VERSION, ArtifactLocatorV1, ArtifactReadRangeV1,
MAX_ARTIFACT_CHUNK_BYTES, OPERATION_LOCATOR_FORMAT_VERSION, OperationAuthorityKindV1, OperationLocatorError,
OperationLocatorV1, VerifiedArtifactChunkV1, verify_artifact_chunk_v1,
};
pub use performance::{
BoundedStageSnapshotV2, PerformanceObservationV2, PerformanceStageReadingV2, PerformanceStageV2,
bounded_stage_snapshot_v2, record_performance_stage_v2,
};
pub use replayable_source::*;
pub use resource_spec::*;
pub use shared_root_history::*;
pub use state_outcomes::*;
pub use workload_class::*;
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AwsPartition {
Aws,
AwsCn,
AwsUsGov,
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum AwsBucketKind {
GeneralPurpose,
Directory { availability_zone_id: String },
}
impl AwsBucketKind {
#[must_use]
pub const fn is_directory(&self) -> bool {
matches!(self, Self::Directory { .. })
}
#[must_use]
pub fn availability_zone_id(&self) -> Option<&str> {
match self {
Self::GeneralPurpose => None,
Self::Directory { availability_zone_id } => Some(availability_zone_id),
}
}
}
macro_rules! semantic_number {
($name:ident, $representation:ty, $documentation:literal) => {
#[doc = $documentation]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct $name($representation);
impl $name {
#[must_use]
pub const fn new(value: $representation) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> $representation {
self.0
}
#[must_use]
pub const fn checked_next(self) -> Option<Self> {
match self.0.checked_add(1) {
Some(value) => Some(Self(value)),
None => None,
}
}
}
impl Display for $name {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.0, formatter)
}
}
};
}
semantic_number!(
ShardId,
u16,
"A semantic identifier for a shard in deterministic models and protocols."
);
semantic_number!(Epoch, u32, "A monotonically increasing shard-writer fencing epoch.");
semantic_number!(
SeqNo,
u64,
"A per-shard monotonically increasing operation sequence number."
);
semantic_number!(
ManifestGeneration,
u64,
"A monotonically increasing immutable-manifest generation."
);
semantic_number!(
SealedRootProofGeneration,
u64,
"A monotonically increasing sealed-root proof receipt generation. Zero is never issued, so it is the reserved never-proved value."
);
semantic_number!(
ManifestGraphProofGeneration,
u64,
"A monotonically increasing post-recovery manifest-graph proof generation. Zero is never issued."
);
semantic_number!(
RoutingHash,
u32,
"The stored routing hash used for shard-range selection."
);
semantic_number!(
IdHash,
u128,
"The mutable-index document identity hash used by visibility overlays."
);
semantic_number!(SegmentId, u128, "A stable immutable segment identifier.");
semantic_number!(
StateIncarnation,
u64,
"A fenced lifetime identity for one state primary/projection pair."
);
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct VectorFieldId(u32);
impl VectorFieldId {
pub const fn new(value: u32) -> Result<Self, VectorFieldIdError> {
if value == 0 {
return Err(VectorFieldIdError);
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> u32 {
self.0
}
}
impl Display for VectorFieldId {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.0, formatter)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VectorFieldIdError;
impl Display for VectorFieldIdError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("vector field identity must be nonzero")
}
}
impl Error for VectorFieldIdError {}
impl<'de> Deserialize<'de> for VectorFieldId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
Self::new(u32::deserialize(deserializer)?).map_err(de::Error::custom)
}
}
pub const RESERVED_SEGMENT_ID_NAMESPACE_START: u128 = 1_u128 << 127;
#[must_use]
pub const fn is_reserved_segment_id(segment_id: SegmentId) -> bool {
segment_id.get() >= RESERVED_SEGMENT_ID_NAMESPACE_START
}
pub fn next_ordinary_segment_id<I>(current: I) -> Option<SegmentId>
where
I: IntoIterator<Item = SegmentId>,
{
let maximum = current
.into_iter()
.filter(|segment_id| !is_reserved_segment_id(*segment_id))
.max();
let next = maximum.map_or(Some(SegmentId::new(1)), SegmentId::checked_next)?;
(!is_reserved_segment_id(next)).then_some(next)
}
semantic_number!(
ChunkId,
u64,
"A node-local monotonically increasing WAL chunk identifier."
);
semantic_number!(NodeId, u64, "A semantic indexer-node identifier.");
semantic_number!(
ControllerEpoch,
u64,
"A nonzero monotonically increasing placement-controller claim epoch."
);
semantic_number!(
NodeLifecycleGeneration,
u64,
"A monotonically increasing generation of one node lifecycle record."
);
semantic_number!(
CredentialGeneration,
u64,
"A monotonically increasing workload-credential binding generation."
);
semantic_number!(
BootstrapGeneration,
u64,
"A monotonically increasing irreversible cluster-bootstrap generation."
);
semantic_number!(
MembershipGeneration,
u64,
"A monotonically increasing external Raft membership generation."
);
semantic_number!(
PlacementProgressGeneration,
u64,
"A monotonically increasing committed placement reconciliation progress generation."
);
semantic_number!(
RangeTransitionGeneration,
u64,
"A monotonically increasing online split or merge transition generation."
);
semantic_number!(
AssignmentGeneration,
u64,
"A monotonically increasing generation for one exact shard-route assignment."
);
semantic_number!(
ReadLeaseGeneration,
u64,
"A monotonically increasing generation for one replicated generation-read lease."
);
semantic_number!(
SchemaGeneration,
u32,
"A monotonically increasing mapping/schema generation."
);
semantic_number!(
PolicyGeneration,
u64,
"A monotonically increasing authoritative index-policy generation."
);
semantic_number!(
ResourceRevision,
u64,
"A monotonically increasing revision of one index resource specification."
);
semantic_number!(
CrdGeneration,
u64,
"A nonzero Kubernetes metadata generation for one exact custom-resource incarnation."
);
semantic_number!(
CatalogManagementGeneration,
u64,
"A monotonically increasing catalog-owned CAS generation for API/CRD management ownership."
);
semantic_number!(
IndexRevision,
u64,
"A monotonically increasing revision of one immutable index incarnation's catalog record."
);
semantic_number!(
HistoryLedgerRevision,
u64,
"A monotonically increasing CAS revision of one index's retained-history ledger."
);
semantic_number!(
AdmissionGeneration,
u64,
"A monotonically increasing catalog admission fence for one index incarnation."
);
semantic_number!(
RoutingRevision,
u64,
"A monotonically increasing revision of one index's complete routing partition."
);
semantic_number!(
MemtableGeneration,
u64,
"A node-local active or frozen memtable generation."
);
semantic_number!(
BackendAuthority,
u64,
"A stable scheduler partition for one authoritative durability backend."
);
semantic_number!(
ClusterRevision,
u64,
"A monotonically increasing revision of the replicated cluster view."
);
semantic_number!(
RootRevision,
u64,
"A monotonically increasing revision of the authoritative durable root set."
);
semantic_number!(
LogicalMillis,
u64,
"Metastore-owned logical milliseconds used for leases without wall-clock dependence."
);
semantic_number!(
TaskCheckpointGeneration,
u64,
"A monotonically increasing CAS generation for one durable task checkpoint."
);
semantic_number!(
AdministrationRevision,
u64,
"A monotonically increasing revision of the collection-scoped administration catalog."
);
semantic_number!(
AliasRevision,
u64,
"A monotonically increasing revision of one exact alias record."
);
semantic_number!(
ComponentTemplateRevision,
u64,
"A monotonically increasing revision of one component-template incarnation."
);
semantic_number!(
IndexTemplateRevision,
u64,
"A monotonically increasing revision of one index-template incarnation."
);
semantic_number!(
IngestPipelineRevision,
u64,
"A monotonically increasing revision of one catalog-owned ingest pipeline."
);
semantic_number!(
DataStreamRevision,
u64,
"A monotonically increasing revision of one data-stream incarnation."
);
semantic_number!(
LifecyclePolicyRevision,
u64,
"A monotonically increasing revision of one lifecycle-policy incarnation."
);
semantic_number!(
SecurityPolicyRevision,
u64,
"A monotonically increasing revision of the sole coordinated security-policy snapshot."
);
semantic_number!(
CollectionSecurityGeneration,
u64,
"A monotonically increasing authorization generation for one Collection's public-security records."
);
semantic_number!(
QuotaPolicyGeneration,
u64,
"A monotonically increasing quota-policy generation scoped to one Collection."
);
semantic_number!(
SecurityRoleGeneration,
u64,
"A monotonically increasing generation of one Collection-local security role."
);
semantic_number!(
SecurityPrincipalGeneration,
u64,
"A monotonically increasing generation of one Collection-local security principal."
);
semantic_number!(
SecurityDirectSequence,
u64,
"A monotonically increasing position in the coordinated security direct-command replay journal."
);
semantic_number!(
ApiKeyGeneration,
u64,
"A monotonically increasing generation of one API-key verifier record."
);
pub const AUTHORIZATION_BINDING_FORMAT_VERSION: u16 = 1;
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AuthorizationBindingV1 {
pub format_version: u16,
pub recovery_era: RecoveryEra,
pub public_security_root_digest: [u8; 32],
pub ordering_revision: SecurityPolicyRevision,
pub collection_id: CollectionId,
pub collection_security_generation: CollectionSecurityGeneration,
pub principal_id: PrincipalId,
pub principal_generation: SecurityPrincipalGeneration,
pub api_key_id: ApiKeyId,
pub api_key_generation: ApiKeyGeneration,
pub role_generations_digest: [u8; 32],
pub quota_policy_generation: QuotaPolicyGeneration,
pub authorized_scope_digest: [u8; 32],
}
impl AuthorizationBindingV1 {
#[must_use]
pub fn is_valid(self) -> bool {
self.format_version == AUTHORIZATION_BINDING_FORMAT_VERSION
&& self.public_security_root_digest != [0; 32]
&& self.ordering_revision.get() != 0
&& self.collection_security_generation.get() != 0
&& self.principal_id.get() != 0
&& self.principal_generation.get() != 0
&& self.api_key_id.get() != 0
&& self.api_key_generation.get() != 0
&& self.role_generations_digest != [0; 32]
&& self.quota_policy_generation.get() != 0
&& self.authorized_scope_digest != [0; 32]
}
#[must_use]
pub fn same_local_authority(self, other: Self) -> bool {
self.same_generation_authority(other)
&& self.role_generations_digest == other.role_generations_digest
&& self.authorized_scope_digest == other.authorized_scope_digest
}
#[must_use]
pub fn same_generation_authority(self, other: Self) -> bool {
self.is_valid()
&& other.is_valid()
&& self.format_version == other.format_version
&& self.recovery_era == other.recovery_era
&& self.collection_id == other.collection_id
&& self.collection_security_generation == other.collection_security_generation
&& self.principal_id == other.principal_id
&& self.principal_generation == other.principal_generation
&& self.api_key_id == other.api_key_id
&& self.api_key_generation == other.api_key_generation
}
}
semantic_number!(
AuditSequence,
u64,
"A recovery-backed sequence in the tamper-evident security audit chain."
);
semantic_number!(
AuditRangeId,
u64,
"A recovery-backed identity for one fenced Gateway audit sequence range."
);
semantic_number!(
QuotaGrantId,
u64,
"A recovery-backed identity for one aggregate Gateway quota grant."
);
semantic_number!(
QuotaWindow,
u64,
"A coordinated logical quota-window identity that cannot be supplied by an adapter."
);
semantic_number!(
ProcessMonotonicMillis,
u64,
"Process-monotonic milliseconds used only for local freshness, timeout, and lifetime bounds."
);
semantic_number!(
UnixMillis,
u64,
"Unix milliseconds used only for certificate validity observations."
);
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct GatewayIncarnation([u8; 32]);
impl GatewayIncarnation {
pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
if value == [0; 32] {
return Err(IdentityError::Zero {
kind: "GatewayIncarnation",
});
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
impl<'de> Deserialize<'de> for GatewayIncarnation {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = <[u8; 32]>::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PublicTlsProfileV1 {
Http,
Grpc,
Native,
Resp,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PublicGatewayListenerV1 {
#[serde(rename = "api")]
LodestoneApi,
ElasticsearchHttp,
OtlpHttp,
JaegerHttp,
OtlpGrpc,
SubscriptionGrpc,
SubscriptionWebSocket,
Resp,
}
impl PublicGatewayListenerV1 {
#[must_use]
pub const fn required_tls_profile(self) -> PublicTlsProfileV1 {
match self {
Self::LodestoneApi => PublicTlsProfileV1::Native,
Self::ElasticsearchHttp | Self::OtlpHttp | Self::JaegerHttp | Self::SubscriptionWebSocket => {
PublicTlsProfileV1::Http
}
Self::OtlpGrpc | Self::SubscriptionGrpc => PublicTlsProfileV1::Grpc,
Self::Resp => PublicTlsProfileV1::Resp,
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum NodeRole {
MetastoreVoter,
Placement,
Orchestrator,
Gateway,
Indexer,
Searcher,
Compactor,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NodeToken {
pub recovery_era: RecoveryEra,
pub node_id: NodeId,
pub node_incarnation: NodeIncarnation,
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct KubernetesObjectUidDigest([u8; 32]);
impl KubernetesObjectUidDigest {
pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
if value == [0; 32] {
return Err(IdentityError::Zero {
kind: "KubernetesObjectUidDigest",
});
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
impl<'de> Deserialize<'de> for KubernetesObjectUidDigest {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = <[u8; 32]>::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
impl NodeToken {
pub fn new(
recovery_era: RecoveryEra,
node_id: NodeId,
node_incarnation: NodeIncarnation,
) -> Result<Self, IdentityError> {
if node_id.get() == 0 {
return Err(IdentityError::Zero { kind: "NodeId" });
}
Ok(Self {
recovery_era,
node_id,
node_incarnation,
})
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ControllerToken {
pub recovery_era: RecoveryEra,
pub controller_epoch: ControllerEpoch,
}
impl ControllerToken {
pub fn new(recovery_era: RecoveryEra, controller_epoch: ControllerEpoch) -> Result<Self, IdentityError> {
if controller_epoch.get() == 0 {
return Err(IdentityError::Zero {
kind: "ControllerEpoch",
});
}
Ok(Self {
recovery_era,
controller_epoch,
})
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct TaskId(u128);
impl TaskId {
#[must_use]
pub const fn new(value: u128) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> u128 {
self.0
}
#[must_use]
pub const fn checked_next(self) -> Option<Self> {
match self.0.checked_add(1) {
Some(value) => Some(Self(value)),
None => None,
}
}
}
impl Display for TaskId {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.0, formatter)
}
}
impl Serialize for TaskId {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.collect_str(self)
}
}
impl<'de> Deserialize<'de> for TaskId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
struct TaskIdVisitor;
impl Visitor<'_> for TaskIdVisitor {
type Value = TaskId;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("a decimal task-id string")
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
value
.parse::<u128>()
.map(TaskId)
.map_err(|_| E::custom("task id is outside the u128 domain"))
}
}
deserializer.deserialize_str(TaskIdVisitor)
}
}
macro_rules! semantic_decimal_identity {
($name:ident, $documentation:literal) => {
#[doc = $documentation]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct $name(TaskId);
impl $name {
#[must_use]
pub const fn new(value: u128) -> Self {
Self(TaskId::new(value))
}
#[must_use]
pub const fn get(self) -> u128 {
self.0.get()
}
}
impl Display for $name {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.0, formatter)
}
}
};
}
semantic_decimal_identity!(
ComponentTemplateId,
"A recovery-backed identity for one component-template incarnation."
);
semantic_decimal_identity!(
IndexTemplateId,
"A recovery-backed identity for one index-template incarnation."
);
semantic_decimal_identity!(
IngestPipelineId,
"A recovery-backed identity for one ingest-pipeline incarnation."
);
semantic_decimal_identity!(
DataStreamId,
"A recovery-backed identity for one data-stream incarnation."
);
semantic_decimal_identity!(
LifecyclePolicyId,
"A recovery-backed identity for one lifecycle-policy incarnation."
);
semantic_decimal_identity!(
ExternalKeyResolverCapabilityId,
"An opaque capability identity authorizing one external content-key resolver class."
);
semantic_decimal_identity!(
PrincipalId,
"A recovery-backed, never-reused security-principal identity."
);
semantic_decimal_identity!(ApiKeyId, "A recovery-backed, never-reused API-key identity.");
semantic_decimal_identity!(BackupId, "A recovery-backed, never-reused portable-backup identity.");
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct AdministrationActionId([u8; 32]);
impl AdministrationActionId {
pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
if value == [0; 32] {
return Err(IdentityError::Zero {
kind: "AdministrationActionId",
});
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
impl<'de> Deserialize<'de> for AdministrationActionId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = <[u8; 32]>::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct SecurityActionId([u8; 32]);
impl SecurityActionId {
pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
if value == [0; 32] {
return Err(IdentityError::Zero {
kind: "SecurityActionId",
});
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
impl<'de> Deserialize<'de> for SecurityActionId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = <[u8; 32]>::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct BackupActionId([u8; 32]);
impl BackupActionId {
pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
if value == [0; 32] {
return Err(IdentityError::Zero { kind: "BackupActionId" });
}
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
impl<'de> Deserialize<'de> for BackupActionId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = <[u8; 32]>::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct RoutingRangeEnd(u64);
impl RoutingRangeEnd {
pub const FULL: Self = Self(u32::MAX as u64 + 1);
pub fn new(value: u64) -> Result<Self, RoutingRangeError> {
if value == 0 || value > Self::FULL.0 {
Err(RoutingRangeError::InvalidEnd(value))
} else {
Ok(Self(value))
}
}
#[must_use]
pub const fn get(self) -> u64 {
self.0
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ShardDescriptor {
pub shard_id: ShardId,
pub epoch: Epoch,
pub hash_start: RoutingHash,
pub hash_end: RoutingRangeEnd,
}
impl ShardDescriptor {
pub fn new(
shard_id: ShardId,
epoch: Epoch,
hash_start: RoutingHash,
hash_end: RoutingRangeEnd,
) -> Result<Self, RoutingRangeError> {
if u64::from(hash_start.get()) >= hash_end.get() {
return Err(RoutingRangeError::EmptyOrReversed {
start: hash_start,
end: hash_end,
});
}
Ok(Self {
shard_id,
epoch,
hash_start,
hash_end,
})
}
#[must_use]
pub const fn contains(self, routing_hash: RoutingHash) -> bool {
let value = routing_hash.get() as u64;
value >= self.hash_start.get() as u64 && value < self.hash_end.get()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RoutingRangeError {
InvalidEnd(u64),
EmptyOrReversed { start: RoutingHash, end: RoutingRangeEnd },
}
impl Display for RoutingRangeError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidEnd(end) => write!(
formatter,
"routing range end {end} is outside 1..={}",
RoutingRangeEnd::FULL.get()
),
Self::EmptyOrReversed { start, end } => write!(
formatter,
"routing range start {start} must be below exclusive end {}",
end.get()
),
}
}
}
impl Error for RoutingRangeError {}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DurabilityClass {
RegionalObject,
ZonalLowLatency,
Relaxed,
Ephemeral,
}
impl DurabilityClass {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::RegionalObject => "regional_object",
Self::ZonalLowLatency => "zonal_low_latency",
Self::Relaxed => "relaxed",
Self::Ephemeral => "ephemeral",
}
}
pub const fn require_available(self) -> Result<(), DurabilityClassUnavailable> {
match self {
Self::RegionalObject | Self::ZonalLowLatency => Ok(()),
Self::Relaxed | Self::Ephemeral => Err(DurabilityClassUnavailable { class: self }),
}
}
pub const fn require_state_available(self) -> Result<(), DurabilityClassUnavailable> {
match self {
Self::RegionalObject | Self::ZonalLowLatency | Self::Ephemeral => Ok(()),
Self::Relaxed => Err(DurabilityClassUnavailable { class: self }),
}
}
}
impl Display for DurabilityClass {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DurabilityClassUnavailable {
pub class: DurabilityClass,
}
impl Display for DurabilityClassUnavailable {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
write!(
formatter,
"durability class '{}' is recognized but unavailable",
self.class
)
}
}
impl Error for DurabilityClassUnavailable {}
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum VisibilityMode {
#[default]
Applied,
WaitFor,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DurabilityBoundary {
Registered,
IncarnationMemory,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum DurabilityLoss {
Zero,
WholeIncarnation,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum WriteStage {
Validated,
Admitted,
Sequenced,
ObjectPut,
Registered,
AppliedToActiveMemtable,
MicroCommitted,
Sealed,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum IndexMode {
Mutable,
AppendLog,
State,
}
impl IndexMode {
#[must_use]
pub const fn has_mutation_visibility(self) -> bool {
matches!(self, Self::Mutable | Self::State)
}
#[must_use]
pub const fn is_state(self) -> bool {
matches!(self, Self::State)
}
}
pub const MAPPING_FORMAT_VERSION: u16 = 7;
pub const MAX_MAPPING_FIELDS: usize = 16_384;
pub const VECTOR_MAPPING_FORMAT_VERSION: u16 = 2;
pub const SPARSE_VECTOR_MAPPING_FORMAT_VERSION: u16 = 1;
pub const VECTOR_SET_MAPPING_FORMAT_VERSION: u16 = 1;
pub const MAX_VECTOR_MAPPING_FIELDS: usize = 8;
pub const MAX_VECTOR_DIMENSIONS: u16 = 16_384;
pub const MAX_VECTOR_BYTES_PER_FIELD: usize = 64 * 1_024;
pub const MAX_VECTOR_BYTES_PER_DOCUMENT: usize = 64 * 1_024;
pub const MAX_VECTOR_BYTES_PER_QUERY: usize = 64 * 1_024;
pub const MAX_VECTOR_BYTES_PER_REQUEST: usize = 16 * MAX_VECTOR_BYTES_PER_QUERY;
pub const MAX_SPARSE_VECTOR_DIMENSION: u32 = 16_777_215;
pub const MAX_SPARSE_VECTOR_NONZERO: usize = 4_096;
pub const MAX_SPARSE_VECTOR_BYTES: usize = 32 * 1_024;
pub const MAX_VECTOR_SET_DOCUMENT_VECTORS: usize = 128;
pub const MAX_VECTOR_SET_QUERY_VECTORS: usize = 32;
pub const MAX_VECTOR_SET_BYTES_PER_FIELD: usize = 64 * 1_024;
pub const MAX_VECTOR_SET_BYTES_PER_QUERY: usize = 64 * 1_024;
pub const MAX_FRESH_VECTOR_RECORDS: usize = 16_384;
pub const MAX_FRESH_VECTOR_BYTES: usize = 128 * 1_024 * 1_024;
pub const MAX_VECTOR_BUILD_SCRATCH_BYTES: u64 = 16 * 1_024 * 1_024 * 1_024;
pub const MAX_VECTOR_DECODED_CACHE_BYTES: u64 = 16 * 1_024 * 1_024 * 1_024;
pub const MAX_VECTOR_BYTES_PER_RESPONSE: usize = 64 * 1_024;
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DenseVectorElementType {
F32,
F16,
I8,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SparseVectorWeightType {
F32,
F16,
}
impl SparseVectorWeightType {
#[must_use]
pub const fn width_bytes(self) -> usize {
match self {
Self::F32 => 4,
Self::F16 => 2,
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SparseVectorEntry {
pub dimension: u32,
#[serde(deserialize_with = "deserialize_sparse_weight")]
pub weight: f64,
}
fn deserialize_sparse_weight<'de, D>(deserializer: D) -> Result<f64, D::Error>
where
D: Deserializer<'de>,
{
serde_json::Number::deserialize(deserializer)?
.to_string()
.parse::<f64>()
.map_err(de::Error::custom)
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SparseVectorValue {
#[serde(deserialize_with = "deserialize_bounded_sparse_entries")]
pub entries: Vec<SparseVectorEntry>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SparseVectorDefinition {
pub maximum_dimension: u32,
pub weight_type: SparseVectorWeightType,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct DecodedSparseVectorEntry {
pub dimension: u32,
pub weight: f32,
}
impl SparseVectorDefinition {
pub fn validate(self) -> Result<(), SparseVectorValueError> {
if self.maximum_dimension > MAX_SPARSE_VECTOR_DIMENSION {
return Err(SparseVectorValueError::InvalidMaximumDimension(self.maximum_dimension));
}
let maximum_bytes = MAX_SPARSE_VECTOR_NONZERO
.checked_mul(4_usize.saturating_add(self.weight_type.width_bytes()))
.ok_or(SparseVectorValueError::ByteLengthOverflow)?;
if maximum_bytes > MAX_SPARSE_VECTOR_BYTES {
return Err(SparseVectorValueError::FieldBytesExceeded(maximum_bytes));
}
Ok(())
}
}
impl SparseVectorValue {
pub fn decode(
&self,
definition: SparseVectorDefinition,
) -> Result<Vec<DecodedSparseVectorEntry>, SparseVectorValueError> {
definition.validate()?;
self.validate_bounded(definition, MAX_SPARSE_VECTOR_BYTES)?;
self.entries
.iter()
.enumerate()
.map(|(ordinal, entry)| {
let weight = entry.weight as f32;
if !weight.is_finite() {
return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
}
let weight = match definition.weight_type {
SparseVectorWeightType::F32 => canonical_vector_zero(weight),
SparseVectorWeightType::F16 => {
let bits = f32_to_f16_rne(weight).ok_or(SparseVectorValueError::F16Overflow(ordinal))?;
canonical_vector_zero(f16_to_f32(bits))
}
};
if weight == 0.0 {
return Err(SparseVectorValueError::ZeroWeight(ordinal));
}
Ok(DecodedSparseVectorEntry {
dimension: entry.dimension,
weight,
})
})
.collect()
}
pub fn validate_bounded(
&self,
definition: SparseVectorDefinition,
maximum_bytes: usize,
) -> Result<usize, SparseVectorValueError> {
definition.validate()?;
if self.entries.is_empty() {
return Err(SparseVectorValueError::Empty);
}
if self.entries.len() > MAX_SPARSE_VECTOR_NONZERO {
return Err(SparseVectorValueError::TooManyEntries(self.entries.len()));
}
let bytes = self
.entries
.len()
.checked_mul(4_usize.saturating_add(definition.weight_type.width_bytes()))
.ok_or(SparseVectorValueError::ByteLengthOverflow)?;
if bytes > maximum_bytes.min(MAX_SPARSE_VECTOR_BYTES) {
return Err(SparseVectorValueError::FieldBytesExceeded(bytes));
}
let mut previous = None;
for (ordinal, entry) in self.entries.iter().enumerate() {
if entry.dimension > definition.maximum_dimension {
return Err(SparseVectorValueError::DimensionOutOfRange {
ordinal,
dimension: entry.dimension,
maximum: definition.maximum_dimension,
});
}
if previous.is_some_and(|dimension| entry.dimension <= dimension) {
return Err(SparseVectorValueError::DimensionsNotStrict(ordinal));
}
if !entry.weight.is_finite() {
return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
}
if entry.weight == 0.0 {
return Err(SparseVectorValueError::ZeroWeight(ordinal));
}
previous = Some(entry.dimension);
}
Ok(bytes)
}
pub fn canonical_native_bytes(
&self,
definition: SparseVectorDefinition,
) -> Result<Vec<u8>, SparseVectorValueError> {
let decoded = self.decode(definition)?;
let mut bytes = Vec::with_capacity(
decoded
.len()
.saturating_mul(4_usize.saturating_add(definition.weight_type.width_bytes())),
);
for entry in decoded {
bytes.extend_from_slice(&entry.dimension.to_le_bytes());
match definition.weight_type {
SparseVectorWeightType::F32 => bytes.extend_from_slice(&entry.weight.to_bits().to_le_bytes()),
SparseVectorWeightType::F16 => {
let bits = f32_to_f16_rne(entry.weight).ok_or(SparseVectorValueError::F16Overflow(0))?;
bytes.extend_from_slice(&bits.to_le_bytes());
}
}
}
Ok(bytes)
}
}
pub fn decode_sparse_native_bytes(
definition: SparseVectorDefinition,
bytes: &[u8],
) -> Result<Vec<DecodedSparseVectorEntry>, SparseVectorValueError> {
definition.validate()?;
let entry_bytes = 4_usize.saturating_add(definition.weight_type.width_bytes());
if bytes.is_empty() || !bytes.len().is_multiple_of(entry_bytes) {
return Err(SparseVectorValueError::FieldBytesExceeded(bytes.len()));
}
let count = bytes.len() / entry_bytes;
if count > MAX_SPARSE_VECTOR_NONZERO || bytes.len() > MAX_SPARSE_VECTOR_BYTES {
return Err(SparseVectorValueError::FieldBytesExceeded(bytes.len()));
}
let mut entries = Vec::with_capacity(count);
let mut previous = None;
for (ordinal, chunk) in bytes.chunks_exact(entry_bytes).enumerate() {
let dimension = u32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
if dimension > definition.maximum_dimension {
return Err(SparseVectorValueError::DimensionOutOfRange {
ordinal,
dimension,
maximum: definition.maximum_dimension,
});
}
if previous.is_some_and(|value| dimension <= value) {
return Err(SparseVectorValueError::DimensionsNotStrict(ordinal));
}
let weight = match definition.weight_type {
SparseVectorWeightType::F32 => f32::from_le_bytes([chunk[4], chunk[5], chunk[6], chunk[7]]),
SparseVectorWeightType::F16 => f16_to_f32(u16::from_le_bytes([chunk[4], chunk[5]])),
};
if !weight.is_finite() {
return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
}
let weight = canonical_vector_zero(weight);
if weight == 0.0 {
return Err(SparseVectorValueError::ZeroWeight(ordinal));
}
entries.push(DecodedSparseVectorEntry { dimension, weight });
previous = Some(dimension);
}
Ok(entries)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SparseVectorValueError {
Empty,
InvalidMaximumDimension(u32),
TooManyEntries(usize),
ByteLengthOverflow,
FieldBytesExceeded(usize),
DimensionsNotStrict(usize),
DimensionOutOfRange {
ordinal: usize,
dimension: u32,
maximum: u32,
},
NonFiniteWeight(usize),
ZeroWeight(usize),
F16Overflow(usize),
}
impl Display for SparseVectorValueError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Empty => formatter.write_str("sparse vector value is empty"),
Self::InvalidMaximumDimension(value) => {
write!(formatter, "sparse maximum dimension {value} is invalid")
}
Self::TooManyEntries(value) => write!(formatter, "sparse vector has {value} non-zero entries"),
Self::ByteLengthOverflow => formatter.write_str("sparse vector byte length overflowed"),
Self::FieldBytesExceeded(value) => {
write!(formatter, "sparse vector uses {value} bytes above its field ceiling")
}
Self::DimensionsNotStrict(ordinal) => {
write!(formatter, "sparse vector dimension {ordinal} is duplicate or unsorted")
}
Self::DimensionOutOfRange {
ordinal,
dimension,
maximum,
} => write!(
formatter,
"sparse vector dimension {ordinal} value {dimension} exceeds {maximum}"
),
Self::NonFiniteWeight(ordinal) => write!(formatter, "sparse vector weight {ordinal} is non-finite"),
Self::ZeroWeight(ordinal) => write!(formatter, "sparse vector weight {ordinal} is zero"),
Self::F16Overflow(ordinal) => write!(formatter, "sparse f16 weight {ordinal} overflows binary16"),
}
}
}
impl Error for SparseVectorValueError {}
fn deserialize_bounded_sparse_entries<'de, D>(deserializer: D) -> Result<Vec<SparseVectorEntry>, D::Error>
where
D: Deserializer<'de>,
{
struct SparseEntriesVisitor;
impl<'de> serde::de::Visitor<'de> for SparseEntriesVisitor {
type Value = Vec<SparseVectorEntry>;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("a bounded sparse entry sequence")
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
if sequence
.size_hint()
.is_some_and(|length| length > MAX_SPARSE_VECTOR_NONZERO)
{
return Err(de::Error::custom("sparse vector exceeds the non-zero ceiling"));
}
let mut entries =
Vec::with_capacity(sequence.size_hint().unwrap_or_default().min(MAX_SPARSE_VECTOR_NONZERO));
while let Some(entry) = sequence.next_element::<SparseVectorEntry>()? {
if entries.len() == MAX_SPARSE_VECTOR_NONZERO {
return Err(de::Error::custom("sparse vector exceeds the non-zero ceiling"));
}
entries.push(entry);
}
Ok(entries)
}
}
deserializer.deserialize_seq(SparseEntriesVisitor)
}
impl DenseVectorElementType {
#[must_use]
pub const fn width_bytes(self) -> usize {
match self {
Self::F32 => 4,
Self::F16 => 2,
Self::I8 => 1,
}
}
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct VectorQuantizationScale(u32);
impl VectorQuantizationScale {
pub fn new(value: f32) -> Result<Self, DenseVectorValueError> {
if !value.is_finite() || value <= 0.0 {
return Err(DenseVectorValueError::InvalidQuantizationScale);
}
Ok(Self(value.to_bits()))
}
#[must_use]
pub const fn get(self) -> f32 {
f32::from_bits(self.0)
}
}
impl Serialize for VectorQuantizationScale {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.get().to_string())
}
}
impl<'de> Deserialize<'de> for VectorQuantizationScale {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let text = String::deserialize(deserializer)?;
let value = text.parse::<f32>().map_err(de::Error::custom)?;
if value.to_string() != text {
return Err(de::Error::custom("vector quantization scale is not canonical"));
}
Self::new(value).map_err(de::Error::custom)
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DenseVectorByteOrder {
LittleEndian,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DenseVectorBinaryValue {
pub element_type: DenseVectorElementType,
pub byte_order: DenseVectorByteOrder,
pub data_base64: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub quantization_scale: Option<VectorQuantizationScale>,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum DenseVectorValue {
Numeric {
#[serde(deserialize_with = "deserialize_bounded_dense_numbers")]
values: Vec<f64>,
},
Binary {
#[serde(flatten)]
value: DenseVectorBinaryValue,
},
}
impl DenseVectorValue {
pub fn decode(&self, definition: DenseVectorDefinition) -> Result<Vec<f32>, DenseVectorValueError> {
definition.validate()?;
match self {
Self::Numeric { values } => decode_dense_numeric_value(definition, values),
Self::Binary { value } => decode_dense_binary_value(definition, value),
}
}
pub fn canonical_native_bytes(&self, definition: DenseVectorDefinition) -> Result<Vec<u8>, DenseVectorValueError> {
definition.validate()?;
match self {
Self::Numeric { values } => encode_dense_numeric_native_bytes(definition, values),
Self::Binary { value } => decode_dense_binary_native_bytes(definition, value),
}
}
#[must_use]
pub fn element_count(&self) -> Option<usize> {
match self {
Self::Numeric { values } => Some(values.len()),
Self::Binary { .. } => None,
}
}
pub fn validate_bounded(&self, maximum_bytes: usize) -> Result<usize, DenseVectorValueError> {
match self {
Self::Numeric { values } => {
if values.is_empty() {
return Err(DenseVectorValueError::Empty);
}
let bytes = values
.len()
.checked_mul(DenseVectorElementType::F32.width_bytes())
.ok_or(DenseVectorValueError::ByteLengthOverflow)?;
if values.len() > usize::from(MAX_VECTOR_DIMENSIONS) || bytes > maximum_bytes {
return Err(DenseVectorValueError::FieldBytesExceeded(bytes));
}
if let Some(ordinal) = values.iter().position(|value| !value.is_finite()) {
return Err(DenseVectorValueError::NonFiniteElement(ordinal));
}
Ok(bytes)
}
Self::Binary { value } => {
match (value.element_type, value.quantization_scale) {
(DenseVectorElementType::I8, Some(_))
| (DenseVectorElementType::F32 | DenseVectorElementType::F16, None) => {}
(DenseVectorElementType::I8, None) => {
return Err(DenseVectorValueError::MissingQuantizationScale);
}
(DenseVectorElementType::F32 | DenseVectorElementType::F16, Some(_)) => {
return Err(DenseVectorValueError::UnexpectedQuantizationScale);
}
}
if value.data_base64.is_empty() || value.data_base64.len() > maximum_bytes.div_ceil(3) * 4 {
return Err(DenseVectorValueError::FieldBytesExceeded(value.data_base64.len()));
}
let bytes = BASE64_STANDARD
.decode(value.data_base64.as_bytes())
.map_err(|_| DenseVectorValueError::InvalidBase64)?;
if bytes.is_empty()
|| bytes.len() > maximum_bytes
|| !bytes.len().is_multiple_of(value.element_type.width_bytes())
|| bytes.len() / value.element_type.width_bytes() > usize::from(MAX_VECTOR_DIMENSIONS)
{
return Err(DenseVectorValueError::FieldBytesExceeded(bytes.len()));
}
if BASE64_STANDARD.encode(&bytes) != value.data_base64 {
return Err(DenseVectorValueError::InvalidBase64);
}
Ok(bytes.len())
}
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DenseVectorDefinition {
pub dimensions: u16,
pub element_type: DenseVectorElementType,
pub metric: VectorMetric,
pub quantization_scale: Option<VectorQuantizationScale>,
}
impl DenseVectorDefinition {
pub fn validate(self) -> Result<(), DenseVectorValueError> {
if self.dimensions == 0 || self.dimensions > MAX_VECTOR_DIMENSIONS {
return Err(DenseVectorValueError::InvalidDimensions(self.dimensions));
}
let bytes = usize::from(self.dimensions)
.checked_mul(self.element_type.width_bytes())
.ok_or(DenseVectorValueError::ByteLengthOverflow)?;
if bytes > MAX_VECTOR_BYTES_PER_FIELD {
return Err(DenseVectorValueError::FieldBytesExceeded(bytes));
}
match (self.element_type, self.quantization_scale) {
(DenseVectorElementType::I8, Some(_))
| (DenseVectorElementType::F32 | DenseVectorElementType::F16, None) => Ok(()),
(DenseVectorElementType::I8, None) => Err(DenseVectorValueError::MissingQuantizationScale),
(DenseVectorElementType::F32 | DenseVectorElementType::F16, Some(_)) => {
Err(DenseVectorValueError::UnexpectedQuantizationScale)
}
}
}
#[must_use]
pub const fn encoded_bytes(self) -> usize {
self.dimensions as usize * self.element_type.width_bytes()
}
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct VectorSetValue {
#[serde(deserialize_with = "deserialize_bounded_vector_set")]
pub vectors: Vec<DenseVectorValue>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VectorSetDefinition {
pub dense: DenseVectorDefinition,
pub maximum_vectors: u16,
}
#[derive(Clone, Debug, PartialEq)]
pub struct DecodedVectorSet {
pub vectors: Vec<Vec<f32>>,
pub native_vectors: Vec<Vec<u8>>,
pub native_bytes: usize,
}
impl VectorSetDefinition {
pub fn validate(self) -> Result<(), VectorSetValueError> {
self.dense.validate().map_err(VectorSetValueError::Dense)?;
if self.maximum_vectors == 0 || usize::from(self.maximum_vectors) > MAX_VECTOR_SET_DOCUMENT_VECTORS {
return Err(VectorSetValueError::InvalidMaximumVectors(self.maximum_vectors));
}
let maximum_bytes = usize::from(self.maximum_vectors)
.checked_mul(self.dense.encoded_bytes())
.ok_or(VectorSetValueError::ByteLengthOverflow)?;
if maximum_bytes > MAX_VECTOR_SET_BYTES_PER_FIELD {
return Err(VectorSetValueError::FieldBytesExceeded(maximum_bytes));
}
Ok(())
}
}
impl VectorSetValue {
pub fn decode(
&self,
definition: VectorSetDefinition,
maximum_vectors: usize,
maximum_bytes: usize,
) -> Result<DecodedVectorSet, VectorSetValueError> {
definition.validate()?;
let admitted_vectors = maximum_vectors
.min(usize::from(definition.maximum_vectors))
.min(MAX_VECTOR_SET_DOCUMENT_VECTORS);
if self.vectors.is_empty() {
return Err(VectorSetValueError::Empty);
}
if self.vectors.len() > admitted_vectors {
return Err(VectorSetValueError::TooManyVectors(self.vectors.len()));
}
let native_bytes = self
.vectors
.len()
.checked_mul(definition.dense.encoded_bytes())
.ok_or(VectorSetValueError::ByteLengthOverflow)?;
if native_bytes > maximum_bytes.min(MAX_VECTOR_SET_BYTES_PER_FIELD) {
return Err(VectorSetValueError::FieldBytesExceeded(native_bytes));
}
let mut vectors = Vec::with_capacity(self.vectors.len());
let mut native_vectors = Vec::with_capacity(self.vectors.len());
for value in &self.vectors {
vectors.push(value.decode(definition.dense).map_err(VectorSetValueError::Dense)?);
native_vectors.push(
value
.canonical_native_bytes(definition.dense)
.map_err(VectorSetValueError::Dense)?,
);
}
Ok(DecodedVectorSet {
vectors,
native_vectors,
native_bytes,
})
}
}
fn deserialize_bounded_vector_set<'de, D>(deserializer: D) -> Result<Vec<DenseVectorValue>, D::Error>
where
D: Deserializer<'de>,
{
struct VectorSetVisitor;
impl<'de> serde::de::Visitor<'de> for VectorSetVisitor {
type Value = Vec<DenseVectorValue>;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("a bounded sequence of dense vector values")
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
if sequence
.size_hint()
.is_some_and(|length| length > MAX_VECTOR_SET_DOCUMENT_VECTORS)
{
return Err(de::Error::custom(
"vector set exceeds the document vector-count ceiling",
));
}
let mut vectors = Vec::with_capacity(
sequence
.size_hint()
.unwrap_or_default()
.min(MAX_VECTOR_SET_DOCUMENT_VECTORS),
);
while let Some(vector) = sequence.next_element::<DenseVectorValue>()? {
if vectors.len() == MAX_VECTOR_SET_DOCUMENT_VECTORS {
return Err(de::Error::custom(
"vector set exceeds the document vector-count ceiling",
));
}
vectors.push(vector);
}
Ok(vectors)
}
}
deserializer.deserialize_seq(VectorSetVisitor)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum VectorSetValueError {
Empty,
InvalidMaximumVectors(u16),
TooManyVectors(usize),
FieldBytesExceeded(usize),
ByteLengthOverflow,
Dense(DenseVectorValueError),
}
impl Display for VectorSetValueError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Empty => formatter.write_str("vector set cannot be empty"),
Self::InvalidMaximumVectors(value) => {
write!(formatter, "vector-set maximum vector count {value} is invalid")
}
Self::TooManyVectors(value) => write!(formatter, "vector set contains {value} vectors"),
Self::FieldBytesExceeded(value) => write!(formatter, "vector set uses {value} element bytes"),
Self::ByteLengthOverflow => formatter.write_str("vector-set byte length overflowed"),
Self::Dense(error) => write!(formatter, "vector-set member is invalid: {error}"),
}
}
}
impl Error for VectorSetValueError {}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DenseVectorValueError {
Empty,
InvalidDimensions(u16),
DimensionMismatch { expected: u16, observed: usize },
InvalidQuantizationScale,
MissingQuantizationScale,
UnexpectedQuantizationScale,
QuantizationScaleMismatch,
ElementTypeMismatch,
ByteLengthOverflow,
FieldBytesExceeded(usize),
NonFiniteElement(usize),
F16Overflow(usize),
I8ElementOutOfRange(usize),
InvalidBase64Length { expected: usize, observed: usize },
InvalidBase64,
DecodedByteLength { expected: usize, observed: usize },
ZeroNorm,
}
impl Display for DenseVectorValueError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Empty => formatter.write_str("dense vector value is empty"),
Self::InvalidDimensions(value) => write!(formatter, "dense vector dimensions {value} are invalid"),
Self::DimensionMismatch { expected, observed } => {
write!(formatter, "dense vector has {observed} elements; expected {expected}")
}
Self::InvalidQuantizationScale => {
formatter.write_str("dense i8 quantization scale must be positive and finite")
}
Self::MissingQuantizationScale => {
formatter.write_str("dense i8 value requires quantization scale metadata")
}
Self::UnexpectedQuantizationScale => {
formatter.write_str("dense f32/f16 value cannot carry quantization metadata")
}
Self::QuantizationScaleMismatch => {
formatter.write_str("dense i8 value quantization scale differs from its mapping")
}
Self::ElementTypeMismatch => {
formatter.write_str("dense binary value element type differs from its mapping")
}
Self::ByteLengthOverflow => formatter.write_str("dense vector byte length overflowed"),
Self::FieldBytesExceeded(value) => {
write!(formatter, "dense vector uses {value} bytes above its field ceiling")
}
Self::NonFiniteElement(ordinal) => write!(formatter, "dense vector element {ordinal} is non-finite"),
Self::F16Overflow(ordinal) => write!(formatter, "dense f16 element {ordinal} overflows binary16"),
Self::I8ElementOutOfRange(ordinal) => {
write!(
formatter,
"dense i8 element {ordinal} is not an exact integer in -128..=127"
)
}
Self::InvalidBase64Length { expected, observed } => {
write!(
formatter,
"dense binary base64 has {observed} bytes; expected {expected}"
)
}
Self::InvalidBase64 => formatter.write_str("dense binary value is not canonical base64"),
Self::DecodedByteLength { expected, observed } => {
write!(
formatter,
"dense binary value has {observed} decoded bytes; expected {expected}"
)
}
Self::ZeroNorm => formatter.write_str("dense cosine vector has zero norm"),
}
}
}
impl Error for DenseVectorValueError {}
fn deserialize_bounded_dense_numbers<'de, D>(deserializer: D) -> Result<Vec<f64>, D::Error>
where
D: Deserializer<'de>,
{
struct DenseNumbersVisitor;
impl<'de> Visitor<'de> for DenseNumbersVisitor {
type Value = Vec<f64>;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
write!(formatter, "at most {MAX_VECTOR_DIMENSIONS} dense vector numbers")
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
if sequence
.size_hint()
.is_some_and(|length| length > usize::from(MAX_VECTOR_DIMENSIONS))
{
return Err(de::Error::custom("dense vector exceeds the element ceiling"));
}
let mut values = Vec::with_capacity(
sequence
.size_hint()
.unwrap_or_default()
.min(usize::from(MAX_VECTOR_DIMENSIONS)),
);
while let Some(value) = sequence.next_element::<f64>()? {
if values.len() == usize::from(MAX_VECTOR_DIMENSIONS) {
return Err(de::Error::custom("dense vector exceeds the element ceiling"));
}
values.push(value);
}
Ok(values)
}
}
deserializer.deserialize_seq(DenseNumbersVisitor)
}
pub fn decode_dense_numeric_value(
definition: DenseVectorDefinition,
values: &[f64],
) -> Result<Vec<f32>, DenseVectorValueError> {
let bytes = encode_dense_numeric_native_bytes(definition, values)?;
decode_dense_native_bytes(definition, &bytes)
}
pub fn decode_dense_binary_value(
definition: DenseVectorDefinition,
value: &DenseVectorBinaryValue,
) -> Result<Vec<f32>, DenseVectorValueError> {
let bytes = decode_dense_binary_native_bytes(definition, value)?;
decode_dense_native_bytes(definition, &bytes)
}
fn decode_dense_binary_native_bytes(
definition: DenseVectorDefinition,
value: &DenseVectorBinaryValue,
) -> Result<Vec<u8>, DenseVectorValueError> {
definition.validate()?;
if value.element_type != definition.element_type {
return Err(DenseVectorValueError::ElementTypeMismatch);
}
if value.quantization_scale != definition.quantization_scale {
return Err(DenseVectorValueError::QuantizationScaleMismatch);
}
let expected_bytes = definition.encoded_bytes();
let expected_base64 = expected_bytes.div_ceil(3) * 4;
if value.data_base64.len() != expected_base64 {
return Err(DenseVectorValueError::InvalidBase64Length {
expected: expected_base64,
observed: value.data_base64.len(),
});
}
let bytes = BASE64_STANDARD
.decode(value.data_base64.as_bytes())
.map_err(|_| DenseVectorValueError::InvalidBase64)?;
if bytes.len() != expected_bytes {
return Err(DenseVectorValueError::DecodedByteLength {
expected: expected_bytes,
observed: bytes.len(),
});
}
if BASE64_STANDARD.encode(&bytes) != value.data_base64 {
return Err(DenseVectorValueError::InvalidBase64);
}
Ok(bytes)
}
fn encode_dense_numeric_native_bytes(
definition: DenseVectorDefinition,
values: &[f64],
) -> Result<Vec<u8>, DenseVectorValueError> {
definition.validate()?;
if values.len() != usize::from(definition.dimensions) {
return Err(DenseVectorValueError::DimensionMismatch {
expected: definition.dimensions,
observed: values.len(),
});
}
let mut bytes = Vec::with_capacity(definition.encoded_bytes());
for (ordinal, value) in values.iter().copied().enumerate() {
if !value.is_finite() {
return Err(DenseVectorValueError::NonFiniteElement(ordinal));
}
match definition.element_type {
DenseVectorElementType::F32 => {
let value = canonical_vector_zero(value as f32);
if !value.is_finite() {
return Err(DenseVectorValueError::NonFiniteElement(ordinal));
}
bytes.extend_from_slice(&value.to_bits().to_le_bytes());
}
DenseVectorElementType::F16 => {
let value = value as f32;
if !value.is_finite() {
return Err(DenseVectorValueError::F16Overflow(ordinal));
}
let bits = f32_to_f16_rne(value).ok_or(DenseVectorValueError::F16Overflow(ordinal))?;
bytes.extend_from_slice(&bits.to_le_bytes());
}
DenseVectorElementType::I8 => {
if value.fract() != 0.0 || !(-128.0..=127.0).contains(&value) {
return Err(DenseVectorValueError::I8ElementOutOfRange(ordinal));
}
bytes.push(value as i8 as u8);
}
}
}
Ok(bytes)
}
pub fn decode_dense_native_bytes(
definition: DenseVectorDefinition,
bytes: &[u8],
) -> Result<Vec<f32>, DenseVectorValueError> {
definition.validate()?;
let expected = definition.encoded_bytes();
if bytes.len() != expected {
return Err(DenseVectorValueError::DecodedByteLength {
expected,
observed: bytes.len(),
});
}
let mut decoded = Vec::with_capacity(usize::from(definition.dimensions));
match definition.element_type {
DenseVectorElementType::F32 => {
for (ordinal, chunk) in bytes.as_chunks::<4>().0.iter().enumerate() {
let value = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
if !value.is_finite() {
return Err(DenseVectorValueError::NonFiniteElement(ordinal));
}
decoded.push(canonical_vector_zero(value));
}
}
DenseVectorElementType::F16 => {
for (ordinal, chunk) in bytes.as_chunks::<2>().0.iter().enumerate() {
let value = f16_to_f32(u16::from_le_bytes([chunk[0], chunk[1]]));
if !value.is_finite() {
return Err(DenseVectorValueError::NonFiniteElement(ordinal));
}
decoded.push(canonical_vector_zero(value));
}
}
DenseVectorElementType::I8 => {
let scale = definition
.quantization_scale
.ok_or(DenseVectorValueError::MissingQuantizationScale)?
.get();
decoded.extend(
bytes
.iter()
.map(|value| canonical_vector_zero((*value as i8 as f32) * scale)),
);
}
}
normalize_dense_for_metric(definition.metric, &mut decoded)?;
Ok(decoded)
}
fn normalize_dense_for_metric(metric: VectorMetric, values: &mut [f32]) -> Result<(), DenseVectorValueError> {
if metric != VectorMetric::Cosine {
return Ok(());
}
let squared_norm = values.iter().try_fold(0.0_f64, |sum, value| {
let next = sum + f64::from(*value) * f64::from(*value);
next.is_finite().then_some(next)
});
let norm = squared_norm
.map(f64::sqrt)
.filter(|value| *value > 0.0 && value.is_finite())
.ok_or(DenseVectorValueError::ZeroNorm)?;
for value in values {
*value = canonical_vector_zero((f64::from(*value) / norm) as f32);
if !value.is_finite() {
return Err(DenseVectorValueError::NonFiniteElement(0));
}
}
Ok(())
}
const fn canonical_vector_zero(value: f32) -> f32 {
if value == 0.0 { 0.0 } else { value }
}
fn f32_to_f16_rne(value: f32) -> Option<u16> {
if !value.is_finite() {
return None;
}
let bits = canonical_vector_zero(value).to_bits();
let sign = ((bits >> 16) & 0x8000) as u16;
let exponent = ((bits >> 23) & 0xff) as i32 - 127;
let fraction = bits & 0x7f_ffff;
if exponent > 15 {
return None;
}
if exponent >= -14 {
let mut half_exponent = (exponent + 15) as u16;
let mut half_fraction = (fraction >> 13) as u16;
let remainder = fraction & 0x1fff;
if remainder > 0x1000 || remainder == 0x1000 && half_fraction & 1 == 1 {
half_fraction += 1;
if half_fraction == 0x400 {
half_fraction = 0;
half_exponent += 1;
if half_exponent >= 0x1f {
return None;
}
}
}
return Some(sign | (half_exponent << 10) | half_fraction);
}
let significand = u64::from((1_u32 << 23) | fraction);
let shift = u32::try_from(-exponent - 1).ok()?;
let rounded = round_shift_right_ties_even(significand, shift);
if rounded >= 0x400 {
Some(sign | 0x0400)
} else {
Some(sign | rounded as u16)
}
}
fn round_shift_right_ties_even(value: u64, shift: u32) -> u64 {
if shift == 0 {
return value;
}
if shift >= 64 {
return 0;
}
let quotient = value >> shift;
let remainder_mask = (1_u64 << shift) - 1;
let remainder = value & remainder_mask;
let halfway = 1_u64 << (shift - 1);
quotient + u64::from(remainder > halfway || remainder == halfway && quotient & 1 == 1)
}
fn f16_to_f32(bits: u16) -> f32 {
let sign = u32::from(bits & 0x8000) << 16;
let exponent = (bits >> 10) & 0x1f;
let fraction = bits & 0x03ff;
match (exponent, fraction) {
(0, 0) => f32::from_bits(sign),
(0, _) => {
let value = f32::from(fraction) * 2.0_f32.powi(-24);
if sign == 0 { value } else { -value }
}
(0x1f, _) => f32::from_bits(sign | 0x7f80_0000 | (u32::from(fraction) << 13)),
_ => {
let f32_exponent = u32::from(exponent) + 112;
f32::from_bits(sign | (f32_exponent << 23) | (u32::from(fraction) << 13))
}
}
}
pub const MAX_MAPPING_FIELD_NAME_BYTES: usize = 1_024;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Mapping {
pub format_version: u16,
pub schema_generation: SchemaGeneration,
pub mode: IndexMode,
pub fields: BTreeMap<String, FieldMapping>,
pub extensions: MappingExtensions,
}
impl Mapping {
#[must_use]
pub fn empty(mode: IndexMode) -> Self {
Self {
format_version: MAPPING_FORMAT_VERSION,
schema_generation: SchemaGeneration::new(1),
mode,
fields: BTreeMap::new(),
extensions: crate::MappingExtensions::default(),
}
}
#[must_use]
pub fn normalized(mut self) -> Self {
if self.format_version == MAPPING_FORMAT_VERSION {
if self.extensions.analyzer_identities.is_empty() {
self.extensions.analyzer_identities = self
.fields
.iter()
.filter(|(_, definition)| definition.field_type == FieldType::Text)
.map(|(field, _)| AnalyzerIdentity {
field: field.clone(),
index_analyzer: AnalyzerName::Standard,
search_analyzer: AnalyzerName::Standard,
})
.collect();
}
if self.extensions.keyword_limits.is_empty() {
self.extensions.keyword_limits = self
.fields
.iter()
.filter(|(_, definition)| definition.field_type == FieldType::Keyword)
.map(|(field, _)| KeywordFieldPolicy {
field: field.clone(),
max_bytes: DEFAULT_KEYWORD_IGNORE_ABOVE,
normalizer: KeywordNormalizer::Keyword,
})
.collect();
}
if self.extensions.nested_paths.is_empty() {
self.extensions.nested_paths = self
.fields
.iter()
.filter(|(_, definition)| definition.field_type == FieldType::Nested)
.map(|(field, _)| field.clone())
.collect();
}
}
self
}
pub fn decode_canonical(bytes: &[u8]) -> Result<Self, MappingCodecError> {
if bytes.len() > MAX_MAPPING_BYTES {
return Err(MappingCodecError::TooLarge(bytes.len()));
}
let mut probe = serde_json::Deserializer::from_slice(bytes);
let version = MappingVersionProbe::deserialize(&mut probe).map_err(|error| {
let message = error.to_string();
if message.starts_with("duplicate mapping key '") {
MappingCodecError::DuplicateKey(message)
} else {
MappingCodecError::Malformed(message)
}
})?;
probe
.end()
.map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
if version != MAPPING_FORMAT_VERSION {
return Err(MappingCodecError::UnsupportedVersion(version));
}
let mapping: Self =
serde_json::from_slice(bytes).map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
mapping.validate().map_err(MappingCodecError::Invalid)?;
let canonical =
serde_json::to_vec(&mapping).map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
if canonical != bytes {
return Err(MappingCodecError::NonCanonical);
}
Ok(mapping)
}
pub fn encode_canonical(&self) -> Result<Vec<u8>, MappingCodecError> {
self.validate().map_err(MappingCodecError::Invalid)?;
serde_json::to_vec(self).map_err(|error| MappingCodecError::Malformed(error.to_string()))
}
pub fn validate(&self) -> Result<(), MappingValidationError> {
if self.format_version != MAPPING_FORMAT_VERSION {
return Err(MappingValidationError::UnsupportedVersion(self.format_version));
}
let normalized = self.clone().normalized();
let mapping = &normalized;
if mapping.schema_generation.get() == 0 {
return Err(MappingValidationError::ZeroSchemaGeneration);
}
if mapping.fields.len() > MAX_MAPPING_FIELDS {
return Err(MappingValidationError::TooManyFields(mapping.fields.len()));
}
let mut vector_field_ids = std::collections::BTreeSet::new();
let mut vector_fields = 0_usize;
for (name, definition) in &mapping.fields {
validate_mapping_field_name(name)?;
for (separator, _) in name.match_indices('.') {
let ancestor = &name[..separator];
if mapping.fields.contains_key(ancestor)
&& !mapping.extensions.nested_paths.iter().any(|path| path == ancestor)
&& mapping
.extensions
.multi_field_projections
.iter()
.find(|projection| projection.child == *name)
.is_none_or(|projection| projection.parent != ancestor)
{
return Err(MappingValidationError::ConflictingFieldPaths {
ancestor: ancestor.to_owned(),
descendant: name.clone(),
});
}
}
match definition.field_type {
FieldType::Vector {
format_version,
field_id,
dimensions,
metric,
element_type,
quantization_scale,
} => {
vector_fields = vector_fields.saturating_add(1);
if format_version != VECTOR_MAPPING_FORMAT_VERSION {
return Err(MappingValidationError::UnsupportedVectorVersion(format_version));
}
if dimensions == 0 || dimensions > MAX_VECTOR_DIMENSIONS {
return Err(MappingValidationError::InvalidVectorDimensions {
field: name.clone(),
dimensions,
});
}
DenseVectorDefinition {
dimensions,
element_type,
metric,
quantization_scale,
}
.validate()
.map_err(|_| MappingValidationError::InvalidVectorElementContract(name.clone()))?;
if definition.stored {
return Err(MappingValidationError::StoredVectorField(name.clone()));
}
if !vector_field_ids.insert(field_id) {
return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
}
}
FieldType::SparseVector {
format_version,
field_id,
maximum_dimension,
weight_type,
} => {
vector_fields = vector_fields.saturating_add(1);
if format_version != SPARSE_VECTOR_MAPPING_FORMAT_VERSION {
return Err(MappingValidationError::UnsupportedSparseVectorVersion(format_version));
}
SparseVectorDefinition {
maximum_dimension,
weight_type,
}
.validate()
.map_err(|_| MappingValidationError::InvalidSparseVectorContract(name.clone()))?;
if definition.stored {
return Err(MappingValidationError::StoredVectorField(name.clone()));
}
if !vector_field_ids.insert(field_id) {
return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
}
}
FieldType::VectorSet {
format_version,
field_id,
dimensions,
metric,
element_type,
quantization_scale,
maximum_vectors,
} => {
vector_fields = vector_fields.saturating_add(1);
if format_version != VECTOR_SET_MAPPING_FORMAT_VERSION {
return Err(MappingValidationError::UnsupportedVectorSetVersion(format_version));
}
VectorSetDefinition {
dense: DenseVectorDefinition {
dimensions,
element_type,
metric,
quantization_scale,
},
maximum_vectors,
}
.validate()
.map_err(|_| MappingValidationError::InvalidVectorSetContract(name.clone()))?;
if definition.stored {
return Err(MappingValidationError::StoredVectorField(name.clone()));
}
if !vector_field_ids.insert(field_id) {
return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
}
}
_ => {}
}
definition.validate(name)?;
}
if vector_fields > MAX_VECTOR_MAPPING_FIELDS {
return Err(MappingValidationError::TooManyVectorFields(vector_fields));
}
mapping.extensions.validate(mapping)?;
Ok(())
}
pub fn validate_current(&self) -> Result<(), MappingValidationError> {
self.validate()
}
#[must_use]
pub fn is_additive_successor_of(&self, predecessor: &Self) -> bool {
let predecessor = predecessor.clone().normalized();
let successor = self.clone().normalized();
predecessor.format_version == MAPPING_FORMAT_VERSION
&& successor.format_version == MAPPING_FORMAT_VERSION
&& predecessor.mode == successor.mode
&& predecessor.extensions.dynamic_policy == successor.extensions.dynamic_policy
&& predecessor
.fields
.iter()
.all(|(path, field)| successor.fields.get(path) == Some(field))
&& predecessor
.extensions
.multi_field_projections
.iter()
.all(|entry| successor.extensions.multi_field_projections.contains(entry))
&& predecessor
.extensions
.analyzer_identities
.iter()
.all(|entry| successor.extensions.analyzer_identities.contains(entry))
&& predecessor
.extensions
.keyword_limits
.iter()
.all(|entry| successor.extensions.keyword_limits.contains(entry))
&& predecessor
.extensions
.nested_paths
.iter()
.all(|entry| successor.extensions.nested_paths.contains(entry))
&& successor
.extensions
.nested_paths
.iter()
.filter(|path| !predecessor.extensions.nested_paths.contains(path))
.all(|path| {
predecessor.fields.keys().all(|field| {
field != path
&& !field
.strip_prefix(path.as_str())
.is_some_and(|suffix| suffix.starts_with('.'))
})
})
}
}
struct MappingVersionProbe;
impl MappingVersionProbe {
fn deserialize<'de, D>(deserializer: D) -> Result<u16, D::Error>
where
D: serde::Deserializer<'de>,
{
struct VersionVisitor;
impl<'de> Visitor<'de> for VersionVisitor {
type Value = u16;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("a mapping JSON object")
}
fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
where
A: MapAccess<'de>,
{
let mut keys = std::collections::BTreeSet::new();
let mut version = None;
while let Some(key) = map.next_key::<String>()? {
if !keys.insert(key.clone()) {
return Err(de::Error::custom(format!("duplicate mapping key '{key}'")));
}
let raw = map.next_value::<Box<serde_json::value::RawValue>>()?;
if key == "format_version" {
version = Some(
serde_json::from_str::<u16>(raw.get())
.map_err(|_| de::Error::custom("mapping format_version is invalid"))?,
);
}
}
version.ok_or_else(|| de::Error::custom("mapping format_version is missing"))
}
}
deserializer.deserialize_map(VersionVisitor)
}
}
impl Serialize for Mapping {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
use serde::ser::Error as _;
if self.format_version != MAPPING_FORMAT_VERSION {
return Err(S::Error::custom(format!(
"unsupported mapping format version {}",
self.format_version
)));
}
let normalized = self.clone().normalized();
MappingWireRef {
format_version: normalized.format_version,
generation: normalized.schema_generation,
mode: normalized.mode,
dynamic_policy: &normalized.extensions.dynamic_policy,
fields: &normalized.fields,
multi_field_projections: &normalized.extensions.multi_field_projections,
analyzer_identities: &normalized.extensions.analyzer_identities,
keyword_limits: &normalized.extensions.keyword_limits,
nested_paths: &normalized.extensions.nested_paths,
}
.serialize(serializer)
}
}
impl<'de> Deserialize<'de> for Mapping {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
use serde::de::Error as _;
let value = serde_json::Value::deserialize(deserializer)?;
let version = value
.get("format_version")
.and_then(serde_json::Value::as_u64)
.and_then(|value| u16::try_from(value).ok())
.ok_or_else(|| D::Error::custom("mapping format_version is missing or invalid"))?;
if version != MAPPING_FORMAT_VERSION {
return Err(D::Error::custom(format!(
"unsupported mapping format version {version}"
)));
}
let wire: MappingWire = serde_json::from_value(value).map_err(D::Error::custom)?;
Ok(Self {
format_version: wire.format_version,
schema_generation: wire.generation,
mode: wire.mode,
fields: wire.fields,
extensions: MappingExtensions {
dynamic_policy: wire.dynamic_policy,
multi_field_projections: wire.multi_field_projections,
analyzer_identities: wire.analyzer_identities,
keyword_limits: wire.keyword_limits,
nested_paths: wire.nested_paths,
},
})
}
}
#[derive(Serialize)]
struct MappingWireRef<'a> {
format_version: u16,
generation: SchemaGeneration,
mode: IndexMode,
dynamic_policy: &'a DynamicMappingPolicy,
fields: &'a BTreeMap<String, FieldMapping>,
multi_field_projections: &'a Vec<MultiFieldProjection>,
analyzer_identities: &'a Vec<AnalyzerIdentity>,
keyword_limits: &'a Vec<KeywordFieldPolicy>,
nested_paths: &'a Vec<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct MappingWire {
format_version: u16,
generation: SchemaGeneration,
mode: IndexMode,
dynamic_policy: DynamicMappingPolicy,
fields: BTreeMap<String, FieldMapping>,
multi_field_projections: Vec<MultiFieldProjection>,
analyzer_identities: Vec<AnalyzerIdentity>,
keyword_limits: Vec<KeywordFieldPolicy>,
nested_paths: Vec<String>,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct MappingExtensions {
pub dynamic_policy: DynamicMappingPolicy,
pub multi_field_projections: Vec<MultiFieldProjection>,
pub analyzer_identities: Vec<AnalyzerIdentity>,
pub keyword_limits: Vec<KeywordFieldPolicy>,
pub nested_paths: Vec<String>,
}
impl MappingExtensions {
fn validate(&self, mapping: &Mapping) -> Result<(), MappingValidationError> {
if self.multi_field_projections.len() > MAX_MAPPING_FIELDS
|| self.analyzer_identities.len() > MAX_MAPPING_FIELDS
|| self.keyword_limits.len() > MAX_MAPPING_FIELDS
{
return Err(MappingValidationError::TooManyMappingMetadata);
}
if self.nested_paths.len() > MAX_NESTED_MAPPING_PATHS {
return Err(MappingValidationError::TooManyNestedPaths(self.nested_paths.len()));
}
require_sorted_unique(
self.multi_field_projections
.iter()
.map(|projection| projection.child.as_str()),
"multi_field_projections",
)?;
require_sorted_unique(
self.analyzer_identities.iter().map(|identity| identity.field.as_str()),
"analyzer_identities",
)?;
require_sorted_unique(
self.keyword_limits.iter().map(|policy| policy.field.as_str()),
"keyword_limits",
)?;
require_sorted_unique(self.nested_paths.iter().map(String::as_str), "nested_paths")?;
let nested_paths = self
.nested_paths
.iter()
.map(String::as_str)
.collect::<std::collections::BTreeSet<_>>();
for child in &self.nested_paths {
for (separator, _) in child.match_indices('.') {
let parent = &child[..separator];
if nested_paths.contains(parent) {
return Err(MappingValidationError::NestedPathOverlap {
parent: parent.to_owned(),
child: child.clone(),
});
}
}
}
for projection in &self.multi_field_projections {
validate_mapping_field_name(&projection.child)?;
validate_mapping_field_name(&projection.parent)?;
let Some(child) = mapping.fields.get(&projection.child) else {
return Err(MappingValidationError::MissingMultiField(projection.child.clone()));
};
let Some(parent) = mapping.fields.get(&projection.parent) else {
return Err(MappingValidationError::MissingMultiFieldSource {
field: projection.child.clone(),
source: projection.parent.clone(),
});
};
if !projection
.child
.strip_prefix(&projection.parent)
.is_some_and(|suffix| suffix.starts_with('.') && suffix.len() > 1)
{
return Err(MappingValidationError::InvalidMultiFieldPath(projection.child.clone()));
}
if matches!(
child.field_type,
FieldType::Vector { .. }
| FieldType::SparseVector { .. }
| FieldType::VectorSet { .. }
| FieldType::Nested
) || matches!(
parent.field_type,
FieldType::Vector { .. }
| FieldType::SparseVector { .. }
| FieldType::VectorSet { .. }
| FieldType::Nested
) {
return Err(MappingValidationError::InvalidMultiFieldEndpoint(
projection.child.clone(),
));
}
if !multi_field_types_are_compatible(parent.field_type, child.field_type) {
return Err(MappingValidationError::InvalidMultiFieldEndpoint(
projection.child.clone(),
));
}
}
for identity in &self.analyzer_identities {
let Some(field) = mapping.fields.get(&identity.field) else {
return Err(MappingValidationError::UnknownFieldMetadata(identity.field.clone()));
};
if field.field_type != FieldType::Text {
return Err(MappingValidationError::AnalyzerOnNonTextField(identity.field.clone()));
}
}
for policy in &self.keyword_limits {
let Some(field) = mapping.fields.get(&policy.field) else {
return Err(MappingValidationError::UnknownFieldMetadata(policy.field.clone()));
};
if field.field_type != FieldType::Keyword {
return Err(MappingValidationError::KeywordPolicyOnNonKeywordField(
policy.field.clone(),
));
}
policy.validate()?;
if let Some(MappingScalar::Keyword(value)) = &field.null_value
&& value.len() > policy.max_bytes as usize
{
return Err(MappingValidationError::InvalidKeywordNullValue(policy.field.clone()));
}
}
for path in &self.nested_paths {
if mapping
.fields
.get(path)
.is_none_or(|field| field.field_type != FieldType::Nested)
{
return Err(MappingValidationError::InvalidNestedPath(path.clone()));
}
if path.split('.').count() > MAX_NESTED_MAPPING_DEPTH {
return Err(MappingValidationError::NestedPathTooDeep(path.clone()));
}
}
let text_fields = mapping
.fields
.iter()
.filter_map(|(path, field)| (field.field_type == FieldType::Text).then_some(path.as_str()))
.collect::<std::collections::BTreeSet<_>>();
let analyzed_fields = self
.analyzer_identities
.iter()
.map(|identity| identity.field.as_str())
.collect::<std::collections::BTreeSet<_>>();
if text_fields != analyzed_fields {
return Err(MappingValidationError::IncompleteAnalyzerCoverage);
}
let keyword_fields = mapping
.fields
.iter()
.filter_map(|(path, field)| (field.field_type == FieldType::Keyword).then_some(path.as_str()))
.collect::<std::collections::BTreeSet<_>>();
let limited_fields = self
.keyword_limits
.iter()
.map(|policy| policy.field.as_str())
.collect::<std::collections::BTreeSet<_>>();
if keyword_fields != limited_fields {
return Err(MappingValidationError::IncompleteKeywordCoverage);
}
let declared_nested = mapping
.fields
.iter()
.filter_map(|(path, field)| (field.field_type == FieldType::Nested).then_some(path.as_str()))
.collect::<std::collections::BTreeSet<_>>();
let listed_nested = self
.nested_paths
.iter()
.map(String::as_str)
.collect::<std::collections::BTreeSet<_>>();
if declared_nested != listed_nested {
return Err(MappingValidationError::IncompleteNestedCoverage);
}
Ok(())
}
}
const fn multi_field_types_are_compatible(parent: FieldType, child: FieldType) -> bool {
matches!(
(parent, child),
(
FieldType::Text | FieldType::Keyword,
FieldType::Text | FieldType::Keyword
) | (
FieldType::U64 | FieldType::Date,
FieldType::U64 | FieldType::Date | FieldType::F64
) | (FieldType::I64, FieldType::I64 | FieldType::F64)
| (FieldType::F64, FieldType::F64)
| (FieldType::Bool, FieldType::Bool)
)
}
fn require_sorted_unique<'a>(
values: impl IntoIterator<Item = &'a str>,
section: &'static str,
) -> Result<(), MappingValidationError> {
let mut previous = None;
for value in values {
if previous.is_some_and(|prior| prior >= value) {
return Err(MappingValidationError::MetadataNotSorted(section));
}
previous = Some(value);
}
Ok(())
}
pub const MAX_DYNAMIC_FIELDS_PER_REQUEST: usize = 128;
pub const MAX_NESTED_MAPPING_DEPTH: usize = 8;
pub const MAX_NESTED_MAPPING_PATHS: usize = 64;
pub const DEFAULT_KEYWORD_IGNORE_ABOVE: u32 = 32_766;
pub const MAX_SCALAR_ARRAY_DEPTH: usize = 32;
pub const MAX_SCALAR_ARRAY_NODES: usize = 4_096;
pub const MAX_SCALAR_ARRAY_VALUES: usize = 2_048;
pub const MAX_MAPPING_BYTES: usize = 1_048_576;
pub const DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION: u16 = 1;
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DynamicMappingPolicy {
True,
#[default]
False,
Strict,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum DynamicProposedTypeV1 {
Boolean,
I64,
U64,
F64,
TextWithKeyword {
index_analyzer: AnalyzerName,
search_analyzer: AnalyzerName,
keyword_child: String,
keyword_normalizer: KeywordNormalizer,
keyword_max_bytes: u32,
},
}
impl DynamicProposedTypeV1 {
pub fn from_first_candidate(
field_path: &str,
candidate: &MappingScalar,
) -> Result<Self, DynamicMappingProposalError> {
validate_mapping_field_name(field_path).map_err(DynamicMappingProposalError::InvalidField)?;
Ok(match candidate {
MappingScalar::Boolean(_) => Self::Boolean,
MappingScalar::I64(_) => Self::I64,
MappingScalar::U64(_) => Self::U64,
MappingScalar::F64Bits(bits) => {
let value = f64::from_bits(*bits);
if !value.is_finite() {
return Err(DynamicMappingProposalError::InvalidCandidate);
}
Self::F64
}
MappingScalar::Keyword(_) => Self::TextWithKeyword {
index_analyzer: AnalyzerName::Standard,
search_analyzer: AnalyzerName::Standard,
keyword_child: format!("{field_path}.keyword"),
keyword_normalizer: KeywordNormalizer::Keyword,
keyword_max_bytes: DEFAULT_KEYWORD_IGNORE_ABOVE,
},
})
}
fn validate(&self, field_path: &str) -> Result<(), DynamicMappingProposalError> {
if let Self::TextWithKeyword {
keyword_child,
keyword_max_bytes,
..
} = self
&& (keyword_child != &format!("{field_path}.keyword")
|| *keyword_max_bytes == 0
|| *keyword_max_bytes > DEFAULT_KEYWORD_IGNORE_ABOVE)
{
return Err(DynamicMappingProposalError::InvalidProposedType);
}
Ok(())
}
}
#[derive(Serialize)]
struct DynamicMappingProposalDigestInput<'a> {
format_version: u16,
collection_index_ref: CollectionIndexRef,
base_generation: SchemaGeneration,
field_path: &'a str,
first_non_null_candidate: &'a MappingScalar,
proposed_type: &'a DynamicProposedTypeV1,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DynamicMappingProposalV1 {
pub format_version: u16,
pub collection_index_ref: CollectionIndexRef,
pub base_generation: SchemaGeneration,
pub field_path: String,
pub first_non_null_candidate: MappingScalar,
pub proposed_type: DynamicProposedTypeV1,
pub proposal_digest: u64,
}
impl DynamicMappingProposalV1 {
pub fn new(
collection_index_ref: CollectionIndexRef,
base_generation: SchemaGeneration,
field_path: String,
mut first_non_null_candidate: MappingScalar,
) -> Result<Self, DynamicMappingProposalError> {
if let MappingScalar::F64Bits(bits) = &mut first_non_null_candidate
&& f64::from_bits(*bits) == 0.0
{
*bits = 0.0_f64.to_bits();
}
let proposed_type = DynamicProposedTypeV1::from_first_candidate(&field_path, &first_non_null_candidate)?;
let mut proposal = Self {
format_version: DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION,
collection_index_ref,
base_generation,
field_path,
first_non_null_candidate,
proposed_type,
proposal_digest: 0,
};
proposal.proposal_digest = proposal.exact_digest()?;
proposal.validate()?;
Ok(proposal)
}
pub fn exact_digest(&self) -> Result<u64, DynamicMappingProposalError> {
let input = DynamicMappingProposalDigestInput {
format_version: self.format_version,
collection_index_ref: self.collection_index_ref,
base_generation: self.base_generation,
field_path: &self.field_path,
first_non_null_candidate: &self.first_non_null_candidate,
proposed_type: &self.proposed_type,
};
let bytes =
serde_json::to_vec(&input).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
if bytes.len() > MAX_MAPPING_BYTES {
return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
}
Ok(bytes.iter().fold(0xcbf2_9ce4_8422_2325_u64, |digest, byte| {
(digest ^ u64::from(*byte)).wrapping_mul(0x0000_0100_0000_01b3)
}))
}
pub fn validate(&self) -> Result<(), DynamicMappingProposalError> {
if self.format_version != DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION {
return Err(DynamicMappingProposalError::UnsupportedVersion(self.format_version));
}
validate_mapping_field_name(&self.field_path).map_err(DynamicMappingProposalError::InvalidField)?;
if matches!(
self.first_non_null_candidate,
MappingScalar::F64Bits(bits) if f64::from_bits(bits) == 0.0 && bits != 0.0_f64.to_bits()
) {
return Err(DynamicMappingProposalError::InvalidCandidate);
}
self.proposed_type.validate(&self.field_path)?;
if DynamicProposedTypeV1::from_first_candidate(&self.field_path, &self.first_non_null_candidate)?
!= self.proposed_type
|| self.exact_digest()? != self.proposal_digest
{
return Err(DynamicMappingProposalError::IdentityMismatch);
}
Ok(())
}
#[must_use]
pub fn matches_successor(&self, predecessor: &Mapping, successor: &Mapping) -> bool {
let Some(next_generation) = predecessor.schema_generation.checked_next() else {
return false;
};
if self.validate().is_err()
|| self.base_generation != predecessor.schema_generation
|| successor.schema_generation != next_generation
{
return false;
}
let mut expected = predecessor.clone().normalized();
expected.schema_generation = successor.schema_generation;
let (field_type, text) = match &self.proposed_type {
DynamicProposedTypeV1::Boolean => (FieldType::Bool, None),
DynamicProposedTypeV1::I64 => (FieldType::I64, None),
DynamicProposedTypeV1::U64 => (FieldType::U64, None),
DynamicProposedTypeV1::F64 => (FieldType::F64, None),
DynamicProposedTypeV1::TextWithKeyword {
index_analyzer,
search_analyzer,
keyword_child,
keyword_normalizer,
keyword_max_bytes,
} => (
FieldType::Text,
Some((
*index_analyzer,
*search_analyzer,
keyword_child,
*keyword_normalizer,
*keyword_max_bytes,
)),
),
};
if expected
.fields
.insert(
self.field_path.clone(),
FieldMapping {
field_type,
stored: false,
null_value: None,
},
)
.is_some()
{
return false;
}
if let Some((index_analyzer, search_analyzer, keyword_child, keyword_normalizer, keyword_max_bytes)) = text {
if expected
.fields
.insert(
keyword_child.clone(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
)
.is_some()
{
return false;
}
expected.extensions.multi_field_projections.push(MultiFieldProjection {
parent: self.field_path.clone(),
child: keyword_child.clone(),
});
expected.extensions.analyzer_identities.push(AnalyzerIdentity {
field: self.field_path.clone(),
index_analyzer,
search_analyzer,
});
expected.extensions.keyword_limits.push(KeywordFieldPolicy {
field: keyword_child.clone(),
max_bytes: keyword_max_bytes,
normalizer: keyword_normalizer,
});
expected
.extensions
.multi_field_projections
.sort_by(|left, right| left.child.cmp(&right.child));
expected
.extensions
.analyzer_identities
.sort_by(|left, right| left.field.cmp(&right.field));
expected
.extensions
.keyword_limits
.sort_by(|left, right| left.field.cmp(&right.field));
}
expected == successor.clone().normalized()
}
pub fn encode_canonical(&self) -> Result<Vec<u8>, DynamicMappingProposalError> {
self.validate()?;
let bytes = serde_json::to_vec(self).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
if bytes.len() > MAX_MAPPING_BYTES {
return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
}
Ok(bytes)
}
pub fn decode_canonical(bytes: &[u8]) -> Result<Self, DynamicMappingProposalError> {
if bytes.len() > MAX_MAPPING_BYTES {
return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
}
let mut probe = serde_json::Deserializer::from_slice(bytes);
let version = MappingVersionProbe::deserialize(&mut probe)
.map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
probe
.end()
.map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
if version != DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION {
return Err(DynamicMappingProposalError::UnsupportedVersion(version));
}
let proposal: Self =
serde_json::from_slice(bytes).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
proposal.validate()?;
if proposal.encode_canonical()? != bytes {
return Err(DynamicMappingProposalError::NonCanonical);
}
Ok(proposal)
}
}
#[derive(Debug)]
pub enum DynamicMappingProposalError {
UnsupportedVersion(u16),
InvalidField(MappingValidationError),
InvalidCandidate,
InvalidProposedType,
IdentityMismatch,
TooLarge(usize),
NonCanonical,
Codec(String),
}
impl Display for DynamicMappingProposalError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedVersion(version) => {
write!(formatter, "unsupported dynamic mapping proposal version {version}")
}
Self::InvalidField(error) => write!(formatter, "invalid dynamic mapping field: {error}"),
Self::InvalidCandidate => formatter.write_str("invalid dynamic mapping first candidate"),
Self::InvalidProposedType => formatter.write_str("invalid dynamic mapping proposed type"),
Self::IdentityMismatch => formatter.write_str("dynamic mapping proposal identity mismatch"),
Self::TooLarge(bytes) => write!(formatter, "dynamic mapping proposal is too large: {bytes} bytes"),
Self::NonCanonical => formatter.write_str("dynamic mapping proposal is not canonical"),
Self::Codec(error) => write!(formatter, "dynamic mapping proposal codec error: {error}"),
}
}
}
impl Error for DynamicMappingProposalError {}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MultiFieldProjection {
pub parent: String,
pub child: String,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AnalyzerIdentity {
pub field: String,
pub index_analyzer: AnalyzerName,
pub search_analyzer: AnalyzerName,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AnalyzerName {
Keyword,
Simple,
Standard,
Whitespace,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct KeywordFieldPolicy {
pub field: String,
pub max_bytes: u32,
pub normalizer: KeywordNormalizer,
}
impl KeywordFieldPolicy {
fn validate(&self) -> Result<(), MappingValidationError> {
if self.max_bytes == 0 || self.max_bytes > DEFAULT_KEYWORD_IGNORE_ABOVE {
return Err(MappingValidationError::InvalidKeywordLimit {
field: self.field.clone(),
limit: self.max_bytes,
});
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum KeywordNormalizer {
Keyword,
Lowercase,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FieldMapping {
#[serde(rename = "type")]
pub field_type: FieldType,
#[serde(default)]
pub stored: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub null_value: Option<MappingScalar>,
}
impl FieldMapping {
fn validate(&self, field: &str) -> Result<(), MappingValidationError> {
let Some(value) = &self.null_value else {
return Ok(());
};
let valid = match (self.field_type, value) {
(FieldType::Text | FieldType::Keyword, MappingScalar::Keyword(_))
| (FieldType::U64 | FieldType::Date, MappingScalar::U64(_))
| (FieldType::I64, MappingScalar::I64(_))
| (FieldType::Bool, MappingScalar::Boolean(_)) => true,
(FieldType::U64 | FieldType::Date, MappingScalar::I64(value)) => *value >= 0,
(FieldType::F64, MappingScalar::F64Bits(bits)) => f64::from_bits(*bits).is_finite(),
(FieldType::F64, MappingScalar::I64(_) | MappingScalar::U64(_)) => true,
_ => false,
};
if valid {
Ok(())
} else {
Err(MappingValidationError::InvalidNullValue(field.to_owned()))
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum MappingScalar {
Boolean(bool),
I64(i64),
U64(u64),
F64Bits(u64),
Keyword(String),
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldType {
Text,
Keyword,
U64,
Date,
I64,
F64,
Bool,
Nested,
Vector {
format_version: u16,
field_id: VectorFieldId,
dimensions: u16,
metric: VectorMetric,
element_type: DenseVectorElementType,
#[serde(default, skip_serializing_if = "Option::is_none")]
quantization_scale: Option<VectorQuantizationScale>,
},
SparseVector {
format_version: u16,
field_id: VectorFieldId,
maximum_dimension: u32,
weight_type: SparseVectorWeightType,
},
VectorSet {
format_version: u16,
field_id: VectorFieldId,
dimensions: u16,
metric: VectorMetric,
element_type: DenseVectorElementType,
#[serde(default, skip_serializing_if = "Option::is_none")]
quantization_scale: Option<VectorQuantizationScale>,
maximum_vectors: u16,
},
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum VectorMetric {
Cosine,
DotProduct,
SquaredL2,
}
#[must_use]
pub fn is_canonical_field_path(path: &str) -> bool {
!path.is_empty()
&& path.len() <= MAX_MAPPING_FIELD_NAME_BYTES
&& !path.bytes().any(|byte| byte <= 0x1f || byte == 0x7f)
&& path.split('.').all(|component| !component.is_empty())
}
pub fn resolve_field_path<'a, T>(
root: &'a T,
path: &str,
member: impl FnMut(&'a T, &str) -> Option<&'a T>,
) -> Option<&'a T> {
if !is_canonical_field_path(path) {
return None;
}
path.split('.').try_fold(root, member)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FieldPathResolution<'a, T> {
Missing,
Found(&'a T),
Blocked,
}
pub fn resolve_field_path_checked<'a, T>(
root: &'a T,
path: &str,
mut member: impl FnMut(&'a T, &str) -> Option<&'a T>,
is_container: impl Fn(&T) -> bool,
) -> FieldPathResolution<'a, T> {
if !is_canonical_field_path(path) {
return FieldPathResolution::Missing;
}
let mut current = root;
for component in path.split('.') {
if !is_container(current) {
return FieldPathResolution::Blocked;
}
let Some(next) = member(current, component) else {
return FieldPathResolution::Missing;
};
current = next;
}
FieldPathResolution::Found(current)
}
pub fn validate_mapping_field_name(name: &str) -> Result<(), MappingValidationError> {
if !is_canonical_field_path(name) || name.starts_with('_') {
return Err(MappingValidationError::InvalidFieldName(name.to_owned()));
}
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MappingValidationError {
UnsupportedVersion(u16),
ZeroSchemaGeneration,
TooManyFields(usize),
InvalidFieldName(String),
ConflictingFieldPaths { ancestor: String, descendant: String },
UnsupportedVectorVersion(u16),
UnsupportedSparseVectorVersion(u16),
UnsupportedVectorSetVersion(u16),
TooManyVectorFields(usize),
InvalidVectorDimensions { field: String, dimensions: u16 },
InvalidVectorElementContract(String),
InvalidSparseVectorContract(String),
InvalidVectorSetContract(String),
DuplicateVectorFieldId(VectorFieldId),
StoredVectorField(String),
MissingMultiField(String),
MissingMultiFieldSource { field: String, source: String },
InvalidMultiFieldPath(String),
InvalidMultiFieldEndpoint(String),
UnknownFieldMetadata(String),
AnalyzerOnNonTextField(String),
KeywordPolicyOnNonKeywordField(String),
InvalidKeywordLimit { field: String, limit: u32 },
InvalidKeywordNullValue(String),
InvalidNullValue(String),
InvalidNestedPath(String),
NestedPathTooDeep(String),
NestedPathOverlap { parent: String, child: String },
IncompleteAnalyzerCoverage,
IncompleteKeywordCoverage,
IncompleteNestedCoverage,
MetadataNotSorted(&'static str),
TooManyMappingMetadata,
TooManyNestedPaths(usize),
}
impl Display for MappingValidationError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedVersion(version) => write!(formatter, "unsupported mapping format version {version}"),
Self::ZeroSchemaGeneration => formatter.write_str("mapping schema generation must be nonzero"),
Self::TooManyFields(count) => write!(formatter, "mapping has {count} fields"),
Self::InvalidFieldName(name) => write!(formatter, "invalid mapped field name '{name}'"),
Self::ConflictingFieldPaths { ancestor, descendant } => write!(
formatter,
"mapped scalar path '{ancestor}' conflicts with descendant '{descendant}'"
),
Self::UnsupportedVectorVersion(version) => {
write!(formatter, "unsupported vector mapping format version {version}")
}
Self::UnsupportedSparseVectorVersion(version) => {
write!(formatter, "unsupported sparse-vector mapping format version {version}")
}
Self::UnsupportedVectorSetVersion(version) => {
write!(formatter, "unsupported vector-set mapping format version {version}")
}
Self::TooManyVectorFields(count) => write!(formatter, "mapping has {count} vector fields"),
Self::InvalidVectorDimensions { field, dimensions } => {
write!(formatter, "vector field '{field}' has invalid dimensions {dimensions}")
}
Self::InvalidVectorElementContract(field) => {
write!(
formatter,
"vector field '{field}' has invalid element or quantization semantics"
)
}
Self::InvalidSparseVectorContract(field) => {
write!(
formatter,
"sparse vector field '{field}' has invalid dimension or weight semantics"
)
}
Self::InvalidVectorSetContract(field) => {
write!(
formatter,
"vector-set field '{field}' has invalid element, dimension, or count semantics"
)
}
Self::DuplicateVectorFieldId(field_id) => {
write!(formatter, "vector field identity {field_id} is duplicated")
}
Self::StoredVectorField(field) => {
write!(
formatter,
"vector field '{field}' cannot use Tantivy stored-field semantics"
)
}
Self::MissingMultiField(field) => write!(formatter, "multi-field projection '{field}' has no field"),
Self::MissingMultiFieldSource { field, source } => {
write!(
formatter,
"multi-field projection '{field}' has missing source '{source}'"
)
}
Self::InvalidMultiFieldPath(field) => {
write!(
formatter,
"multi-field projection '{field}' is not a direct child of its source"
)
}
Self::InvalidMultiFieldEndpoint(field) => {
write!(
formatter,
"multi-field projection '{field}' has a nested or vector endpoint"
)
}
Self::UnknownFieldMetadata(field) => {
write!(formatter, "mapping metadata references unknown field '{field}'")
}
Self::AnalyzerOnNonTextField(field) => {
write!(formatter, "analyzer identity is attached to non-text field '{field}'")
}
Self::KeywordPolicyOnNonKeywordField(field) => {
write!(formatter, "keyword policy is attached to non-keyword field '{field}'")
}
Self::InvalidKeywordLimit { field, limit } => {
write!(formatter, "keyword field '{field}' has invalid ignore_above {limit}")
}
Self::InvalidKeywordNullValue(field) => {
write!(
formatter,
"keyword field '{field}' has a null_value beyond ignore_above"
)
}
Self::InvalidNullValue(field) => write!(formatter, "field '{field}' has an invalid null_value"),
Self::InvalidNestedPath(path) => write!(formatter, "nested path '{path}' has no nested field"),
Self::NestedPathTooDeep(path) => write!(formatter, "nested path '{path}' exceeds maximum depth"),
Self::NestedPathOverlap { parent, child } => {
write!(formatter, "nested path '{child}' is below nested path '{parent}'")
}
Self::IncompleteAnalyzerCoverage => {
formatter.write_str("every text field must have exactly one analyzer identity")
}
Self::IncompleteKeywordCoverage => {
formatter.write_str("every keyword field must have exactly one keyword limit")
}
Self::IncompleteNestedCoverage => formatter.write_str("nested_paths must exactly cover nested fields"),
Self::MetadataNotSorted(section) => {
write!(formatter, "mapping section '{section}' is not sorted and unique")
}
Self::TooManyMappingMetadata => formatter.write_str("mapping metadata exceeds the field bound"),
Self::TooManyNestedPaths(count) => write!(formatter, "mapping has {count} nested paths"),
}
}
}
impl Error for MappingValidationError {}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MappingCodecError {
TooLarge(usize),
DuplicateKey(String),
Malformed(String),
UnsupportedVersion(u16),
Invalid(MappingValidationError),
NonCanonical,
}
impl Display for MappingCodecError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::TooLarge(bytes) => write!(formatter, "mapping payload has {bytes} bytes"),
Self::DuplicateKey(reason) | Self::Malformed(reason) => formatter.write_str(reason),
Self::UnsupportedVersion(version) => write!(formatter, "unsupported mapping format version {version}"),
Self::Invalid(error) => write!(formatter, "invalid mapping: {error}"),
Self::NonCanonical => formatter.write_str("mapping payload is not canonical JSON"),
}
}
}
impl Error for MappingCodecError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Invalid(error) => Some(error),
_ => None,
}
}
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StorageClass {
#[default]
Standard,
Cold,
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheClass {
Pinned,
#[default]
Normal,
ColdTolerant,
}
impl CacheClass {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Pinned => "pinned",
Self::Normal => "normal",
Self::ColdTolerant => "cold_tolerant",
}
}
}
impl Display for CacheClass {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
impl StorageClass {
#[must_use]
pub const fn is_standard(&self) -> bool {
matches!(self, Self::Standard)
}
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Standard => "standard",
Self::Cold => "cold",
}
}
}
impl Display for StorageClass {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum LifecycleVerb {
Transition,
AggregateThenDelete,
}
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct DocumentId(String);
impl DocumentId {
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
if value.is_empty() {
return Err(IdentifierError::EmptyDocumentId);
}
if value.len() > 512 {
return Err(IdentifierError::DocumentIdTooLong(value.len()));
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
#[must_use]
pub fn into_inner(self) -> String {
self.0
}
}
impl<'de> Deserialize<'de> for DocumentId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(serde::de::Error::custom)
}
}
impl Display for DocumentId {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct IdempotencyKey(String);
impl IdempotencyKey {
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
if value.is_empty() {
return Err(IdentifierError::EmptyIdempotencyKey);
}
if value.len() > 256 {
return Err(IdentifierError::IdempotencyKeyTooLong(value.len()));
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl<'de> Deserialize<'de> for IdempotencyKey {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(serde::de::Error::custom)
}
}
impl Display for IdempotencyKey {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct PitId(String);
impl PitId {
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
if value.is_empty() {
return Err(IdentifierError::EmptyPitId);
}
if value.len() > 512 {
return Err(IdentifierError::PitIdTooLong(value.len()));
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl<'de> Deserialize<'de> for PitId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(serde::de::Error::custom)
}
}
impl Display for PitId {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
pub const SCOPED_PIT_ID_FORMAT_VERSION: u16 = 1;
#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ScopedPitId {
format_version: u16,
recovery_era: RecoveryEra,
nonce: PitId,
}
impl ScopedPitId {
#[must_use]
pub const fn new(recovery_era: RecoveryEra, nonce: PitId) -> Self {
Self {
format_version: SCOPED_PIT_ID_FORMAT_VERSION,
recovery_era,
nonce,
}
}
#[must_use]
pub const fn recovery_era(&self) -> RecoveryEra {
self.recovery_era
}
#[must_use]
pub const fn nonce(&self) -> &PitId {
&self.nonce
}
}
impl Ord for ScopedPitId {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.recovery_era
.to_be_bytes()
.cmp(&other.recovery_era.to_be_bytes())
.then_with(|| self.nonce.cmp(&other.nonce))
}
}
impl PartialOrd for ScopedPitId {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl<'de> Deserialize<'de> for ScopedPitId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Wire {
format_version: u16,
recovery_era: RecoveryEra,
nonce: PitId,
}
let wire = Wire::deserialize(deserializer)?;
if wire.format_version != SCOPED_PIT_ID_FORMAT_VERSION {
return Err(serde::de::Error::custom(
IdentifierError::UnsupportedScopedPitIdVersion(wire.format_version),
));
}
Ok(Self::new(wire.recovery_era, wire.nonce))
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct OperationFingerprint([u8; 32]);
impl OperationFingerprint {
#[must_use]
pub const fn new(bytes: [u8; 32]) -> Self {
Self(bytes)
}
#[must_use]
pub fn from_payload(bytes: &[u8]) -> Self {
Self(*blake3::hash(bytes).as_bytes())
}
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
}
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct ObjectKey(String);
impl ObjectKey {
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
if value.is_empty() {
return Err(IdentifierError::EmptyObjectKey);
}
if value.starts_with('/') || value.contains('\\') {
return Err(IdentifierError::InvalidObjectKey(value));
}
if value
.split('/')
.any(|component| component.is_empty() || component == "." || component == "..")
{
return Err(IdentifierError::InvalidObjectKey(value));
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
#[must_use]
pub fn into_inner(self) -> String {
self.0
}
#[must_use]
pub fn allocated_capacity_bytes(&self) -> usize {
self.0.capacity()
}
}
impl<'de> Deserialize<'de> for ObjectKey {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(serde::de::Error::custom)
}
}
impl Display for ObjectKey {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
pub const ROOT_SET_REF_FORMAT_VERSION: u16 = 2;
pub const MAX_ROOT_SET_PAGES: u32 = 256;
pub const MAX_ROOT_SET_SCHEMA_GENERATIONS: usize = 256;
pub const MAX_ROOT_SET_OBJECT_KEY_BYTES: usize = 1_024;
pub const PINNED_MANIFEST_BINDING_FORMAT_VERSION: u16 = 1;
#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PhysicalObjectId {
pub backend_authority: BackendAuthorityId,
pub object_incarnation: ObjectIncarnation,
pub object_key: ObjectKey,
}
impl PhysicalObjectId {
pub fn validate(&self) -> Result<(), RootSetRefError> {
let length = self.object_key.as_str().len();
if !(1..=MAX_ROOT_SET_OBJECT_KEY_BYTES).contains(&length) {
return Err(RootSetRefError::ObjectKeyLength(length));
}
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ObjectIdentity {
pub physical: PhysicalObjectId,
pub content_checksum: [u8; 32],
}
impl ObjectIdentity {
pub fn validate(&self) -> Result<(), RootSetRefError> {
self.physical.validate()
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RootSetRef {
pub format_version: u16,
pub root: PhysicalObjectId,
pub content_digest: [u8; 32],
pub item_count: u64,
pub page_count: u32,
pub schema_generations: Vec<SchemaGeneration>,
}
impl RootSetRef {
pub fn validate(&self) -> Result<(), RootSetRefError> {
if self.format_version != ROOT_SET_REF_FORMAT_VERSION {
return Err(RootSetRefError::UnsupportedVersion(self.format_version));
}
self.root.validate()?;
if self.item_count == 0
|| self.page_count == 0
|| self.page_count > MAX_ROOT_SET_PAGES
|| u64::from(self.page_count) > self.item_count
{
return Err(RootSetRefError::InvalidCounts {
items: self.item_count,
pages: self.page_count,
});
}
if self.schema_generations.len() > MAX_ROOT_SET_SCHEMA_GENERATIONS
|| self.schema_generations.windows(2).any(|pair| pair[0] >= pair[1])
{
return Err(RootSetRefError::InvalidSchemaGenerations);
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RootSetRefError {
UnsupportedVersion(u16),
ObjectKeyLength(usize),
InvalidCounts { items: u64, pages: u32 },
InvalidSchemaGenerations,
}
impl Display for RootSetRefError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedVersion(version) => write!(formatter, "unsupported root-set reference version {version}"),
Self::ObjectKeyLength(length) => {
write!(formatter, "root-set object key is {length} bytes; expected 1..=1024")
}
Self::InvalidCounts { items, pages } => {
write!(formatter, "root-set counts are invalid: items={items}, pages={pages}")
}
Self::InvalidSchemaGenerations => {
formatter.write_str("root-set schema generations are not bounded strict ascending values")
}
}
}
}
impl Error for RootSetRefError {}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum PinnedManifestDecodeProvenanceV1 {
ScopedManifestV3,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PinnedManifestBindingV1 {
pub format_version: u16,
pub shard_key: ShardKey,
pub admission_generation: AdmissionGeneration,
pub routing_revision: RoutingRevision,
pub data_backend_authority: BackendAuthorityId,
pub generation: ManifestGeneration,
pub fence: FenceToken,
pub primary_term: PrimaryTerm,
pub manifest: ObjectIdentity,
pub root_set: RootSetRef,
pub committed_through: SeqNo,
pub decode_provenance: PinnedManifestDecodeProvenanceV1,
}
impl PinnedManifestBindingV1 {
pub fn validate(&self) -> Result<(), PinnedManifestBindingError> {
if self.format_version != PINNED_MANIFEST_BINDING_FORMAT_VERSION {
return Err(PinnedManifestBindingError::UnsupportedVersion(self.format_version));
}
if self.admission_generation.get() == 0
|| self.routing_revision.get() == 0
|| self.generation.get() == 0
|| self.fence.epoch().get() == 0
|| self.fence.epoch() != self.primary_term.epoch_component()
{
return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
}
self.manifest
.validate()
.map_err(|error| PinnedManifestBindingError::InvalidObject(error.to_string()))?;
self.root_set
.validate()
.map_err(|error| PinnedManifestBindingError::InvalidRootSet(error.to_string()))?;
if self.manifest.physical.backend_authority != self.data_backend_authority
|| self.root_set.root.backend_authority != self.data_backend_authority
{
return Err(PinnedManifestBindingError::BackendAuthorityMismatch);
}
Ok(())
}
}
pub fn vector_snapshot_digest_v1(
scope: CollectionIndexRef,
schema_generation: SchemaGeneration,
roots: &[PinnedManifestBindingV1],
) -> Result<[u8; 32], PinnedManifestBindingError> {
if schema_generation.get() == 0
|| roots
.iter()
.any(|root| root.shard_key.collection_index_ref() != scope || root.validate().is_err())
|| roots
.windows(2)
.any(|pair| pair[0].shard_key.shard_id() >= pair[1].shard_key.shard_id())
{
return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
}
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone-vector-snapshot-v1\0");
hasher.update(&scope.collection_id().get().to_be_bytes());
hasher.update(&scope.index_id().get().to_be_bytes());
hasher.update(&schema_generation.get().to_be_bytes());
let count = u32::try_from(roots.len()).map_err(|_| PinnedManifestBindingError::InvalidSemanticIdentity)?;
hasher.update(&count.to_be_bytes());
for root in roots {
hasher.update(&root.shard_key.shard_id().get().to_be_bytes());
hasher.update(&root.admission_generation.get().to_be_bytes());
hasher.update(&root.routing_revision.get().to_be_bytes());
hasher.update(&root.generation.get().to_be_bytes());
hasher.update(&root.fence.epoch().get().to_be_bytes());
hasher.update(&root.primary_term.get().to_be_bytes());
hasher.update(&root.manifest.content_checksum);
hasher.update(&root.root_set.content_digest);
hasher.update(&root.committed_through.get().to_be_bytes());
}
Ok(*hasher.finalize().as_bytes())
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PinnedManifestBindingError {
UnsupportedVersion(u16),
InvalidSemanticIdentity,
InvalidObject(String),
InvalidRootSet(String),
BackendAuthorityMismatch,
}
impl Display for PinnedManifestBindingError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedVersion(version) => {
write!(formatter, "unsupported pinned manifest binding version {version}")
}
Self::InvalidSemanticIdentity => formatter.write_str("pinned manifest semantic identity is invalid"),
Self::InvalidObject(reason) => write!(formatter, "pinned manifest object is invalid: {reason}"),
Self::InvalidRootSet(reason) => write!(formatter, "pinned manifest root set is invalid: {reason}"),
Self::BackendAuthorityMismatch => formatter.write_str("pinned manifest authorities do not match"),
}
}
}
impl Error for PinnedManifestBindingError {}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum IdentifierError {
EmptyDocumentId,
DocumentIdTooLong(usize),
EmptyIdempotencyKey,
IdempotencyKeyTooLong(usize),
EmptyPitId,
PitIdTooLong(usize),
UnsupportedScopedPitIdVersion(u16),
EmptyObjectKey,
InvalidObjectKey(String),
}
impl Display for IdentifierError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::EmptyDocumentId => formatter.write_str("document id must not be empty"),
Self::DocumentIdTooLong(length) => {
write!(formatter, "document id is {length} bytes; maximum is 512")
}
Self::EmptyIdempotencyKey => formatter.write_str("idempotency key must not be empty"),
Self::IdempotencyKeyTooLong(length) => {
write!(formatter, "idempotency key is {length} bytes; maximum is 256")
}
Self::EmptyPitId => formatter.write_str("PIT id must not be empty"),
Self::PitIdTooLong(length) => {
write!(formatter, "PIT id is {length} bytes; maximum is 512")
}
Self::UnsupportedScopedPitIdVersion(version) => {
write!(formatter, "unsupported scoped PIT identifier version {version}")
}
Self::EmptyObjectKey => formatter.write_str("object key must not be empty"),
Self::InvalidObjectKey(key) => {
write!(formatter, "invalid relative object key '{key}'")
}
}
}
}
impl Error for IdentifierError {}
#[must_use]
pub const fn routing_hash_v1(bytes: &[u8]) -> RoutingHash {
let mut hash = 0x811c_9dc5_u32;
let mut offset = 0;
while offset < bytes.len() {
hash ^= bytes[offset] as u32;
hash = hash.wrapping_mul(0x0100_0193);
offset += 1;
}
RoutingHash::new(hash)
}
#[must_use]
pub fn identity_hash_v1(bytes: &[u8]) -> IdHash {
let digest = blake3::hash(bytes);
let digest_bytes = digest.as_bytes();
IdHash::new(u128::from_le_bytes([
digest_bytes[0],
digest_bytes[1],
digest_bytes[2],
digest_bytes[3],
digest_bytes[4],
digest_bytes[5],
digest_bytes[6],
digest_bytes[7],
digest_bytes[8],
digest_bytes[9],
digest_bytes[10],
digest_bytes[11],
digest_bytes[12],
digest_bytes[13],
digest_bytes[14],
digest_bytes[15],
]))
}
#[cfg(test)]
mod policy_axis_tests {
use super::{LifecycleVerb, StorageClass};
#[test]
fn reserved_policy_axis_names_are_stable() {
assert!(serde_json::to_string(&StorageClass::Standard).is_ok_and(|value| value == "\"standard\""));
assert!(serde_json::to_string(&StorageClass::Cold).is_ok_and(|value| value == "\"cold\""));
assert!(serde_json::to_string(&LifecycleVerb::Transition).is_ok_and(|value| value == "\"transition\""));
assert!(
serde_json::to_string(&LifecycleVerb::AggregateThenDelete)
.is_ok_and(|value| value == "\"aggregate_then_delete\"")
);
}
}
#[must_use]
pub fn crc32c(bytes: &[u8]) -> u32 {
let mut crc = u32::MAX;
for byte in bytes {
crc ^= u32::from(*byte);
for _ in 0..8 {
let mask = (crc & 1).wrapping_neg();
crc = (crc >> 1) ^ (0x82f6_3b78 & mask);
}
}
!crc
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum Role {
Gateway,
Indexer,
Searcher,
Compactor,
Metastore,
}
impl Role {
pub const ALL: [Self; 5] = [
Self::Gateway,
Self::Indexer,
Self::Searcher,
Self::Compactor,
Self::Metastore,
];
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Gateway => "gateway",
Self::Indexer => "indexer",
Self::Searcher => "searcher",
Self::Compactor => "compactor",
Self::Metastore => "metastore",
}
}
}
impl Display for Role {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
impl FromStr for Role {
type Err = RoleParseError;
fn from_str(value: &str) -> Result<Self, Self::Err> {
match value {
"gateway" => Ok(Self::Gateway),
"indexer" => Ok(Self::Indexer),
"searcher" => Ok(Self::Searcher),
"compactor" => Ok(Self::Compactor),
"metastore" => Ok(Self::Metastore),
_ => Err(RoleParseError::Unknown(value.to_owned())),
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RoleSelection {
roles: Vec<Role>,
}
impl RoleSelection {
#[must_use]
pub fn all() -> Self {
Self {
roles: Role::ALL.to_vec(),
}
}
pub fn parse(value: &str) -> Result<Self, RoleParseError> {
if value == "all" {
return Ok(Self::all());
}
let mut roles = Vec::new();
for candidate in value.split(',') {
if candidate.is_empty() {
return Err(RoleParseError::Empty);
}
let role = candidate.parse::<Role>()?;
if !roles.contains(&role) {
roles.push(role);
}
}
if roles.is_empty() {
return Err(RoleParseError::Empty);
}
roles.sort_unstable();
Ok(Self { roles })
}
#[must_use]
pub fn roles(&self) -> &[Role] {
&self.roles
}
#[must_use]
pub fn display_list(&self) -> String {
self.roles
.iter()
.map(|role| role.as_str())
.collect::<Vec<_>>()
.join(",")
}
}
impl Default for RoleSelection {
fn default() -> Self {
Self::all()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RoleParseError {
Empty,
Unknown(String),
}
impl Display for RoleParseError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Empty => formatter.write_str("role selection must not be empty"),
Self::Unknown(role) => write!(
formatter,
"unknown role '{role}'; expected all, gateway, indexer, searcher, compactor, or metastore"
),
}
}
}
impl Error for RoleParseError {}
pub fn process_fault_point(point: &str) {
process_fault_point_inner(point, None);
}
pub fn process_fault_point_with_evidence(point: &str, evidence: &str) {
const MAX_EVIDENCE_BYTES: usize = 512;
if evidence.is_empty()
|| evidence.len() > MAX_EVIDENCE_BYTES
|| !evidence
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'=' | b'.'))
{
return;
}
process_fault_point_inner(point, Some(evidence));
}
pub const GC_PROTECTION_OBSERVATION_PATH_ENV: &str = "LOD_TEST_P52_GC_PROTECTION_OBSERVATION_PATH";
pub const MAX_GC_PROTECTION_OBSERVATION_BYTES: usize = 8 * 1024 * 1024;
#[derive(Debug)]
pub enum GcProtectionObservationError {
InvalidPath,
Empty,
TooLarge {
actual: usize,
maximum: usize,
},
AlreadyPublished(PathBuf),
Io {
operation: &'static str,
path: PathBuf,
source: std::io::Error,
},
}
impl Display for GcProtectionObservationError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidPath => formatter.write_str("GC protection observation path has no file name"),
Self::Empty => formatter.write_str("GC protection observation is empty"),
Self::TooLarge { actual, maximum } => write!(
formatter,
"GC protection observation is {actual} bytes; maximum is {maximum}"
),
Self::AlreadyPublished(path) => {
write!(
formatter,
"GC protection observation was already published at {}",
path.display()
)
}
Self::Io {
operation,
path,
source,
} => write!(
formatter,
"GC protection observation {operation} failed for {}: {source}",
path.display()
),
}
}
}
impl Error for GcProtectionObservationError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Io { source, .. } => Some(source),
_ => None,
}
}
}
#[must_use]
pub fn gc_protection_observation_path() -> Option<PathBuf> {
std::env::var_os(GC_PROTECTION_OBSERVATION_PATH_ENV)
.filter(|value| !value.is_empty())
.map(PathBuf::from)
}
pub fn write_gc_protection_observation(path: &Path, encoded: &[u8]) -> Result<(), GcProtectionObservationError> {
if encoded.is_empty() {
return Err(GcProtectionObservationError::Empty);
}
if encoded.len() > MAX_GC_PROTECTION_OBSERVATION_BYTES {
return Err(GcProtectionObservationError::TooLarge {
actual: encoded.len(),
maximum: MAX_GC_PROTECTION_OBSERVATION_BYTES,
});
}
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.filter(|name| !name.is_empty())
.ok_or(GcProtectionObservationError::InvalidPath)?;
let temporary = path.with_file_name(format!(".{file_name}.{}.tmp", std::process::id()));
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary)
.map_err(|source| GcProtectionObservationError::Io {
operation: "temporary-file creation",
path: temporary.clone(),
source,
})?;
let write_result = file.write_all(encoded).and_then(|()| file.sync_all());
drop(file);
if let Err(source) = write_result {
let _cleanup_result = std::fs::remove_file(&temporary);
return Err(GcProtectionObservationError::Io {
operation: "temporary-file write",
path: temporary,
source,
});
}
if let Err(source) = std::fs::hard_link(&temporary, path) {
let _cleanup_result = std::fs::remove_file(&temporary);
if source.kind() == std::io::ErrorKind::AlreadyExists {
return Err(GcProtectionObservationError::AlreadyPublished(path.to_path_buf()));
}
return Err(GcProtectionObservationError::Io {
operation: "atomic publication",
path: path.to_path_buf(),
source,
});
}
std::fs::remove_file(&temporary).map_err(|source| GcProtectionObservationError::Io {
operation: "temporary-file cleanup",
path: temporary,
source,
})?;
Ok(())
}
fn process_fault_point_inner(point: &str, evidence: Option<&str>) {
struct FaultConfig {
point: String,
ready_path: PathBuf,
release_path: PathBuf,
timeout: Duration,
}
static CONFIG: OnceLock<Option<FaultConfig>> = OnceLock::new();
let config = CONFIG.get_or_init(|| {
let point = std::env::var("LOD_TEST_LISTENER_FAULT_POINT").ok()?;
let ready_path = std::env::var_os("LOD_TEST_LISTENER_FAULT_READY").map(PathBuf::from)?;
let release_path = std::env::var_os("LOD_TEST_LISTENER_FAULT_RELEASE").map(PathBuf::from)?;
let timeout_millis = std::env::var("LOD_TEST_LISTENER_FAULT_TIMEOUT_MILLIS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(30_000)
.min(60_000);
Some(FaultConfig {
point,
ready_path,
release_path,
timeout: Duration::from_millis(timeout_millis),
})
});
let Some(config) = config.as_ref().filter(|config| config.point == point) else {
return;
};
let payload = evidence.map_or_else(|| point.to_owned(), |fact| format!("{point}\n{fact}"));
if std::fs::write(&config.ready_path, payload.as_bytes()).is_err() {
return;
}
let deadline = Instant::now() + config.timeout;
while !config.release_path.exists() && Instant::now() < deadline {
std::thread::yield_now();
}
}
#[cfg(test)]
mod hash_backend_tests;
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use super::{
CollectionId, CollectionIndexRef, DocumentId, DurabilityClass, DynamicMappingProposalError,
DynamicMappingProposalV1, Epoch, FieldMapping, FieldType, GcProtectionObservationError, IndexId, IndexMode,
MAPPING_FORMAT_VERSION, MAX_GC_PROTECTION_OBSERVATION_BYTES, MAX_MAPPING_FIELD_NAME_BYTES, MAX_MAPPING_FIELDS,
ManifestGeneration, Mapping, MappingExtensions, MappingScalar, MappingValidationError, ObjectKey, PitId, Role,
RoleSelection, RoutingHash, RoutingRangeEnd, SchemaGeneration, ScopedPitId, SeqNo, ShardDescriptor, ShardId,
TaskId, VisibilityMode, crc32c, validate_mapping_field_name, write_gc_protection_observation,
};
static OBSERVATION_TEST_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[test]
fn object_key_memory_charge_preserves_spare_string_capacity() -> Result<(), Box<dyn std::error::Error>> {
let mut value = String::with_capacity(4096);
value.push_str("a/b");
let capacity = value.capacity();
let key = ObjectKey::new(value)?;
assert_eq!(key.as_str(), "a/b");
assert_eq!(key.allocated_capacity_bytes(), capacity);
assert!(key.allocated_capacity_bytes() > key.as_str().len());
assert_eq!(serde_json::to_string(&key)?, "\"a/b\"");
Ok(())
}
#[test]
fn all_expands_to_every_role() {
let selection = RoleSelection::parse("all");
assert_eq!(selection.as_ref().map(|value| value.roles()), Ok(Role::ALL.as_slice()));
}
#[test]
fn explicit_roles_are_deduplicated_and_stable() {
let selection = RoleSelection::parse("searcher,gateway,searcher");
assert_eq!(
selection.as_ref().map(|value| value.roles()),
Ok([Role::Gateway, Role::Searcher].as_slice())
);
}
#[test]
fn unknown_role_is_rejected() {
let selection = RoleSelection::parse("unknown");
assert!(selection.is_err());
}
#[test]
fn semantic_numbers_advance_without_crossing_types() {
assert_eq!(Epoch::new(7).checked_next(), Some(Epoch::new(8)));
assert_eq!(SeqNo::new(41).checked_next(), Some(SeqNo::new(42)));
assert_eq!(ManifestGeneration::new(u64::MAX).checked_next(), None);
}
#[test]
fn sparse_values_are_canonical_bounded_and_mapping_typed() -> Result<(), Box<dyn std::error::Error>> {
let definition = super::SparseVectorDefinition {
maximum_dimension: 12,
weight_type: super::SparseVectorWeightType::F16,
};
let value = super::SparseVectorValue {
entries: vec![
super::SparseVectorEntry {
dimension: 1,
weight: 1.5,
},
super::SparseVectorEntry {
dimension: 9,
weight: -2.0,
},
],
};
let decoded = value.decode(definition)?;
let native = value.canonical_native_bytes(definition)?;
assert_eq!(super::decode_sparse_native_bytes(definition, &native)?, decoded);
assert_eq!(native.len(), 12);
let encoded = serde_json::to_vec(&value)?;
assert_eq!(serde_json::from_slice::<super::SparseVectorValue>(&encoded)?, value);
#[derive(Debug, serde::Deserialize, PartialEq, serde::Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum TaggedSparse {
Sparse {
query_vector: super::SparseVectorValue,
#[serde(deserialize_with = "super::deserialize_sparse_weight")]
weight: f64,
},
}
let tagged = TaggedSparse::Sparse {
query_vector: value.clone(),
weight: 0.5,
};
let encoded = serde_json::to_vec(&tagged)?;
assert_eq!(serde_json::from_slice::<TaggedSparse>(&encoded)?, tagged);
for invalid in [
super::SparseVectorValue {
entries: vec![
super::SparseVectorEntry {
dimension: 1,
weight: 1.0,
},
super::SparseVectorEntry {
dimension: 1,
weight: 2.0,
},
],
},
super::SparseVectorValue {
entries: vec![super::SparseVectorEntry {
dimension: 2,
weight: 0.0,
}],
},
super::SparseVectorValue {
entries: vec![super::SparseVectorEntry {
dimension: 13,
weight: 1.0,
}],
},
super::SparseVectorValue {
entries: vec![super::SparseVectorEntry {
dimension: 2,
weight: f64::NAN,
}],
},
] {
assert!(invalid.decode(definition).is_err());
}
Ok(())
}
#[test]
fn vector_set_values_are_explicit_typed_and_bounded() -> Result<(), Box<dyn std::error::Error>> {
let definition = super::VectorSetDefinition {
dense: super::DenseVectorDefinition {
dimensions: 2,
element_type: super::DenseVectorElementType::F32,
metric: super::VectorMetric::DotProduct,
quantization_scale: None,
},
maximum_vectors: 2,
};
let value = super::VectorSetValue {
vectors: vec![
super::DenseVectorValue::Numeric { values: vec![1.0, 0.0] },
super::DenseVectorValue::Numeric { values: vec![0.0, 1.0] },
],
};
let decoded = value.decode(
definition,
super::MAX_VECTOR_SET_DOCUMENT_VECTORS,
super::MAX_VECTOR_SET_BYTES_PER_FIELD,
)?;
assert_eq!(decoded.vectors, vec![vec![1.0, 0.0], vec![0.0, 1.0]]);
assert_eq!(decoded.native_bytes, 16);
assert!(
super::VectorSetValue { vectors: Vec::new() }
.decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
.is_err()
);
assert!(
super::VectorSetValue {
vectors: vec![super::DenseVectorValue::Numeric { values: vec![1.0, 0.0] }; 3],
}
.decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
.is_err()
);
assert!(
super::VectorSetValue {
vectors: vec![super::DenseVectorValue::Numeric { values: vec![1.0] }],
}
.decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
.is_err()
);
let oversized_json = serde_json::json!({
"vectors": (0..=super::MAX_VECTOR_SET_DOCUMENT_VECTORS)
.map(|_| serde_json::json!({"kind": "numeric", "values": [1.0, 0.0]}))
.collect::<Vec<_>>()
});
assert!(serde_json::from_value::<super::VectorSetValue>(oversized_json).is_err());
Ok(())
}
#[test]
fn task_ids_round_trip_the_complete_u128_domain_and_reject_numbers() {
let task_id = TaskId::new(u128::MAX - 7);
let encoded = serde_json::to_string(&task_id);
let expected = format!("\"{}\"", u128::MAX - 7);
assert_eq!(encoded.as_ref().ok().map(String::as_str), Some(expected.as_str()));
assert_eq!(
encoded
.as_deref()
.ok()
.and_then(|encoded| serde_json::from_str(encoded).ok()),
Some(task_id)
);
assert!(serde_json::from_str::<TaskId>("41").is_err());
}
#[test]
fn gc_protection_observation_writer_is_bounded_atomic_and_single_assignment()
-> Result<(), Box<dyn std::error::Error>> {
let sequence = OBSERVATION_TEST_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let root = std::env::temp_dir().join(format!(
"lodestone-p52-gc-protection-observation-{}-{sequence}",
std::process::id()
));
std::fs::create_dir(&root)?;
let path = root.join("applied.json");
let encoded = br#"{"format_version":1}"#;
write_gc_protection_observation(&path, encoded)?;
assert_eq!(std::fs::read(&path)?, encoded);
assert!(matches!(
write_gc_protection_observation(&path, encoded),
Err(GcProtectionObservationError::AlreadyPublished(actual)) if actual == path
));
assert!(matches!(
write_gc_protection_observation(
&root.join("oversized.json"),
&vec![0; MAX_GC_PROTECTION_OBSERVATION_BYTES.saturating_add(1)]
),
Err(GcProtectionObservationError::TooLarge { .. })
));
assert!(!root.join("oversized.json").exists());
std::fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn durable_identifiers_reject_ambiguous_values() {
assert!(DocumentId::new("").is_err());
assert!(PitId::new("").is_err());
assert!(PitId::new("opaque-pit-lease").is_ok());
assert!(ObjectKey::new("../escape").is_err());
assert!(ObjectKey::new("valid/prefix/object").is_ok());
}
#[test]
fn scoped_pit_id_round_trips_and_rejects_newer_formats() -> Result<(), Box<dyn std::error::Error>> {
let pit_id = ScopedPitId::new(super::RecoveryEra::new(7)?, PitId::new("opaque-pit-lease")?);
let encoded = serde_json::to_vec(&pit_id)?;
assert_eq!(serde_json::from_slice::<ScopedPitId>(&encoded)?, pit_id);
let mut unsupported: serde_json::Value = serde_json::from_slice(&encoded)?;
let Some(object) = unsupported.as_object_mut() else {
return Err("scoped PIT identifier was not an object".into());
};
object.insert("format_version".to_owned(), serde_json::json!(2));
assert!(serde_json::from_value::<ScopedPitId>(unsupported).is_err());
Ok(())
}
#[test]
fn crc32c_matches_the_standard_check_value() {
assert_eq!(crc32c(b"123456789"), 0xe306_9283);
}
#[test]
fn durability_names_round_trip_and_admitted_classes_are_available() {
let classes = [
DurabilityClass::RegionalObject,
DurabilityClass::ZonalLowLatency,
DurabilityClass::Relaxed,
DurabilityClass::Ephemeral,
];
for class in classes {
let encoded = serde_json::to_string(&class);
assert_eq!(
encoded.as_ref().ok().and_then(|bytes| serde_json::from_str(bytes).ok()),
Some(class)
);
}
assert!(DurabilityClass::RegionalObject.require_available().is_ok());
assert!(DurabilityClass::ZonalLowLatency.require_available().is_ok());
assert!(DurabilityClass::Relaxed.require_available().is_err());
assert!(DurabilityClass::Ephemeral.require_available().is_err());
assert!(DurabilityClass::RegionalObject.require_state_available().is_ok());
assert!(DurabilityClass::ZonalLowLatency.require_state_available().is_ok());
assert!(DurabilityClass::Ephemeral.require_state_available().is_ok());
assert!(DurabilityClass::Relaxed.require_state_available().is_err());
assert!(
serde_json::to_string(&super::DurabilityBoundary::IncarnationMemory)
.is_ok_and(|encoded| encoded == "\"incarnation_memory\"")
);
assert!(serde_json::from_str::<DurabilityClass>("\"future\"").is_err());
assert!(matches!(
serde_json::from_str::<VisibilityMode>("\"wait_for\""),
Ok(VisibilityMode::WaitFor)
));
}
#[test]
fn shard_descriptor_covers_exactly_its_hash_range() {
let descriptor = ShardDescriptor::new(
ShardId::new(7),
Epoch::new(3),
RoutingHash::new(10),
RoutingRangeEnd::new(20).unwrap_or(RoutingRangeEnd::FULL),
);
assert!(descriptor.is_ok());
let descriptor = descriptor.unwrap_or(ShardDescriptor {
shard_id: ShardId::new(7),
epoch: Epoch::new(3),
hash_start: RoutingHash::new(10),
hash_end: RoutingRangeEnd::FULL,
});
assert!(!descriptor.contains(RoutingHash::new(9)));
assert!(descriptor.contains(RoutingHash::new(10)));
assert!(descriptor.contains(RoutingHash::new(19)));
assert!(!descriptor.contains(RoutingHash::new(20)));
}
#[test]
fn shared_mapping_model_validates_semantics_and_refuses_retired_versions_and_unknown_fields()
-> Result<(), Box<dyn std::error::Error>> {
let mapping = Mapping {
format_version: MAPPING_FORMAT_VERSION,
schema_generation: SchemaGeneration::new(1),
mode: IndexMode::Mutable,
fields: BTreeMap::from([(
"message".to_owned(),
FieldMapping {
field_type: FieldType::Text,
stored: false,
null_value: None,
},
)]),
extensions: crate::MappingExtensions::default(),
};
assert_eq!(mapping.validate(), Ok(()));
assert_eq!(mapping.validate_current(), Ok(()));
let canonical = serde_json::to_value(&mapping)?;
for retired_version in 1..MAPPING_FORMAT_VERSION {
let retired = Mapping {
format_version: retired_version,
..mapping.clone()
};
assert_eq!(
retired.validate(),
Err(MappingValidationError::UnsupportedVersion(retired_version))
);
assert!(matches!(
retired.encode_canonical(),
Err(super::MappingCodecError::Invalid(
MappingValidationError::UnsupportedVersion(version)
)) if version == retired_version
));
let mut retired_bytes = canonical.clone();
retired_bytes["format_version"] = serde_json::json!(retired_version);
let retired_bytes = serde_json::to_vec(&retired_bytes)?;
let original = retired_bytes.clone();
assert!(matches!(
Mapping::decode_canonical(&retired_bytes),
Err(super::MappingCodecError::UnsupportedVersion(version)) if version == retired_version
));
assert_eq!(retired_bytes, original);
assert!(serde_json::from_slice::<Mapping>(&retired_bytes).is_err());
}
let future = Mapping {
format_version: MAPPING_FORMAT_VERSION + 1,
..mapping.clone()
};
assert_eq!(
future.validate(),
Err(MappingValidationError::UnsupportedVersion(MAPPING_FORMAT_VERSION + 1))
);
let invalid = Mapping {
schema_generation: SchemaGeneration::new(0),
..mapping.clone()
};
assert_eq!(invalid.validate(), Err(MappingValidationError::ZeroSchemaGeneration));
assert!(
serde_json::from_str::<Mapping>(
r#"{"format_version":1,"schema_generation":1,"mode":"mutable","fields":{},"future":true}"#
)
.is_err()
);
for invalid in [".message", "message.", "object..message"] {
let mut invalid_path = mapping.clone();
invalid_path.fields = BTreeMap::from([(
invalid.to_owned(),
FieldMapping {
field_type: FieldType::Text,
stored: false,
null_value: None,
},
)]);
assert_eq!(
invalid_path.validate(),
Err(MappingValidationError::InvalidFieldName(invalid.to_owned()))
);
}
for (ancestor, descendant) in [("a", "a.b"), ("a.b", "a.b.c")] {
let mut conflicting = mapping.clone();
conflicting.fields = BTreeMap::from([
(
ancestor.to_owned(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
),
(
descendant.to_owned(),
FieldMapping {
field_type: FieldType::U64,
stored: false,
null_value: None,
},
),
]);
assert_eq!(
conflicting.validate(),
Err(MappingValidationError::ConflictingFieldPaths {
ancestor: ancestor.to_owned(),
descendant: descendant.to_owned(),
})
);
}
let mut siblings = mapping;
siblings.fields = BTreeMap::from([
(
"a.b".to_owned(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
),
(
"a.bc".to_owned(),
FieldMapping {
field_type: FieldType::U64,
stored: false,
null_value: None,
},
),
]);
assert_eq!(siblings.validate(), Ok(()));
Ok(())
}
#[test]
fn ec6_mapping_refuses_nested_paths_below_nested_paths() {
let nested = || FieldMapping {
field_type: FieldType::Nested,
stored: false,
null_value: None,
};
let mapping = Mapping {
format_version: MAPPING_FORMAT_VERSION,
schema_generation: SchemaGeneration::new(1),
mode: IndexMode::Mutable,
fields: BTreeMap::from([
("items".to_owned(), nested()),
("items-options".to_owned(), nested()),
("items.options".to_owned(), nested()),
]),
extensions: MappingExtensions {
nested_paths: vec![
"items".to_owned(),
"items-options".to_owned(),
"items.options".to_owned(),
],
..MappingExtensions::default()
},
};
assert_eq!(
mapping.validate(),
Err(MappingValidationError::NestedPathOverlap {
parent: "items".to_owned(),
child: "items.options".to_owned(),
})
);
let mut predecessor = Mapping::empty(IndexMode::Mutable);
predecessor.fields.insert(
"items.color".to_owned(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
);
let mut successor = predecessor.clone();
successor.schema_generation = SchemaGeneration::new(2);
successor.fields.insert("items".to_owned(), nested());
successor.extensions.nested_paths = vec!["items".to_owned()];
assert!(!successor.is_additive_successor_of(&predecessor));
let mut safe_successor = predecessor.clone();
safe_successor.schema_generation = SchemaGeneration::new(2);
safe_successor.fields.insert("other".to_owned(), nested());
safe_successor.fields.insert(
"other.color".to_owned(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
);
safe_successor.extensions.nested_paths = vec!["other".to_owned()];
assert!(safe_successor.is_additive_successor_of(&predecessor));
}
#[test]
fn ec5_dynamic_mapping_proposal_normalizes_signed_zero_and_refuses_noncanonical_decode()
-> Result<(), Box<dyn std::error::Error>> {
let proposal = DynamicMappingProposalV1::new(
CollectionIndexRef::new(CollectionId::new(7)?, IndexId::new(3)?),
SchemaGeneration::new(1),
"temperature".to_owned(),
MappingScalar::F64Bits((-0.0_f64).to_bits()),
)?;
assert_eq!(
proposal.first_non_null_candidate,
MappingScalar::F64Bits(0.0_f64.to_bits())
);
let mut noncanonical = proposal;
noncanonical.first_non_null_candidate = MappingScalar::F64Bits((-0.0_f64).to_bits());
noncanonical.proposal_digest = noncanonical.exact_digest()?;
let encoded = serde_json::to_vec(&noncanonical)?;
assert!(matches!(
DynamicMappingProposalV1::decode_canonical(&encoded),
Err(DynamicMappingProposalError::InvalidCandidate)
));
Ok(())
}
#[test]
fn ec5_deployment_mapping_fixtures_use_the_current_validated_format() -> Result<(), Box<dyn std::error::Error>> {
for encoded in [
include_str!("../../../fixtures/runtime/secure-strict-mapping.json"),
include_str!("../../../fixtures/performance/search-mapping.json"),
include_str!("../../../fixtures/reindex/mapping.json"),
include_str!("../../../tests/performance/es-comparison-mapping-v1.json"),
] {
let mapping: Mapping = serde_json::from_str(encoded)?;
mapping.validate_current()?;
assert_eq!(mapping.schema_generation, SchemaGeneration::new(1));
}
Ok(())
}
#[test]
fn public_replayable_source_mapping_contract_matches_current_model() -> Result<(), serde_json::Error> {
let contract: serde_json::Value = serde_json::from_str(include_str!(
"../../../public/development/replayable-source-mapping-v1.json"
))?;
assert_eq!(contract["mapping"]["format_version"], MAPPING_FORMAT_VERSION);
assert_eq!(contract["mapping"]["maximum_fields"], MAX_MAPPING_FIELDS);
assert_eq!(
contract["mapping"]["maximum_field_name_bytes"],
MAX_MAPPING_FIELD_NAME_BYTES
);
assert_eq!(
contract["mapping"]["scalar_field_types"],
serde_json::json!(["text", "keyword", "u64", "date", "i64", "f64", "bool"])
);
let mapping = Mapping {
format_version: MAPPING_FORMAT_VERSION,
schema_generation: SchemaGeneration::new(1),
mode: IndexMode::AppendLog,
fields: BTreeMap::from([
(
"message".to_owned(),
FieldMapping {
field_type: FieldType::Text,
stored: false,
null_value: None,
},
),
(
"record_id".to_owned(),
FieldMapping {
field_type: FieldType::Keyword,
stored: false,
null_value: None,
},
),
]),
extensions: crate::MappingExtensions::default(),
};
assert_eq!(
serde_json::Value::String(serde_json::to_string(&mapping)?),
contract["example"]
);
for invalid in ["", ".message", "message.", "object..message", "_reserved", "a\u{7f}b"] {
assert_eq!(
validate_mapping_field_name(invalid),
Err(MappingValidationError::InvalidFieldName(invalid.to_owned()))
);
}
Ok(())
}
#[test]
fn shared_field_path_resolver_uses_canonical_nested_components() {
#[derive(Default)]
struct Node {
children: BTreeMap<&'static str, Node>,
value: Option<u64>,
}
let root = Node {
children: BTreeMap::from([(
"metrics",
Node {
children: BTreeMap::from([(
"count",
Node {
children: BTreeMap::new(),
value: Some(7),
},
)]),
value: None,
},
)]),
value: None,
};
assert_eq!(
super::resolve_field_path(&root, "metrics.count", |node, component| node.children.get(component))
.and_then(|node| node.value),
Some(7)
);
assert!(
super::resolve_field_path(&root, "metrics..count", |node, component| node.children.get(component))
.is_none()
);
}
#[test]
fn root_set_reference_is_exact_bounded_and_canonical() -> Result<(), Box<dyn std::error::Error>> {
let root = super::RootSetRef {
format_version: super::ROOT_SET_REF_FORMAT_VERSION,
root: super::PhysicalObjectId {
backend_authority: super::BackendAuthorityId::new("standard-data")?,
object_incarnation: super::ObjectIncarnation::new(9)?,
object_key: super::ObjectKey::new("root-sets/0009.json")?,
},
content_digest: [7; 32],
item_count: 4,
page_count: 2,
schema_generations: vec![super::SchemaGeneration::new(1), super::SchemaGeneration::new(3)],
};
root.validate()?;
assert_eq!(
serde_json::from_slice::<super::RootSetRef>(&serde_json::to_vec(&root)?)?,
root
);
let mut duplicate = root.clone();
duplicate.schema_generations = vec![super::SchemaGeneration::new(1), super::SchemaGeneration::new(1)];
assert_eq!(
duplicate.validate(),
Err(super::RootSetRefError::InvalidSchemaGenerations)
);
let mut empty = root;
empty.item_count = 0;
assert_eq!(
empty.validate(),
Err(super::RootSetRefError::InvalidCounts { items: 0, pages: 2 })
);
let mut single_page = empty;
single_page.item_count = 1;
single_page.page_count = 1;
single_page.validate()?;
let mut too_many_pages = single_page.clone();
too_many_pages.page_count = 2;
assert_eq!(
too_many_pages.validate(),
Err(super::RootSetRefError::InvalidCounts { items: 1, pages: 2 })
);
let mut predecessor = single_page;
predecessor.format_version = 1;
assert_eq!(
predecessor.validate(),
Err(super::RootSetRefError::UnsupportedVersion(1))
);
Ok(())
}
#[test]
fn kubernetes_object_uid_digest_refuses_zero_on_construction_and_decode() -> Result<(), Box<dyn std::error::Error>>
{
assert!(super::KubernetesObjectUidDigest::new([0; 32]).is_err());
assert!(
serde_json::from_str::<super::KubernetesObjectUidDigest>(&format!("[{}]", "0,".repeat(31) + "0")).is_err()
);
let digest = super::KubernetesObjectUidDigest::new([7; 32])?;
let decoded: super::KubernetesObjectUidDigest = serde_json::from_slice(&serde_json::to_vec(&digest)?)?;
assert_eq!(decoded, digest);
Ok(())
}
#[test]
fn gateway_d1_semantic_identities_and_listener_profiles_are_exact() -> Result<(), Box<dyn std::error::Error>> {
assert!(super::GatewayIncarnation::new([0; 32]).is_err());
assert!(serde_json::from_str::<super::GatewayIncarnation>(&format!("[{}]", "0,".repeat(31) + "0")).is_err());
let incarnation = super::GatewayIncarnation::new([7; 32])?;
assert_eq!(
serde_json::from_slice::<super::GatewayIncarnation>(&serde_json::to_vec(&incarnation)?)?,
incarnation
);
assert_eq!(super::AuditRangeId::new(4).get(), 4);
assert_eq!(super::QuotaGrantId::new(5).get(), 5);
assert_eq!(super::QuotaWindow::new(6).get(), 6);
assert_eq!(super::ProcessMonotonicMillis::new(7).get(), 7);
assert_eq!(super::UnixMillis::new(8).get(), 8);
for (listener, profile) in [
(
super::PublicGatewayListenerV1::LodestoneApi,
super::PublicTlsProfileV1::Native,
),
(
super::PublicGatewayListenerV1::ElasticsearchHttp,
super::PublicTlsProfileV1::Http,
),
(
super::PublicGatewayListenerV1::OtlpHttp,
super::PublicTlsProfileV1::Http,
),
(
super::PublicGatewayListenerV1::JaegerHttp,
super::PublicTlsProfileV1::Http,
),
(
super::PublicGatewayListenerV1::OtlpGrpc,
super::PublicTlsProfileV1::Grpc,
),
(
super::PublicGatewayListenerV1::SubscriptionGrpc,
super::PublicTlsProfileV1::Grpc,
),
(
super::PublicGatewayListenerV1::SubscriptionWebSocket,
super::PublicTlsProfileV1::Http,
),
(super::PublicGatewayListenerV1::Resp, super::PublicTlsProfileV1::Resp),
] {
assert_eq!(listener.required_tls_profile(), profile);
}
Ok(())
}
#[test]
fn ordinary_segment_allocation_ignores_reserved_ids_and_stops_at_boundary() {
let reserved = super::SegmentId::new(super::RESERVED_SEGMENT_ID_NAMESPACE_START + 17);
assert_eq!(
super::next_ordinary_segment_id([super::SegmentId::new(7), reserved, super::SegmentId::new(9),]),
Some(super::SegmentId::new(10))
);
assert_eq!(
super::next_ordinary_segment_id([reserved]),
Some(super::SegmentId::new(1))
);
assert_eq!(
super::next_ordinary_segment_id([super::SegmentId::new(super::RESERVED_SEGMENT_ID_NAMESPACE_START - 1,)]),
None
);
}
#[test]
fn dense_numeric_and_binary_values_decode_to_identical_semantics() -> Result<(), Box<dyn std::error::Error>> {
use base64::Engine as _;
for (element_type, scale, numeric, bytes) in [
(
super::DenseVectorElementType::F32,
None,
vec![1.5, -2.0],
[1.5_f32.to_le_bytes(), (-2.0_f32).to_le_bytes()].concat(),
),
(
super::DenseVectorElementType::F16,
None,
vec![1.5, -2.0],
[0x3e00_u16.to_le_bytes(), 0xc000_u16.to_le_bytes()].concat(),
),
(
super::DenseVectorElementType::I8,
Some(super::VectorQuantizationScale::new(0.25)?),
vec![6.0, -8.0],
vec![6_u8, (-8_i8) as u8],
),
] {
let definition = super::DenseVectorDefinition {
dimensions: 2,
element_type,
metric: super::VectorMetric::DotProduct,
quantization_scale: scale,
};
let numeric = super::decode_dense_numeric_value(definition, &numeric)?;
let binary = super::decode_dense_binary_value(
definition,
&super::DenseVectorBinaryValue {
element_type,
byte_order: super::DenseVectorByteOrder::LittleEndian,
data_base64: super::BASE64_STANDARD.encode(bytes),
quantization_scale: scale,
},
)?;
assert_eq!(numeric, binary);
}
Ok(())
}
#[test]
fn dense_value_refuses_narrowing_and_noncanonical_binary_controls() -> Result<(), Box<dyn std::error::Error>> {
use base64::Engine as _;
let f16 = super::DenseVectorDefinition {
dimensions: 1,
element_type: super::DenseVectorElementType::F16,
metric: super::VectorMetric::DotProduct,
quantization_scale: None,
};
assert_eq!(super::decode_dense_numeric_value(f16, &[1.000_488_281_25])?, vec![1.0]);
assert_eq!(
super::decode_dense_numeric_value(f16, &[1.001_464_843_75])?,
vec![1.001_953_1]
);
assert!(matches!(
super::decode_dense_numeric_value(f16, &[70_000.0]),
Err(super::DenseVectorValueError::F16Overflow(0))
));
let i8_definition = super::DenseVectorDefinition {
dimensions: 1,
element_type: super::DenseVectorElementType::I8,
metric: super::VectorMetric::DotProduct,
quantization_scale: Some(super::VectorQuantizationScale::new(0.5)?),
};
assert!(matches!(
super::decode_dense_numeric_value(i8_definition, &[1.5]),
Err(super::DenseVectorValueError::I8ElementOutOfRange(0))
));
let encoded = super::BASE64_STANDARD.encode([1_u8]);
let mut binary = super::DenseVectorBinaryValue {
element_type: super::DenseVectorElementType::I8,
byte_order: super::DenseVectorByteOrder::LittleEndian,
data_base64: encoded,
quantization_scale: Some(super::VectorQuantizationScale::new(0.25)?),
};
assert_eq!(
super::decode_dense_binary_value(i8_definition, &binary),
Err(super::DenseVectorValueError::QuantizationScaleMismatch)
);
binary.quantization_scale = i8_definition.quantization_scale;
binary.data_base64.push('=');
assert!(matches!(
super::decode_dense_binary_value(i8_definition, &binary),
Err(super::DenseVectorValueError::InvalidBase64Length { .. })
));
let wrong_order =
br#"{"kind":"binary","element_type":"f32","byte_order":"big_endian","data_base64":"AAAAAA=="}"#;
assert!(serde_json::from_slice::<super::DenseVectorValue>(wrong_order).is_err());
let signed_zero = super::decode_dense_numeric_value(
super::DenseVectorDefinition {
dimensions: 2,
element_type: super::DenseVectorElementType::F32,
metric: super::VectorMetric::DotProduct,
quantization_scale: None,
},
&[-0.0, 0.0],
)?;
assert!(signed_zero.iter().all(|value| value.to_bits() == 0));
assert_eq!(
super::decode_dense_numeric_value(
super::DenseVectorDefinition {
dimensions: 2,
element_type: super::DenseVectorElementType::F32,
metric: super::VectorMetric::Cosine,
quantization_scale: None,
},
&[0.0, -0.0],
),
Err(super::DenseVectorValueError::ZeroNorm)
);
Ok(())
}
#[test]
fn dense_value_refuses_length_scale_and_numeric_boundary_controls() -> Result<(), Box<dyn std::error::Error>> {
use base64::Engine as _;
let f32_definition = super::DenseVectorDefinition {
dimensions: 2,
element_type: super::DenseVectorElementType::F32,
metric: super::VectorMetric::DotProduct,
quantization_scale: None,
};
for bytes in [vec![0_u8; 4], vec![0_u8; 12]] {
let value = super::DenseVectorBinaryValue {
element_type: super::DenseVectorElementType::F32,
byte_order: super::DenseVectorByteOrder::LittleEndian,
data_base64: super::BASE64_STANDARD.encode(bytes),
quantization_scale: None,
};
assert!(matches!(
super::decode_dense_binary_value(f32_definition, &value),
Err(super::DenseVectorValueError::InvalidBase64Length { .. })
));
}
let wrong_type = super::DenseVectorBinaryValue {
element_type: super::DenseVectorElementType::F16,
byte_order: super::DenseVectorByteOrder::LittleEndian,
data_base64: super::BASE64_STANDARD.encode([0_u8; 8]),
quantization_scale: None,
};
assert_eq!(
super::decode_dense_binary_value(f32_definition, &wrong_type),
Err(super::DenseVectorValueError::ElementTypeMismatch)
);
assert!(matches!(
super::decode_dense_numeric_value(f32_definition, &[f64::INFINITY, 0.0]),
Err(super::DenseVectorValueError::NonFiniteElement(0))
));
let f16_definition = super::DenseVectorDefinition {
dimensions: 1,
element_type: super::DenseVectorElementType::F16,
metric: super::VectorMetric::DotProduct,
quantization_scale: None,
};
assert_eq!(
super::decode_dense_numeric_value(f16_definition, &[2.0_f64.powi(-25)])?[0].to_bits(),
0
);
assert_eq!(
super::decode_dense_numeric_value(f16_definition, &[2.0_f64.powi(-24)])?[0],
2.0_f32.powi(-24)
);
assert_eq!(
super::decode_dense_numeric_value(f16_definition, &[3.0 * 2.0_f64.powi(-25)])?[0],
2.0_f32.powi(-23)
);
let infinite_half = super::DenseVectorBinaryValue {
element_type: super::DenseVectorElementType::F16,
byte_order: super::DenseVectorByteOrder::LittleEndian,
data_base64: super::BASE64_STANDARD.encode(0x7c00_u16.to_le_bytes()),
quantization_scale: None,
};
assert!(matches!(
super::decode_dense_binary_value(f16_definition, &infinite_half),
Err(super::DenseVectorValueError::NonFiniteElement(0))
));
let i8_definition = super::DenseVectorDefinition {
dimensions: 1,
element_type: super::DenseVectorElementType::I8,
metric: super::VectorMetric::DotProduct,
quantization_scale: Some(super::VectorQuantizationScale::new(1.0)?),
};
for value in [-129.0, 128.0] {
assert!(matches!(
super::decode_dense_numeric_value(i8_definition, &[value]),
Err(super::DenseVectorValueError::I8ElementOutOfRange(0))
));
}
for scale in [r#""0""#, r#""-1""#, r#""1.0""#] {
assert!(serde_json::from_str::<super::VectorQuantizationScale>(scale).is_err());
}
Ok(())
}
}