/// Validate the schema of all the writes in `batches`.
///
/// # Errors
///
/// If `namespace` does not exist, [`SchemaError::NamespaceLookup`] is
/// returned.
///
/// If the schema validation fails due to a schema conflict in the request,
/// [`SchemaError::Conflict`] is returned.
///
/// If the schema validation fails due to a service limit being reached,
/// [`SchemaError::ServiceLimit`] is returned.
///
/// A request that fails validation on one or more tables fails the request
/// as a whole - calling this method has "all or nothing" semantics.
Validate the schema of all the writes in `batches`.
Validate the schema of all the writes in ` batches ` .
async fn write(
        &self,
        namespace: &DatabaseName<'static>,
        batches: Self::WriteInput,
        _span_ctx: Option<SpanContext>,
    ) -> Result<Self::WriteOutput, Self::WriteError> {
        let mut repos = self.catalog.repositories().await;
        // Load the namespace schema from the cache, falling back to pulling it
        // from the global catalog (if it exists).
        let schema = self.cache.get_schema(namespace);
        let schema = match schema {
            Some(v) => v,
            None => {
                // Pull the schema from the global catalog or error if it does
                // not exist.
                let schema = get_schema_by_name(namespace, repos.deref_mut())
                    .await
                    .map_err(|e| {
                        warn!(error=%e, %namespace, "failed to retrieve namespace schema");
                        SchemaError::NamespaceLookup(e)
                    })
                    .map(Arc::new)?;
                self.cache
                    .put_schema(namespace.clone(), Arc::clone(&schema));
                trace!(%namespace, "schema cache populated");
                schema
            }
        };
        let maybe_new_schema = validate_or_insert_schema(
            batches.iter().map(|(k, v)| (k.as_str(), v)),
            &schema,
            repos.deref_mut(),
        )
        .await
        .map_err(|e| {
            match e {
                // Schema conflicts
                CatalogError::ColumnTypeMismatch {
                    ref name,
                    ref existing,
                    ref new,
                } => {
                    warn!(
                        %namespace,
                        column_name=%name,
                        existing_column_type=%existing,
                        request_column_type=%new,
                        "schema conflict"
                    );
                    self.schema_conflict.inc(1);
                    SchemaError::Conflict(e)
                }
                // Service limits
                CatalogError::ColumnCreateLimitError { .. }
                | CatalogError::TableCreateLimitError { .. } => {
                    warn!(%namespace, error=%e, "service protection limit reached");
                    self.service_limit_hit.inc(1);
                    SchemaError::ServiceLimit(e)
                }
                e => {
                    error!(%namespace, error=%e, "schema validation failed");
                    SchemaError::UnexpectedCatalogError(e)
                }
            }
        })?
        .map(Arc::new);
        trace!(%namespace, "schema validation complete");
        // If the schema has been updated, immediately add it to the cache
        // (before passing through the write) in order to allow subsequent,
        // parallel requests to use it while waiting on this request to
        // complete.
        match maybe_new_schema {
            Some(v) => {
                // This call MAY overwrite a more-up-to-date cache entry if
                // racing with another request for the same namespace, but the
                // cache will eventually converge in subsequent requests.
                self.cache.put_schema(namespace.clone(), v);
                trace!(%namespace, "schema cache updated");
            }
            None => {
                trace!(%namespace, "schema unchanged");
            }
        }
        Ok(batches)
    }
async fn write ( & self , namespace : & DatabaseName < ' static > , batches : Self :: WriteInput , _span_ctx : Option < SpanContext > , ) -> Result < Self :: WriteOutput , Self :: WriteError > { let mut repos = self . catalog . repositories ( ) . await ; // Load the namespace schema from the cache, falling back to pulling it // from the global catalog (if it exists). let schema = self . cache . get_schema ( namespace ) ; let schema = match schema { Some ( v ) => v , None => { // Pull the schema from the global catalog or error if it does // not exist. let schema = get_schema_by_name ( namespace , repos . deref_mut ( ) ) . await . map_err ( | e | { warn ! ( error e namespace " " ) ; SchemaError :: NamespaceLookup ( e ) } ) . map ( Arc :: new ) ? ; self . cache . put_schema ( namespace . clone ( ) , Arc :: clone ( & schema ) ) ; trace ! ( namespace " " ) ; schema } } ; let maybe_new_schema = validate_or_insert_schema ( batches . iter ( ) . map ( | ( k , v ) | ( k . as_str ( ) , v ) ) , & schema , repos . deref_mut ( ) , ) . await . map_err ( | e | { match e { // Schema conflicts CatalogError :: ColumnTypeMismatch { ref name , ref existing , ref new , } => { warn ! ( namespace column_name name existing_column_type existing request_column_type new " " ) ; self . schema_conflict . inc ( 1 ) ; SchemaError :: Conflict ( e ) } // Service limits CatalogError :: ColumnCreateLimitError { .. } | CatalogError :: TableCreateLimitError { .. } => { warn ! ( namespace error e " " ) ; self . service_limit_hit . inc ( 1 ) ; SchemaError :: ServiceLimit ( e ) } e => { error ! ( namespace error e " " ) ; SchemaError :: UnexpectedCatalogError ( e ) } } } ) ? . map ( Arc :: new ) ; trace ! ( namespace " " ) ; // If the schema has been updated, immediately add it to the cache // (before passing through the write) in order to allow subsequent, // parallel requests to use it while waiting on this request to // complete. match maybe_new_schema { Some ( v ) => { // This call MAY overwrite a more-up-to-date cache entry if // racing with another request for the same namespace, but the // cache will eventually converge in subsequent requests. self . cache . put_schema ( namespace . clone ( ) , v ) ; trace ! ( namespace " " ) ; } None => { trace ! ( namespace " " ) ; } } Ok ( batches ) }
Validate the schema of all the writes in `batches`.
Validate the schema of all the writes in ` batches ` .
&self
namespace
&DatabaseName<'static>
batches
Self::WriteInput
_span_ctx
Option<SpanContext>
errors
If `namespace` does not exist, [`SchemaError::NamespaceLookup`] is
returned.

If the schema validation fails due to a schema conflict in the request,
[`SchemaError::Conflict`] is returned.

If the schema validation fails due to a service limit being reached,
[`SchemaError::ServiceLimit`] is returned.

A request that fails validation on one or more tables fails the request
as a whole - calling this method has "all or nothing" semantics.
If ` namespace ` does not exist [ ` SchemaError :: NamespaceLookup ` ] is returned . If the schema validation fails due to a schema conflict in the request [ ` SchemaError :: Conflict ` ] is returned . If the schema validation fails due to a service limit being reached [ ` SchemaError :: ServiceLimit ` ] is returned . A request that fails validation on one or more tables fails the request as a whole - calling this method has " all or nothing " semantics .