/// Receive a gRPC [`WriteRequest`] and dispatch it to the DML handler.
Receive a gRPC [`WriteRequest`] and dispatch it to the DML handler.
Receive a gRPC [ ` WriteRequest ` ] and dispatch it to the DML handler .
async fn write(
        &self,
        request: Request<WriteRequest>,
    ) -> Result<Response<WriteResponse>, Status> {
        let span_ctx: Option<SpanContext> = request.extensions().get().cloned();
        let database_batch = request
            .into_inner()
            .database_batch
            .ok_or_else(|| FieldViolation::required("database_batch"))?;
        let tables =
            mutable_batch_pb::decode::decode_database_batch(&database_batch).map_err(|e| {
                FieldViolation {
                    field: "database_batch".into(),
                    description: format!("Invalid DatabaseBatch: {}", e),
                }
            })?;
        let (row_count, column_count) =
            tables.values().fold((0, 0), |(acc_rows, acc_cols), batch| {
                let cols = batch
                    .schema(Selection::All)
                    .expect("failed to get schema")
                    .len();
                let rows = batch.rows();
                (acc_rows + rows, acc_cols + cols)
            });
        let namespace = database_batch
            .database_name
            .try_into()
            .map_err(|e| FieldViolation {
                field: "database_name".into(),
                description: format!("Invalid namespace: {}", e),
            })?;
        let num_tables = tables.len();
        debug!(
            num_tables,
            %namespace,
            "routing grpc write",
        );
        let summary = self
            .dml_handler
            .write(&namespace, tables, span_ctx)
            .await
            .map_err(|e| match e.into() {
                e @ DmlError::DatabaseNotFound(_) => Status::not_found(e.to_string()),
                e @ DmlError::Schema(_) => Status::aborted(e.to_string()),
                e @ (DmlError::Internal(_)
                | DmlError::WriteBuffer(_)
                | DmlError::NamespaceCreation(_)
                | DmlError::Partition(PartitionError::BatchWrite(_))) => {
                    Status::internal(e.to_string())
                }
            })?;
        self.write_metric_rows.inc(row_count as _);
        self.write_metric_columns.inc(column_count as _);
        self.write_metric_tables.inc(num_tables as _);
        let mut response = Response::new(WriteResponse {});
        let metadata = response.metadata_mut();
        metadata.insert(
            WRITE_TOKEN_GRPC_HEADER,
            AsciiMetadataValue::try_from(&summary.to_token()).map_err(|e| {
                Status::internal(format!(
                    "Could not convert WriteSummary token to AsciiMetadataValue: {e}"
                ))
            })?,
        );
        Ok(response)
    }
async fn write ( & self , request : Request < WriteRequest > , ) -> Result < Response < WriteResponse > , Status > { let span_ctx : Option < SpanContext > = request . extensions ( ) . get ( ) . cloned ( ) ; let database_batch = request . into_inner ( ) . database_batch . ok_or_else ( | | FieldViolation :: required ( " " ) ) ? ; let tables = mutable_batch_pb :: decode :: decode_database_batch ( & database_batch ) . map_err ( | e | { FieldViolation { field : " " . into ( ) , description : format ! ( " " e ) , } } ) ? ; let ( row_count , column_count ) = tables . values ( ) . fold ( ( 0 , 0 ) , | ( acc_rows , acc_cols ) , batch | { let cols = batch . schema ( Selection :: All ) . expect ( " " ) . len ( ) ; let rows = batch . rows ( ) ; ( acc_rows + rows , acc_cols + cols ) } ) ; let namespace = database_batch . database_name . try_into ( ) . map_err ( | e | FieldViolation { field : " " . into ( ) , description : format ! ( " " e ) , } ) ? ; let num_tables = tables . len ( ) ; debug ! ( num_tables namespace " " ) ; let summary = self . dml_handler . write ( & namespace , tables , span_ctx ) . await . map_err ( | e | match e . into ( ) { e @ DmlError :: DatabaseNotFound ( _ ) => Status :: not_found ( e . to_string ( ) ) , e @ DmlError :: Schema ( _ ) => Status :: aborted ( e . to_string ( ) ) , e @ ( DmlError :: Internal ( _ ) | DmlError :: WriteBuffer ( _ ) | DmlError :: NamespaceCreation ( _ ) | DmlError :: Partition ( PartitionError :: BatchWrite ( _ ) ) ) => { Status :: internal ( e . to_string ( ) ) } } ) ? ; self . write_metric_rows . inc ( row_count as _ ) ; self . write_metric_columns . inc ( column_count as _ ) ; self . write_metric_tables . inc ( num_tables as _ ) ; let mut response = Response :: new ( WriteResponse { } ) ; let metadata = response . metadata_mut ( ) ; metadata . insert ( WRITE_TOKEN_GRPC_HEADER , AsciiMetadataValue :: try_from ( & summary . to_token ( ) ) . map_err ( | e | { Status :: internal ( format ! ( " " ) ) } ) ? , ) ; Ok ( response ) }
Receive a gRPC [`WriteRequest`] and dispatch it to the DML handler.
Receive a gRPC [ ` WriteRequest ` ] and dispatch it to the DML handler .
&self
request
Request<WriteRequest>