radixdb-executor 1.1.0

SQL binding, planning, and execution engine for RadixDB
Documentation
use radixdb_catalog::{
    CatalogEdge, CatalogGeneration, CatalogMutation, CatalogName, CatalogObject, CatalogPayload,
    EdgeKind, ObjectId, ObjectKind, ObjectPrecondition, TablePayload, ViewPayload,
};
use radixdb_core::{sha256_digest, Error, Result};
use radixdb_sql::ast::{CreateViewStatement, DropViewStatement, SelectStatement, Statement};

use super::transaction::{catalog_argument, DdlDelta, ObjectIdSource};

const OUTPUT_SIGNATURE_DOMAIN: &[u8] = b"radixdb.view.output.v1\0";

#[derive(Debug, Clone, Copy)]
pub struct ViewCatalog<'generation> {
    generation: &'generation CatalogGeneration,
    view: &'generation CatalogObject,
    payload: &'generation ViewPayload,
}

impl<'generation> ViewCatalog<'generation> {
    pub fn load(generation: &'generation CatalogGeneration, view_name: &str) -> Result<Self> {
        let view = generation
            .find_relation(ObjectId::BOOTSTRAP_NAMESPACE, view_name)
            .map_err(catalog_argument)?
            .filter(|object| object.kind() == ObjectKind::View)
            .ok_or_else(|| Error::ViewNotFound(view_name.to_owned()))?;
        let CatalogPayload::View(payload) = view.payload() else {
            return Err(Error::internal(
                "catalog admitted a view with a non-view payload",
            ));
        };
        validate_all_views(generation)?;
        Ok(Self {
            generation,
            view,
            payload,
        })
    }

    pub const fn object(&self) -> &'generation CatalogObject {
        self.view
    }

    pub const fn id(&self) -> ObjectId {
        self.view.id()
    }

    pub const fn payload(&self) -> &'generation ViewPayload {
        self.payload
    }

    pub fn dependencies(&self) -> impl ExactSizeIterator<Item = &'generation CatalogObject> + '_ {
        self.payload.dependency_ids().iter().map(|id| {
            self.generation
                .object(*id)
                .expect("validated view dependency ID must resolve")
        })
    }
}

#[derive(Debug, PartialEq, Eq)]
struct BoundViewDefinition {
    canonical_sql: String,
    dependency_ids: Vec<ObjectId>,
    output_signature: [u8; 32],
}

pub(super) fn bind_create_view(
    statement: &CreateViewStatement,
    generation: &CatalogGeneration,
    ids: &mut ObjectIdSource,
) -> Result<DdlDelta> {
    validate_all_views(generation)?;
    let namespace_id = ObjectId::BOOTSTRAP_NAMESPACE;
    let view_name = statement.view_name.value.as_str();
    let bound = bind_select(&statement.query, generation)?;
    if let Some(existing) = generation
        .find_relation(namespace_id, view_name)
        .map_err(catalog_argument)?
    {
        if statement.if_not_exists && existing.kind() == ObjectKind::View {
            let CatalogPayload::View(payload) = existing.payload() else {
                unreachable!("catalog kind/payload equality was validated")
            };
            let exact = payload.canonical_sql().as_str() == bound.canonical_sql
                && payload.dependency_ids() == bound.dependency_ids
                && payload.output_signature() == &bound.output_signature;
            if exact {
                ViewCatalog::load(generation, view_name)?;
                return Ok(DdlDelta::default());
            }
        }
        return Err(Error::ViewAlreadyExists(view_name.to_owned()));
    }

    let view_id = ids.next(generation)?;
    let payload = ViewPayload::new(
        bound.canonical_sql,
        bound.dependency_ids.clone(),
        bound.output_signature,
    )
    .map_err(catalog_argument)?;
    let view = CatalogObject::new(
        view_id,
        Some(namespace_id),
        Some(namespace_id),
        ObjectId::BOOTSTRAP_OWNER,
        CatalogName::new(view_name).map_err(catalog_argument)?,
        1,
        CatalogPayload::View(payload),
    )
    .map_err(catalog_argument)?;
    let mut edges = Vec::with_capacity(bound.dependency_ids.len() + 1);
    edges.push(CatalogEdge::new(
        namespace_id,
        view_id,
        EdgeKind::Contains,
        0,
    ));
    for (ordinal, dependency_id) in bound.dependency_ids.into_iter().enumerate() {
        edges.push(CatalogEdge::new(
            view_id,
            dependency_id,
            EdgeKind::DependsOn,
            u32::try_from(ordinal).map_err(|_| {
                Error::InvalidArgument("view has too many catalog dependencies".to_owned())
            })?,
        ));
    }
    Ok(DdlDelta {
        mutations: vec![CatalogMutation::create(view)],
        edge_additions: edges,
        ..DdlDelta::default()
    })
}

pub(super) fn bind_drop_view(
    statement: &DropViewStatement,
    generation: &CatalogGeneration,
) -> Result<DdlDelta> {
    let view_name = statement.view_name.value.as_str();
    let Some(view) = generation
        .find_relation(ObjectId::BOOTSTRAP_NAMESPACE, view_name)
        .map_err(catalog_argument)?
    else {
        return if statement.if_exists {
            Ok(DdlDelta::default())
        } else {
            Err(Error::ViewNotFound(view_name.to_owned()))
        };
    };
    if view.kind() != ObjectKind::View {
        return Err(Error::ViewNotFound(view_name.to_owned()));
    }
    if let Some(dependent) = generation.graph().dependents(view.id()).next() {
        return Err(Error::InvalidArgument(format!(
            "cannot drop view '{view_name}': catalog object '{}' depends on it",
            dependent.name().display().as_str()
        )));
    }
    Ok(DdlDelta {
        mutations: vec![CatalogMutation::drop(
            ObjectPrecondition::new(view.id(), ObjectKind::View, view.definition_revision())
                .map_err(catalog_argument)?,
        )],
        ..DdlDelta::default()
    })
}

fn bind_persisted_view(
    canonical_sql: &str,
    generation: &CatalogGeneration,
) -> Result<BoundViewDefinition> {
    let statements = radixdb_sql::parse_sql(canonical_sql)
        .map_err(|error| Error::Parse(format!("invalid persisted view SQL: {error}")))?;
    let [Statement::Select(select)] = statements.as_slice() else {
        return Err(Error::InvalidArgument(
            "persisted view definition must contain exactly one SELECT".to_owned(),
        ));
    };
    let rebound = bind_select(select, generation)?;
    if rebound.canonical_sql != canonical_sql {
        return Err(Error::InvalidArgument(
            "persisted view SQL is not canonical".to_owned(),
        ));
    }
    Ok(rebound)
}

pub(super) fn validate_all_views(generation: &CatalogGeneration) -> Result<()> {
    for view in generation.objects_of_kind(ObjectKind::View) {
        let CatalogPayload::View(payload) = view.payload() else {
            return Err(Error::internal(
                "catalog admitted a view with a non-view payload",
            ));
        };
        let rebound = bind_persisted_view(payload.canonical_sql().as_str(), generation)?;
        if rebound.dependency_ids != payload.dependency_ids() {
            return Err(Error::InvalidArgument(format!(
                "view '{}' dependency IDs do not match rebound canonical SQL",
                view.name().display().as_str()
            )));
        }
        if rebound.output_signature != *payload.output_signature() {
            return Err(Error::InvalidArgument(format!(
                "view '{}' output signature does not match rebound canonical SQL",
                view.name().display().as_str()
            )));
        }
    }
    Ok(())
}

fn bind_select(
    select: &SelectStatement,
    generation: &CatalogGeneration,
) -> Result<BoundViewDefinition> {
    let canonical_sql = select.to_string();
    let dependency_names = crate::mutation::view_binding::bind_from_select(select);
    let mut dependencies = dependency_names
        .into_iter()
        .map(|name| {
            generation
                .find_relation(ObjectId::BOOTSTRAP_NAMESPACE, &name)
                .map_err(catalog_argument)?
                .filter(|object| matches!(object.kind(), ObjectKind::Table | ObjectKind::View))
                .ok_or(Error::TableOrViewNotFound(name))
        })
        .collect::<Result<Vec<_>>>()?;
    dependencies.sort_unstable_by_key(|object| object.id());
    let dependency_ids = dependencies.iter().map(|object| object.id()).collect();
    let output_signature = compute_output_signature(&canonical_sql, &dependencies, generation)?;
    Ok(BoundViewDefinition {
        canonical_sql,
        dependency_ids,
        output_signature,
    })
}

fn compute_output_signature(
    canonical_sql: &str,
    dependencies: &[&CatalogObject],
    generation: &CatalogGeneration,
) -> Result<[u8; 32]> {
    let mut input = Vec::new();
    input.extend_from_slice(OUTPUT_SIGNATURE_DOMAIN);
    append_bytes(&mut input, canonical_sql.as_bytes());
    append_u64(&mut input, dependencies.len())?;
    for dependency in dependencies {
        input.extend_from_slice(dependency.id().as_bytes());
        input.extend_from_slice(&dependency.kind().tag().to_le_bytes());
        append_bytes(
            &mut input,
            dependency.name().normalized().as_str().as_bytes(),
        );
        match dependency.payload() {
            CatalogPayload::Table(payload) => {
                append_table_shape(&mut input, payload, generation)?;
            }
            CatalogPayload::View(payload) => {
                input.extend_from_slice(payload.output_signature());
            }
            _ => {
                return Err(Error::internal(
                    "view dependency is neither a table nor a view",
                ))
            }
        }
    }
    Ok(sha256_digest(&input))
}

fn append_table_shape(
    output: &mut Vec<u8>,
    table: &TablePayload,
    generation: &CatalogGeneration,
) -> Result<()> {
    append_u64(output, table.column_ids().len())?;
    for column_id in table.column_ids() {
        let column = generation
            .object(*column_id)
            .ok_or_else(|| Error::internal("table shape references a missing column"))?;
        let CatalogPayload::Column(payload) = column.payload() else {
            return Err(Error::internal(
                "table shape references a non-column payload",
            ));
        };
        output.extend_from_slice(column.id().as_bytes());
        append_bytes(output, column.name().normalized().as_str().as_bytes());
        output.push(payload.data_type().logical_type().as_u8());
        output.extend_from_slice(&payload.data_type().parameter_1().to_le_bytes());
        output.extend_from_slice(&payload.data_type().parameter_2().to_le_bytes());
        output.push(u8::from(payload.nullable()));
    }
    Ok(())
}

fn append_bytes(output: &mut Vec<u8>, value: &[u8]) {
    output.extend_from_slice(&(value.len() as u64).to_le_bytes());
    output.extend_from_slice(value);
}

fn append_u64(output: &mut Vec<u8>, value: usize) -> Result<()> {
    let value = u64::try_from(value)
        .map_err(|_| Error::InvalidArgument("view shape exceeds u64".to_owned()))?;
    output.extend_from_slice(&value.to_le_bytes());
    Ok(())
}