rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-05-20 15:00:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into dynamic-file-user-provider

Browse Source
This commit is contained in:
tison
2024-03-26 14:29:54 +08:00
parent b551556e15 7c1c6e8b8c
commit 31b051650d
81 changed files with 1737 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/catalog/src/kvbackend/manager.rs
View File

@@ -138,8 +138,7 @@ impl CatalogManager for KvBackendCatalogManager {
let stream = self
.table_metadata_manager
.catalog_manager()
.catalog_names()
.await;
.catalog_names();
let keys = stream
.try_collect::<Vec<_>>()
@@ -154,8 +153,7 @@ impl CatalogManager for KvBackendCatalogManager {
let stream = self
.table_metadata_manager
.schema_manager()
.schema_names(catalog)
.await;
.schema_names(catalog);
let mut keys = stream
.try_collect::<BTreeSet<_>>()
.await
@@ -171,8 +169,7 @@ impl CatalogManager for KvBackendCatalogManager {
let stream = self
.table_metadata_manager
.table_name_manager()
.tables(catalog, schema)
.await;
.tables(catalog, schema);
let mut tables = stream
.try_collect::<Vec<_>>()
.await
@@ -297,7 +294,6 @@ impl CatalogManager for KvBackendCatalogManager {
.table_metadata_manager
.table_name_manager()
.tables(catalog, schema)
.await
.map_ok(|(_, v)| v.table_id());
const BATCH_SIZE: usize = 128;
let user_tables = try_stream!({

36
src/client/src/region.rs
View File

@@ -14,7 +14,7 @@
use std::sync::Arc;
use api::v1::region::{QueryRequest, RegionRequest, RegionResponse};
use api::v1::region::{QueryRequest, RegionRequest};
use api::v1::ResponseHeader;
use arc_swap::ArcSwapOption;
use arrow_flight::Ticket;
@@ -23,7 +23,7 @@ use async_trait::async_trait;
use common_error::ext::{BoxedError, ErrorExt};
use common_error::status_code::StatusCode;
use common_grpc::flight::{FlightDecoder, FlightMessage};
use common_meta::datanode_manager::{AffectedRows, Datanode};
use common_meta::datanode_manager::{Datanode, HandleResponse};
use common_meta::error::{self as meta_error, Result as MetaResult};
use common_recordbatch::error::ExternalSnafu;
use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
@@ -46,7 +46,7 @@ pub struct RegionRequester {
#[async_trait]
impl Datanode for RegionRequester {
async fn handle(&self, request: RegionRequest) -> MetaResult<AffectedRows> {
async fn handle(&self, request: RegionRequest) -> MetaResult<HandleResponse> {
self.handle_inner(request).await.map_err(|err| {
if err.should_retry() {
meta_error::Error::RetryLater {
@@ -165,7 +165,7 @@ impl RegionRequester {
Ok(Box::pin(record_batch_stream))
}
async fn handle_inner(&self, request: RegionRequest) -> Result<AffectedRows> {
async fn handle_inner(&self, request: RegionRequest) -> Result<HandleResponse> {
let request_type = request
.body
.as_ref()
@@ -178,10 +178,7 @@ impl RegionRequester {
let mut client = self.client.raw_region_client()?;
let RegionResponse {
header,
affected_rows,
} = client
let response = client
.handle(request)
.await
.map_err(|e| {
@@ -195,19 +192,20 @@ impl RegionRequester {
})?
.into_inner();
check_response_header(header)?;
check_response_header(&response.header)?;
Ok(affected_rows as _)
Ok(HandleResponse::from_region_response(response))
}
pub async fn handle(&self, request: RegionRequest) -> Result<AffectedRows> {
pub async fn handle(&self, request: RegionRequest) -> Result<HandleResponse> {
self.handle_inner(request).await
}
}
pub fn check_response_header(header: Option<ResponseHeader>) -> Result<()> {
pub fn check_response_header(header: &Option<ResponseHeader>) -> Result<()> {
let status = header
.and_then(|header| header.status)
.as_ref()
.and_then(|header| header.status.as_ref())
.context(IllegalDatabaseResponseSnafu {
err_msg: "either response header or status is missing",
})?;
@@ -221,7 +219,7 @@ pub fn check_response_header(header: Option<ResponseHeader>) -> Result<()> {
})?;
ServerSnafu {
code,
msg: status.err_msg,
msg: status.err_msg.clone(),
}
.fail()
}
@@ -236,19 +234,19 @@ mod test {
#[test]
fn test_check_response_header() {
let result = check_response_header(None);
let result = check_response_header(&None);
assert!(matches!(
result.unwrap_err(),
IllegalDatabaseResponse { .. }
));
let result = check_response_header(Some(ResponseHeader { status: None }));
let result = check_response_header(&Some(ResponseHeader { status: None }));
assert!(matches!(
result.unwrap_err(),
IllegalDatabaseResponse { .. }
));
let result = check_response_header(Some(ResponseHeader {
let result = check_response_header(&Some(ResponseHeader {
status: Some(PbStatus {
status_code: StatusCode::Success as u32,
err_msg: String::default(),
@@ -256,7 +254,7 @@ mod test {
}));
assert!(result.is_ok());
let result = check_response_header(Some(ResponseHeader {
let result = check_response_header(&Some(ResponseHeader {
status: Some(PbStatus {
status_code: u32::MAX,
err_msg: String::default(),
@@ -267,7 +265,7 @@ mod test {
IllegalDatabaseResponse { .. }
));
let result = check_response_header(Some(ResponseHeader {
let result = check_response_header(&Some(ResponseHeader {
status: Some(PbStatus {
status_code: StatusCode::Internal as u32,
err_msg: "blabla".to_string(),

16
src/common/base/src/readable_size.rs
View File

@@ -1,20 +1,6 @@
// Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// This file is copied from https://github.com/tikv/raft-engine/blob/8dd2a39f359ff16f5295f35343f626e0c10132fa/src/util.rs
// This file is copied from https://github.com/tikv/raft-engine/blob/0.3.0/src/util.rs
use std::fmt::{self, Debug, Display, Write};
use std::ops::{Div, Mul};

1
src/common/meta/Cargo.toml
View File

@@ -53,6 +53,7 @@ strum.workspace = true
table.workspace = true
tokio.workspace = true
tonic.workspace = true
typetag = "0.2"
[dev-dependencies]
chrono.workspace = true

29
src/common/meta/src/datanode_manager.rs
View File

@@ -12,9 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::sync::Arc;
use api::v1::region::{QueryRequest, RegionRequest};
use api::v1::region::{QueryRequest, RegionRequest, RegionResponse};
pub use common_base::AffectedRows;
use common_recordbatch::SendableRecordBatchStream;
@@ -25,7 +26,7 @@ use crate::peer::Peer;
#[async_trait::async_trait]
pub trait Datanode: Send + Sync {
/// Handles DML, and DDL requests.
async fn handle(&self, request: RegionRequest) -> Result<AffectedRows>;
async fn handle(&self, request: RegionRequest) -> Result<HandleResponse>;
/// Handles query requests
async fn handle_query(&self, request: QueryRequest) -> Result<SendableRecordBatchStream>;
@@ -41,3 +42,27 @@ pub trait DatanodeManager: Send + Sync {
}
pub type DatanodeManagerRef = Arc<dyn DatanodeManager>;
/// This result struct is derived from [RegionResponse]
#[derive(Debug)]
pub struct HandleResponse {
pub affected_rows: AffectedRows,
pub extension: HashMap<String, Vec<u8>>,
}
impl HandleResponse {
pub fn from_region_response(region_response: RegionResponse) -> Self {
Self {
affected_rows: region_response.affected_rows as _,
extension: region_response.extension,
}
}
/// Creates one response without extension
pub fn new(affected_rows: AffectedRows) -> Self {
Self {
affected_rows,
extension: Default::default(),
}
}
}

1
src/common/meta/src/ddl.rs
View File

@@ -32,6 +32,7 @@ pub mod alter_table;
pub mod create_logical_tables;
pub mod create_table;
mod create_table_template;
pub mod drop_database;
pub mod drop_table;
pub mod table_meta;
#[cfg(any(test, feature = "testing"))]

142
src/common/meta/src/ddl/create_logical_tables.rs
View File

@@ -12,30 +12,40 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::ops::Deref;
use api::v1::region::region_request::Body as PbRegionRequest;
use api::v1::region::{CreateRequests, RegionRequest, RegionRequestHeader};
use api::v1::CreateTableExpr;
use api::v1::{CreateTableExpr, SemanticType};
use async_trait::async_trait;
use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
use common_telemetry::info;
use common_telemetry::tracing_context::TracingContext;
use common_telemetry::{info, warn};
use futures_util::future::join_all;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use snafu::{ensure, ResultExt};
use snafu::{ensure, OptionExt, ResultExt};
use store_api::metadata::ColumnMetadata;
use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
use store_api::storage::{RegionId, RegionNumber};
use strum::AsRefStr;
use table::metadata::{RawTableInfo, TableId};
use crate::cache_invalidator::Context;
use crate::ddl::create_table_template::{build_template, CreateRequestBuilder};
use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error, region_storage_path};
use crate::ddl::DdlContext;
use crate::error::{Result, TableAlreadyExistsSnafu};
use crate::error::{
DecodeJsonSnafu, MetadataCorruptionSnafu, Result, TableAlreadyExistsSnafu,
TableInfoNotFoundSnafu,
};
use crate::instruction::CacheIdent;
use crate::key::table_info::TableInfoValue;
use crate::key::table_name::TableNameKey;
use crate::key::table_route::TableRouteValue;
use crate::key::DeserializedValueWithBytes;
use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
use crate::peer::Peer;
use crate::rpc::ddl::CreateTableTask;
@@ -70,6 +80,7 @@ impl CreateLogicalTablesProcedure {
/// - Checks whether physical table exists.
/// - Checks whether logical tables exist.
/// - Allocates the table ids.
/// - Modify tasks to sort logical columns on their names.
///
/// Abort(non-retry):
/// - The physical table does not exist.
@@ -130,7 +141,7 @@ impl CreateLogicalTablesProcedure {
));
}
// Allocates table ids
// Allocates table ids and sort columns on their names.
for (task, table_id) in tasks.iter_mut().zip(already_exists_tables_ids.iter()) {
let table_id = if let Some(table_id) = table_id {
*table_id
@@ -141,6 +152,11 @@ impl CreateLogicalTablesProcedure {
.await?
};
task.set_table_id(table_id);
// sort columns in task
task.sort_columns();
common_telemetry::info!("[DEBUG] sorted task {:?}", task);
}
self.creator
@@ -163,11 +179,12 @@ impl CreateLogicalTablesProcedure {
self.create_regions(region_routes).await
}
/// Creates table metadata
/// Creates table metadata for logical tables and update corresponding physical
/// table's metadata.
///
/// Abort(not-retry):
/// - Failed to create table metadata.
pub async fn on_create_metadata(&self) -> Result<Status> {
pub async fn on_create_metadata(&mut self) -> Result<Status> {
let manager = &self.context.table_metadata_manager;
let physical_table_id = self.creator.data.physical_table_id();
let remaining_tasks = self.creator.data.remaining_tasks();
@@ -202,6 +219,42 @@ impl CreateLogicalTablesProcedure {
.map(|task| task.table_info.ident.table_id)
.collect::<Vec<_>>();
if !self.creator.data.physical_columns.is_empty() {
// fetch old physical table's info
let physical_table_info = self
.context
.table_metadata_manager
.get_full_table_info(self.creator.data.physical_table_id)
.await?
.0
.context(TableInfoNotFoundSnafu {
table_name: format!("table id - {}", self.creator.data.physical_table_id),
})?;
// generate new table info
let new_table_info = self
.creator
.data
.build_new_physical_table_info(&physical_table_info);
// update physical table's metadata
self.context
.table_metadata_manager
.update_table_info(physical_table_info, new_table_info)
.await?;
// invalid table cache
self.context
.cache_invalidator
.invalidate(
&Context::default(),
vec![CacheIdent::TableId(self.creator.data.physical_table_id)],
)
.await?;
} else {
warn!("No physical columns found, leaving the physical table's schema unchanged");
}
info!("Created {num_tables} tables {table_ids:?} metadata for physical table {physical_table_id}");
Ok(Status::done_with_output(table_ids))
@@ -269,11 +322,39 @@ impl CreateLogicalTablesProcedure {
});
}
join_all(create_region_tasks)
// collect response from datanodes
let raw_schemas = join_all(create_region_tasks)
.await
.into_iter()
.map(|response| {
response.map(|mut response| response.extension.remove(ALTER_PHYSICAL_EXTENSION_KEY))
})
.collect::<Result<Vec<_>>>()?;
if raw_schemas.is_empty() {
self.creator.data.state = CreateTablesState::CreateMetadata;
return Ok(Status::executing(false));
}
// verify all datanodes return the same raw schemas
// Safety: previous check ensures this vector is not empty.
let first = raw_schemas.first().unwrap();
ensure!(
raw_schemas.iter().all(|x| x == first),
MetadataCorruptionSnafu {
err_msg: "Raw schemas from datanodes are not the same"
}
);
// decode raw schemas and store it
if let Some(raw_schema) = first {
let physical_columns =
ColumnMetadata::decode_list(raw_schema).context(DecodeJsonSnafu)?;
self.creator.data.physical_columns = physical_columns;
} else {
warn!("creating logical table result doesn't contains extension key `{ALTER_PHYSICAL_EXTENSION_KEY}`,leaving the physical table's schema unchanged");
}
self.creator.data.state = CreateTablesState::CreateMetadata;
// Ensures the procedures after the crash start from the `DatanodeCreateRegions` stage.
@@ -351,6 +432,7 @@ impl TablesCreator {
table_ids_already_exists: vec![None; len],
physical_table_id,
physical_region_numbers: vec![],
physical_columns: vec![],
},
}
}
@@ -364,6 +446,7 @@ pub struct CreateTablesData {
table_ids_already_exists: Vec<Option<TableId>>,
physical_table_id: TableId,
physical_region_numbers: Vec<RegionNumber>,
physical_columns: Vec<ColumnMetadata>,
}
impl CreateTablesData {
@@ -414,6 +497,47 @@ impl CreateTablesData {
})
.collect::<Vec<_>>()
}
/// Generate the new physical table info.
///
/// This method will consumes the physical columns.
fn build_new_physical_table_info(
&mut self,
old_table_info: &DeserializedValueWithBytes<TableInfoValue>,
) -> RawTableInfo {
let mut raw_table_info = old_table_info.deref().table_info.clone();
let existing_primary_key = raw_table_info
.meta
.schema
.column_schemas
.iter()
.map(|col| col.name.clone())
.collect::<HashSet<_>>();
let primary_key_indices = &mut raw_table_info.meta.primary_key_indices;
let value_indices = &mut raw_table_info.meta.value_indices;
value_indices.clear();
let time_index = &mut raw_table_info.meta.schema.timestamp_index;
let columns = &mut raw_table_info.meta.schema.column_schemas;
columns.clear();
for (idx, col) in self.physical_columns.drain(..).enumerate() {
match col.semantic_type {
SemanticType::Tag => {
// push new primary key to the end.
if !existing_primary_key.contains(&col.column_schema.name) {
primary_key_indices.push(idx);
}
}
SemanticType::Field => value_indices.push(idx),
SemanticType::Timestamp => *time_index = Some(idx),
}
columns.push(col.column_schema);
}
raw_table_info
}
}
#[derive(Debug, Clone, Serialize, Deserialize, AsRefStr)]

171
src/common/meta/src/ddl/drop_database.rs Normal file
View File

@@ -0,0 +1,171 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
pub mod cursor;
pub mod end;
pub mod executor;
pub mod metadata;
pub mod start;
use std::fmt::Debug;
use common_procedure::error::{Error as ProcedureError, FromJsonSnafu, ToJsonSnafu};
use common_procedure::{
Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status,
};
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
use snafu::ResultExt;
use tonic::async_trait;
use self::start::DropDatabaseStart;
use crate::ddl::DdlContext;
use crate::error::Result;
use crate::key::table_name::TableNameValue;
use crate::lock_key::{CatalogLock, SchemaLock};
pub struct DropDatabaseProcedure {
/// The context of procedure runtime.
runtime_context: DdlContext,
context: DropDatabaseContext,
state: Box<dyn State>,
}
/// Target of dropping tables.
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub enum DropTableTarget {
Logical,
Physical,
}
/// Context of [DropDatabaseProcedure] execution.
pub struct DropDatabaseContext {
catalog: String,
schema: String,
drop_if_exists: bool,
tables: Option<BoxStream<'static, Result<(String, TableNameValue)>>>,
}
#[async_trait::async_trait]
#[typetag::serde(tag = "drop_database_state")]
pub(crate) trait State: Send + Debug {
/// Yields the next [State] and [Status].
async fn next(
&mut self,
ddl_ctx: &DdlContext,
ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)>;
}
impl DropDatabaseProcedure {
pub const TYPE_NAME: &'static str = "metasrv-procedure::DropDatabase";
pub fn new(catalog: String, schema: String, drop_if_exists: bool, context: DdlContext) -> Self {
Self {
runtime_context: context,
context: DropDatabaseContext {
catalog,
schema,
drop_if_exists,
tables: None,
},
state: Box::new(DropDatabaseStart),
}
}
pub fn from_json(json: &str, runtime_context: DdlContext) -> ProcedureResult<Self> {
let DropDatabaseOwnedData {
catalog,
schema,
drop_if_exists,
state,
} = serde_json::from_str(json).context(FromJsonSnafu)?;
Ok(Self {
runtime_context,
context: DropDatabaseContext {
catalog,
schema,
drop_if_exists,
tables: None,
},
state,
})
}
}
#[async_trait]
impl Procedure for DropDatabaseProcedure {
fn type_name(&self) -> &str {
Self::TYPE_NAME
}
async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
let state = &mut self.state;
let (next, status) = state
.next(&self.runtime_context, &mut self.context)
.await
.map_err(|e| {
if e.is_retry_later() {
ProcedureError::retry_later(e)
} else {
ProcedureError::external(e)
}
})?;
*state = next;
Ok(status)
}
fn dump(&self) -> ProcedureResult<String> {
let data = DropDatabaseData {
catalog: &self.context.catalog,
schema: &self.context.schema,
drop_if_exists: self.context.drop_if_exists,
state: self.state.as_ref(),
};
serde_json::to_string(&data).context(ToJsonSnafu)
}
fn lock_key(&self) -> LockKey {
let lock_key = vec![
CatalogLock::Read(&self.context.catalog).into(),
SchemaLock::write(&self.context.catalog, &self.context.schema).into(),
];
LockKey::new(lock_key)
}
}
#[derive(Debug, Serialize)]
struct DropDatabaseData<'a> {
// The catalog name
catalog: &'a str,
// The schema name
schema: &'a str,
drop_if_exists: bool,
state: &'a dyn State,
}
#[derive(Debug, Deserialize)]
struct DropDatabaseOwnedData {
// The catalog name
catalog: String,
// The schema name
schema: String,
drop_if_exists: bool,
state: Box<dyn State>,
}

141
src/common/meta/src/ddl/drop_database/cursor.rs Normal file
View File

@@ -0,0 +1,141 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_procedure::Status;
use futures::TryStreamExt;
use serde::{Deserialize, Serialize};
use snafu::OptionExt;
use table::metadata::TableId;
use super::executor::DropDatabaseExecutor;
use super::metadata::DropDatabaseRemoveMetadata;
use super::DropTableTarget;
use crate::ddl::drop_database::{DropDatabaseContext, State};
use crate::ddl::DdlContext;
use crate::error::{self, Result};
use crate::key::table_route::TableRouteValue;
use crate::key::DeserializedValueWithBytes;
use crate::table_name::TableName;
#[derive(Debug, Serialize, Deserialize)]
pub struct DropDatabaseCursor {
target: DropTableTarget,
}
impl DropDatabaseCursor {
/// Returns a new [DropDatabaseCursor].
pub fn new(target: DropTableTarget) -> Self {
Self { target }
}
fn handle_reach_end(
&mut self,
ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
match self.target {
DropTableTarget::Logical => {
// Consumes the tables stream.
ctx.tables.take();
Ok((
Box::new(DropDatabaseCursor::new(DropTableTarget::Physical)),
Status::executing(true),
))
}
DropTableTarget::Physical => Ok((
Box::new(DropDatabaseRemoveMetadata),
Status::executing(true),
)),
}
}
async fn handle_table(
&mut self,
ddl_ctx: &DdlContext,
ctx: &mut DropDatabaseContext,
table_name: String,
table_id: TableId,
table_route_value: DeserializedValueWithBytes<TableRouteValue>,
) -> Result<(Box<dyn State>, Status)> {
match (self.target, table_route_value.get_inner_ref()) {
(DropTableTarget::Logical, TableRouteValue::Logical(_))
| (DropTableTarget::Physical, TableRouteValue::Physical(_)) => {
// TODO(weny): Maybe we can drop the table without fetching the `TableInfoValue`
let table_info_value = ddl_ctx
.table_metadata_manager
.table_info_manager()
.get(table_id)
.await?
.context(error::TableNotFoundSnafu {
table_name: &table_name,
})?;
Ok((
Box::new(DropDatabaseExecutor::new(
TableName::new(&ctx.catalog, &ctx.schema, &table_name),
table_id,
table_info_value,
table_route_value,
self.target,
)),
Status::executing(true),
))
}
_ => Ok((
Box::new(DropDatabaseCursor::new(self.target)),
Status::executing(false),
)),
}
}
}
#[async_trait::async_trait]
#[typetag::serde]
impl State for DropDatabaseCursor {
async fn next(
&mut self,
ddl_ctx: &DdlContext,
ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
if ctx.tables.as_deref().is_none() {
let tables = ddl_ctx
.table_metadata_manager
.table_name_manager()
.tables(&ctx.catalog, &ctx.schema);
ctx.tables = Some(tables);
}
// Safety: must exist
match ctx.tables.as_mut().unwrap().try_next().await? {
Some((table_name, table_name_value)) => {
let table_id = table_name_value.table_id();
match ddl_ctx
.table_metadata_manager
.table_route_manager()
.table_route_storage()
.get_raw(table_id)
.await?
{
Some(table_route_value) => {
self.handle_table(ddl_ctx, ctx, table_name, table_id, table_route_value)
.await
}
None => Ok((
Box::new(DropDatabaseCursor::new(self.target)),
Status::executing(false),
)),
}
}
None => self.handle_reach_end(ctx),
}
}
}

22
src/partition/src/metrics.rs → src/common/meta/src/ddl/drop_database/end.rs
View File

@@ -11,3 +11,25 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_procedure::Status;
use serde::{Deserialize, Serialize};
use crate::ddl::drop_database::{DropDatabaseContext, State};
use crate::ddl::DdlContext;
use crate::error::Result;
#[derive(Debug, Serialize, Deserialize)]
pub struct DropDatabaseEnd;
#[async_trait::async_trait]
#[typetag::serde]
impl State for DropDatabaseEnd {
async fn next(
&mut self,
_: &DdlContext,
_: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
Ok((Box::new(DropDatabaseEnd), Status::done()))
}
}

109
src/common/meta/src/ddl/drop_database/executor.rs Normal file
View File

@@ -0,0 +1,109 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_procedure::Status;
use common_telemetry::info;
use serde::{Deserialize, Serialize};
use snafu::OptionExt;
use table::metadata::TableId;
use super::cursor::DropDatabaseCursor;
use super::{DropDatabaseContext, DropTableTarget};
use crate::ddl::drop_database::State;
use crate::ddl::drop_table::executor::DropTableExecutor;
use crate::ddl::DdlContext;
use crate::error::{self, Result};
use crate::key::table_info::TableInfoValue;
use crate::key::table_route::TableRouteValue;
use crate::key::DeserializedValueWithBytes;
use crate::region_keeper::OperatingRegionGuard;
use crate::rpc::router::operating_leader_regions;
use crate::table_name::TableName;
#[derive(Debug, Serialize, Deserialize)]
pub struct DropDatabaseExecutor {
table_name: TableName,
table_id: TableId,
table_info_value: DeserializedValueWithBytes<TableInfoValue>,
table_route_value: DeserializedValueWithBytes<TableRouteValue>,
target: DropTableTarget,
#[serde(skip)]
dropping_regions: Vec<OperatingRegionGuard>,
}
impl DropDatabaseExecutor {
/// Returns a new [DropDatabaseExecutor].
pub fn new(
table_name: TableName,
table_id: TableId,
table_info_value: DeserializedValueWithBytes<TableInfoValue>,
table_route_value: DeserializedValueWithBytes<TableRouteValue>,
target: DropTableTarget,
) -> Self {
Self {
table_name,
table_id,
table_info_value,
table_route_value,
target,
dropping_regions: vec![],
}
}
}
impl DropDatabaseExecutor {
fn register_dropping_regions(&mut self, ddl_ctx: &DdlContext) -> Result<()> {
let region_routes = self.table_route_value.region_routes()?;
let dropping_regions = operating_leader_regions(region_routes);
let mut dropping_region_guards = Vec::with_capacity(dropping_regions.len());
for (region_id, datanode_id) in dropping_regions {
let guard = ddl_ctx
.memory_region_keeper
.register(datanode_id, region_id)
.context(error::RegionOperatingRaceSnafu {
region_id,
peer_id: datanode_id,
})?;
dropping_region_guards.push(guard);
}
self.dropping_regions = dropping_region_guards;
Ok(())
}
}
#[async_trait::async_trait]
#[typetag::serde]
impl State for DropDatabaseExecutor {
async fn next(
&mut self,
ddl_ctx: &DdlContext,
_ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
self.register_dropping_regions(ddl_ctx)?;
let executor = DropTableExecutor::new(self.table_name.clone(), self.table_id, true);
executor
.on_remove_metadata(ddl_ctx, &self.table_info_value, &self.table_route_value)
.await?;
executor.invalidate_table_cache(ddl_ctx).await?;
executor
.on_drop_regions(ddl_ctx, &self.table_route_value)
.await?;
info!("Table: {}({}) is dropped", self.table_name, self.table_id);
Ok((
Box::new(DropDatabaseCursor::new(self.target)),
Status::executing(false),
))
}
}

43
src/common/meta/src/ddl/drop_database/metadata.rs Normal file
View File

@@ -0,0 +1,43 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_procedure::Status;
use serde::{Deserialize, Serialize};
use super::end::DropDatabaseEnd;
use crate::ddl::drop_database::{DropDatabaseContext, State};
use crate::ddl::DdlContext;
use crate::error::Result;
use crate::key::schema_name::SchemaNameKey;
#[derive(Debug, Serialize, Deserialize)]
pub struct DropDatabaseRemoveMetadata;
#[async_trait::async_trait]
#[typetag::serde]
impl State for DropDatabaseRemoveMetadata {
async fn next(
&mut self,
ddl_ctx: &DdlContext,
ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
ddl_ctx
.table_metadata_manager
.schema_manager()
.delete(SchemaNameKey::new(&ctx.catalog, &ctx.schema))
.await?;
return Ok((Box::new(DropDatabaseEnd), Status::done()));
}
}

65
src/common/meta/src/ddl/drop_database/start.rs Normal file
View File

@@ -0,0 +1,65 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_procedure::Status;
use serde::{Deserialize, Serialize};
use snafu::ensure;
use crate::ddl::drop_database::cursor::DropDatabaseCursor;
use crate::ddl::drop_database::end::DropDatabaseEnd;
use crate::ddl::drop_database::{DropDatabaseContext, DropTableTarget, State};
use crate::ddl::DdlContext;
use crate::error::{self, Result};
use crate::key::schema_name::SchemaNameKey;
#[derive(Debug, Serialize, Deserialize)]
pub struct DropDatabaseStart;
#[async_trait::async_trait]
#[typetag::serde]
impl State for DropDatabaseStart {
/// Checks whether schema exists.
/// - Early returns if schema not exists and `drop_if_exists` is `true`.
/// - Throws an error if schema not exists and `drop_if_exists` is `false`.
async fn next(
&mut self,
ddl_ctx: &DdlContext,
ctx: &mut DropDatabaseContext,
) -> Result<(Box<dyn State>, Status)> {
let exists = ddl_ctx
.table_metadata_manager
.schema_manager()
.exists(SchemaNameKey {
catalog: &ctx.catalog,
schema: &ctx.schema,
})
.await?;
if !exists && ctx.drop_if_exists {
return Ok((Box::new(DropDatabaseEnd), Status::done()));
}
ensure!(
exists,
error::SchemaNotFoundSnafu {
table_schema: &ctx.schema,
}
);
Ok((
Box::new(DropDatabaseCursor::new(DropTableTarget::Logical)),
Status::executing(true),
))
}
}

7
src/common/meta/src/ddl/tests/create_logical_tables.rs
View File

@@ -28,6 +28,7 @@ use common_telemetry::debug;
use store_api::storage::RegionId;
use table::metadata::RawTableInfo;
use crate::datanode_manager::HandleResponse;
use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
use crate::ddl::test_util::create_table::build_raw_table_info_from_expr;
use crate::ddl::test_util::{TestColumnDefBuilder, TestCreateTableExprBuilder};
@@ -36,7 +37,7 @@ use crate::error::{Error, Result};
use crate::key::table_route::TableRouteValue;
use crate::peer::Peer;
use crate::rpc::ddl::CreateTableTask;
use crate::test_util::{new_ddl_context, AffectedRows, MockDatanodeHandler, MockDatanodeManager};
use crate::test_util::{new_ddl_context, MockDatanodeHandler, MockDatanodeManager};
// Note: this code may be duplicated with others.
// However, it's by design, ensures the tests are easy to be modified or added.
@@ -332,9 +333,9 @@ pub struct NaiveDatanodeHandler;
#[async_trait::async_trait]
impl MockDatanodeHandler for NaiveDatanodeHandler {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<HandleResponse> {
debug!("Returning Ok(0) for request: {request:?}, peer: {peer:?}");
Ok(0)
Ok(HandleResponse::new(0))
}
async fn handle_query(

13
src/common/meta/src/ddl/tests/create_table.rs
View File

@@ -26,6 +26,7 @@ use common_procedure_test::MockContextProvider;
use common_recordbatch::SendableRecordBatchStream;
use common_telemetry::debug;
use crate::datanode_manager::HandleResponse;
use crate::ddl::create_table::CreateTableProcedure;
use crate::ddl::test_util::create_table::build_raw_table_info_from_expr;
use crate::ddl::test_util::{TestColumnDefBuilder, TestCreateTableExprBuilder};
@@ -34,11 +35,11 @@ use crate::error::{Error, Result};
use crate::key::table_route::TableRouteValue;
use crate::peer::Peer;
use crate::rpc::ddl::CreateTableTask;
use crate::test_util::{new_ddl_context, AffectedRows, MockDatanodeHandler, MockDatanodeManager};
use crate::test_util::{new_ddl_context, MockDatanodeHandler, MockDatanodeManager};
#[async_trait::async_trait]
impl MockDatanodeHandler for () {
async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<HandleResponse> {
unreachable!()
}
@@ -176,7 +177,7 @@ pub struct RetryErrorDatanodeHandler;
#[async_trait::async_trait]
impl MockDatanodeHandler for RetryErrorDatanodeHandler {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<HandleResponse> {
debug!("Returning retry later for request: {request:?}, peer: {peer:?}");
Err(Error::RetryLater {
source: BoxedError::new(
@@ -220,7 +221,7 @@ pub struct UnexpectedErrorDatanodeHandler;
#[async_trait::async_trait]
impl MockDatanodeHandler for UnexpectedErrorDatanodeHandler {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<HandleResponse> {
debug!("Returning mock error for request: {request:?}, peer: {peer:?}");
error::UnexpectedSnafu {
err_msg: "mock error",
@@ -260,9 +261,9 @@ pub struct NaiveDatanodeHandler;
#[async_trait::async_trait]
impl MockDatanodeHandler for NaiveDatanodeHandler {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<HandleResponse> {
debug!("Returning Ok(0) for request: {request:?}, peer: {peer:?}");
Ok(0)
Ok(HandleResponse::new(0))
}
async fn handle_query(

187
src/common/meta/src/ddl_manager.rs
View File

@@ -14,7 +14,9 @@
use std::sync::Arc;
use common_procedure::{watcher, Output, ProcedureId, ProcedureManagerRef, ProcedureWithId};
use common_procedure::{
watcher, BoxedProcedureLoader, Output, ProcedureId, ProcedureManagerRef, ProcedureWithId,
};
use common_telemetry::tracing_context::{FutureExt, TracingContext};
use common_telemetry::{debug, info, tracing};
use snafu::{ensure, OptionExt, ResultExt};
@@ -25,6 +27,7 @@ use crate::datanode_manager::DatanodeManagerRef;
use crate::ddl::alter_table::AlterTableProcedure;
use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
use crate::ddl::create_table::CreateTableProcedure;
use crate::ddl::drop_database::DropDatabaseProcedure;
use crate::ddl::drop_table::DropTableProcedure;
use crate::ddl::table_meta::TableMetadataAllocatorRef;
use crate::ddl::truncate_table::TruncateTableProcedure;
@@ -39,12 +42,12 @@ use crate::key::table_route::TableRouteValue;
use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
use crate::region_keeper::MemoryRegionKeeperRef;
use crate::rpc::ddl::DdlTask::{
AlterLogicalTables, AlterTable, CreateLogicalTables, CreateTable, DropLogicalTables, DropTable,
TruncateTable,
AlterLogicalTables, AlterTable, CreateLogicalTables, CreateTable, DropDatabase,
DropLogicalTables, DropTable, TruncateTable,
};
use crate::rpc::ddl::{
AlterTableTask, CreateTableTask, DropTableTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse,
TruncateTableTask,
AlterTableTask, CreateTableTask, DropDatabaseTask, DropTableTask, SubmitDdlTaskRequest,
SubmitDdlTaskResponse, TruncateTableTask,
};
use crate::rpc::procedure;
use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse};
@@ -54,6 +57,8 @@ use crate::ClusterId;
pub type DdlManagerRef = Arc<DdlManager>;
pub type BoxedProcedureLoaderFactory = dyn Fn(DdlContext) -> BoxedProcedureLoader;
/// The [DdlManager] provides the ability to execute Ddl.
pub struct DdlManager {
procedure_manager: ProcedureManagerRef,
@@ -64,8 +69,8 @@ pub struct DdlManager {
memory_region_keeper: MemoryRegionKeeperRef,
}
/// Returns a new [DdlManager] with all Ddl [BoxedProcedureLoader](common_procedure::procedure::BoxedProcedureLoader)s registered.
impl DdlManager {
/// Returns a new [DdlManager] with all Ddl [BoxedProcedureLoader](common_procedure::procedure::BoxedProcedureLoader)s registered.
pub fn try_new(
procedure_manager: ProcedureManagerRef,
datanode_clients: DatanodeManagerRef,
@@ -103,75 +108,72 @@ impl DdlManager {
}
fn register_loaders(&self) -> Result<()> {
let context = self.create_context();
self.procedure_manager
.register_loader(
let loaders: Vec<(&str, &BoxedProcedureLoaderFactory)> = vec![
(
CreateTableProcedure::TYPE_NAME,
Box::new(move |json| {
let context = context.clone();
CreateTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
}),
)
.context(RegisterProcedureLoaderSnafu {
type_name: CreateTableProcedure::TYPE_NAME,
})?;
let context = self.create_context();
self.procedure_manager
.register_loader(
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
CreateTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
})
},
),
(
CreateLogicalTablesProcedure::TYPE_NAME,
Box::new(move |json| {
let context = context.clone();
CreateLogicalTablesProcedure::from_json(json, context).map(|p| Box::new(p) as _)
}),
)
.context(RegisterProcedureLoaderSnafu {
type_name: CreateLogicalTablesProcedure::TYPE_NAME,
})?;
let context = self.create_context();
self.procedure_manager
.register_loader(
DropTableProcedure::TYPE_NAME,
Box::new(move |json| {
let context = context.clone();
DropTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
}),
)
.context(RegisterProcedureLoaderSnafu {
type_name: DropTableProcedure::TYPE_NAME,
})?;
let context = self.create_context();
self.procedure_manager
.register_loader(
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
CreateLogicalTablesProcedure::from_json(json, context)
.map(|p| Box::new(p) as _)
})
},
),
(
AlterTableProcedure::TYPE_NAME,
Box::new(move |json| {
let context = context.clone();
AlterTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
}),
)
.context(RegisterProcedureLoaderSnafu {
type_name: AlterTableProcedure::TYPE_NAME,
})?;
let context = self.create_context();
self.procedure_manager
.register_loader(
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
AlterTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
})
},
),
(
DropTableProcedure::TYPE_NAME,
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
DropTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
})
},
),
(
TruncateTableProcedure::TYPE_NAME,
Box::new(move |json| {
let context = context.clone();
TruncateTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
}),
)
.context(RegisterProcedureLoaderSnafu {
type_name: TruncateTableProcedure::TYPE_NAME,
})
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
TruncateTableProcedure::from_json(json, context).map(|p| Box::new(p) as _)
})
},
),
(
DropDatabaseProcedure::TYPE_NAME,
&|context: DdlContext| -> BoxedProcedureLoader {
Box::new(move |json: &str| {
let context = context.clone();
DropDatabaseProcedure::from_json(json, context).map(|p| Box::new(p) as _)
})
},
),
];
for (type_name, loader_factory) in loaders {
let context = self.create_context();
self.procedure_manager
.register_loader(type_name, loader_factory(context))
.context(RegisterProcedureLoaderSnafu { type_name })?;
}
Ok(())
}
#[tracing::instrument(skip_all)]
@@ -260,6 +262,24 @@ impl DdlManager {
self.submit_procedure(procedure_with_id).await
}
#[tracing::instrument(skip_all)]
/// Submits and executes a drop table task.
pub async fn submit_drop_database(
&self,
_cluster_id: ClusterId,
DropDatabaseTask {
catalog,
schema,
drop_if_exists,
}: DropDatabaseTask,
) -> Result<(ProcedureId, Option<Output>)> {
let context = self.create_context();
let procedure = DropDatabaseProcedure::new(catalog, schema, drop_if_exists, context);
let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
self.submit_procedure(procedure_with_id).await
}
#[tracing::instrument(skip_all)]
/// Submits and executes a truncate table task.
pub async fn submit_truncate_table_task(
@@ -529,6 +549,28 @@ async fn handle_create_logical_table_tasks(
})
}
async fn handle_drop_database_task(
ddl_manager: &DdlManager,
cluster_id: ClusterId,
drop_database_task: DropDatabaseTask,
) -> Result<SubmitDdlTaskResponse> {
let (id, _) = ddl_manager
.submit_drop_database(cluster_id, drop_database_task.clone())
.await?;
let procedure_id = id.to_string();
info!(
"Database {}.{} is dropped via procedure_id {id:?}",
drop_database_task.catalog, drop_database_task.schema
);
Ok(SubmitDdlTaskResponse {
key: procedure_id.into(),
table_id: None,
..Default::default()
})
}
/// TODO(dennis): let [`DdlManager`] implement [`ProcedureExecutor`] looks weird, find some way to refactor it.
#[async_trait::async_trait]
impl ProcedureExecutor for DdlManager {
@@ -564,6 +606,9 @@ impl ProcedureExecutor for DdlManager {
}
DropLogicalTables(_) => todo!(),
AlterLogicalTables(_) => todo!(),
DropDatabase(drop_database_task) => {
handle_drop_database_task(self, cluster_id, drop_database_task).await
}
}
}
.trace(span)

13
src/common/meta/src/error.rs
View File

@@ -267,6 +267,12 @@ pub enum Error {
location: Location,
},
#[snafu(display("Schema nod found, schema: {}", table_schema))]
SchemaNotFound {
table_schema: String,
location: Location,
},
#[snafu(display("Failed to rename table, reason: {}", reason))]
RenameTable { reason: String, location: Location },
@@ -472,9 +478,10 @@ impl ErrorExt for Error {
InvalidCatalogValue { source, .. } => source.status_code(),
ConvertAlterTableRequest { source, .. } => source.status_code(),
ParseProcedureId { .. } | InvalidNumTopics { .. } | EmptyCreateTableTasks { .. } => {
StatusCode::InvalidArguments
}
ParseProcedureId { .. }
| InvalidNumTopics { .. }
| EmptyCreateTableTasks { .. }
| SchemaNotFound { .. } => StatusCode::InvalidArguments,
}
}

4
src/common/meta/src/key.rs
View File

@@ -273,6 +273,10 @@ impl<T: Serialize + DeserializeOwned + TableMetaValue> DeserializedValueWithByte
self.inner
}
pub fn get_inner_ref(&self) -> &T {
&self.inner
}
/// Returns original `bytes`
pub fn get_raw_bytes(&self) -> Vec<u8> {
self.bytes.to_vec()

2
src/common/meta/src/key/catalog_name.rs
View File

@@ -123,7 +123,7 @@ impl CatalogManager {
self.kv_backend.exists(&raw_key).await
}
pub async fn catalog_names(&self) -> BoxStream<'static, Result<String>> {
pub fn catalog_names(&self) -> BoxStream<'static, Result<String>> {
let start_key = CatalogNameKey::range_start_key();
let req = RangeRequest::new().with_prefix(start_key.as_bytes());

10
src/common/meta/src/key/schema_name.rs
View File

@@ -173,8 +173,16 @@ impl SchemaManager {
.transpose()
}
/// Deletes a [SchemaNameKey].
pub async fn delete(&self, schema: SchemaNameKey<'_>) -> Result<()> {
let raw_key = schema.as_raw_key();
self.kv_backend.delete(&raw_key, false).await?;
Ok(())
}
/// Returns a schema stream, it lists all schemas belong to the target `catalog`.
pub async fn schema_names(&self, catalog: &str) -> BoxStream<'static, Result<String>> {
pub fn schema_names(&self, catalog: &str) -> BoxStream<'static, Result<String>> {
let start_key = SchemaNameKey::range_start_key(catalog);
let req = RangeRequest::new().with_prefix(start_key.as_bytes());

2
src/common/meta/src/key/table_name.rs
View File

@@ -241,7 +241,7 @@ impl TableNameManager {
self.kv_backend.exists(&raw_key).await
}
pub async fn tables(
pub fn tables(
&self,
catalog: &str,
schema: &str,

2
src/common/meta/src/range_stream.rs
View File

@@ -236,6 +236,8 @@ impl<K, V> Stream for PaginationStream<K, V> {
PaginationStreamState::Init => {
let factory = self.factory.take().expect("lost factory");
if !factory.more {
// Ensures the factory always exists.
self.factory = Some(factory);
return Poll::Ready(None);
}
let fut = factory.read_next().boxed();

219
src/common/meta/src/rpc/ddl.rs
View File

@@ -19,10 +19,13 @@ use api::v1::meta::{
AlterTableTask as PbAlterTableTask, AlterTableTasks as PbAlterTableTasks,
CreateTableTask as PbCreateTableTask, CreateTableTasks as PbCreateTableTasks,
DdlTaskRequest as PbDdlTaskRequest, DdlTaskResponse as PbDdlTaskResponse,
DropTableTask as PbDropTableTask, DropTableTasks as PbDropTableTasks, Partition, ProcedureId,
DropDatabaseTask as PbDropDatabaseTask, DropTableTask as PbDropTableTask,
DropTableTasks as PbDropTableTasks, Partition, ProcedureId,
TruncateTableTask as PbTruncateTableTask,
};
use api::v1::{AlterExpr, CreateTableExpr, DropTableExpr, TruncateTableExpr};
use api::v1::{
AlterExpr, CreateTableExpr, DropDatabaseExpr, DropTableExpr, SemanticType, TruncateTableExpr,
};
use base64::engine::general_purpose;
use base64::Engine as _;
use prost::Message;
@@ -43,6 +46,7 @@ pub enum DdlTask {
CreateLogicalTables(Vec<CreateTableTask>),
DropLogicalTables(Vec<DropTableTask>),
AlterLogicalTables(Vec<AlterTableTask>),
DropDatabase(DropDatabaseTask),
}
impl DdlTask {
@@ -79,6 +83,14 @@ impl DdlTask {
})
}
pub fn new_drop_database(catalog: String, schema: String, drop_if_exists: bool) -> Self {
DdlTask::DropDatabase(DropDatabaseTask {
catalog,
schema,
drop_if_exists,
})
}
pub fn new_alter_table(alter_table: AlterExpr) -> Self {
DdlTask::AlterTable(AlterTableTask { alter_table })
}
@@ -137,6 +149,9 @@ impl TryFrom<Task> for DdlTask {
Ok(DdlTask::AlterLogicalTables(tasks))
}
Task::DropDatabaseTask(drop_database) => {
Ok(DdlTask::DropDatabase(drop_database.try_into()?))
}
}
}
}
@@ -179,6 +194,7 @@ impl TryFrom<SubmitDdlTaskRequest> for PbDdlTaskRequest {
Task::AlterTableTasks(PbAlterTableTasks { tasks })
}
DdlTask::DropDatabase(task) => Task::DropDatabaseTask(task.try_into()?),
};
Ok(Self {
@@ -368,6 +384,44 @@ impl CreateTableTask {
pub fn set_table_id(&mut self, table_id: TableId) {
self.table_info.ident.table_id = table_id;
}
/// Sort the columns in [CreateTableExpr] and [RawTableInfo].
///
/// This function won't do any check or verification. Caller should
/// ensure this task is valid.
pub fn sort_columns(&mut self) {
// sort create table expr
// sort column_defs by name
self.create_table
.column_defs
.sort_unstable_by(|a, b| a.name.cmp(&b.name));
// compute new indices of sorted columns
// this part won't do any check or verification.
let mut primary_key_indices = Vec::with_capacity(self.create_table.primary_keys.len());
let mut value_indices =
Vec::with_capacity(self.create_table.column_defs.len() - primary_key_indices.len() - 1);
let mut timestamp_index = None;
for (index, col) in self.create_table.column_defs.iter().enumerate() {
if self.create_table.primary_keys.contains(&col.name) {
primary_key_indices.push(index);
} else if col.semantic_type == SemanticType::Timestamp as i32 {
timestamp_index = Some(index);
} else {
value_indices.push(index);
}
}
// overwrite table info
self.table_info
.meta
.schema
.column_schemas
.sort_unstable_by(|a, b| a.name.cmp(&b.name));
self.table_info.meta.schema.timestamp_index = timestamp_index;
self.table_info.meta.primary_key_indices = primary_key_indices;
self.table_info.meta.value_indices = value_indices;
}
}
impl Serialize for CreateTableTask {
@@ -519,7 +573,7 @@ impl TryFrom<PbTruncateTableTask> for TruncateTableTask {
fn try_from(pb: PbTruncateTableTask) -> Result<Self> {
let truncate_table = pb.truncate_table.context(error::InvalidProtoMsgSnafu {
err_msg: "expected drop table",
err_msg: "expected truncate table",
})?;
Ok(Self {
@@ -551,13 +605,62 @@ impl TryFrom<TruncateTableTask> for PbTruncateTableTask {
}
}
#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub struct DropDatabaseTask {
pub catalog: String,
pub schema: String,
pub drop_if_exists: bool,
}
impl TryFrom<PbDropDatabaseTask> for DropDatabaseTask {
type Error = error::Error;
fn try_from(pb: PbDropDatabaseTask) -> Result<Self> {
let DropDatabaseExpr {
catalog_name,
schema_name,
drop_if_exists,
} = pb.drop_database.context(error::InvalidProtoMsgSnafu {
err_msg: "expected drop database",
})?;
Ok(DropDatabaseTask {
catalog: catalog_name,
schema: schema_name,
drop_if_exists,
})
}
}
impl TryFrom<DropDatabaseTask> for PbDropDatabaseTask {
type Error = error::Error;
fn try_from(
DropDatabaseTask {
catalog,
schema,
drop_if_exists,
}: DropDatabaseTask,
) -> Result<Self> {
Ok(PbDropDatabaseTask {
drop_database: Some(DropDatabaseExpr {
catalog_name: catalog,
schema_name: schema,
drop_if_exists,
}),
})
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use api::v1::{AlterExpr, CreateTableExpr};
use datatypes::schema::SchemaBuilder;
use table::metadata::RawTableInfo;
use api::v1::{AlterExpr, ColumnDef, CreateTableExpr, SemanticType};
use datatypes::schema::{ColumnSchema, RawSchema, SchemaBuilder};
use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
use store_api::storage::ConcreteDataType;
use table::metadata::{RawTableInfo, RawTableMeta, TableType};
use table::test_util::table_info::test_table_info;
use super::{AlterTableTask, CreateTableTask};
@@ -589,4 +692,108 @@ mod tests {
let de = serde_json::from_slice(&output).unwrap();
assert_eq!(task, de);
}
#[test]
fn test_sort_columns() {
// construct RawSchema
let raw_schema = RawSchema {
column_schemas: vec![
ColumnSchema::new(
"column3".to_string(),
ConcreteDataType::string_datatype(),
true,
),
ColumnSchema::new(
"column1".to_string(),
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
ColumnSchema::new(
"column2".to_string(),
ConcreteDataType::float64_datatype(),
true,
),
],
timestamp_index: Some(1),
version: 0,
};
// construct RawTableMeta
let raw_table_meta = RawTableMeta {
schema: raw_schema,
primary_key_indices: vec![0],
value_indices: vec![2],
engine: METRIC_ENGINE_NAME.to_string(),
next_column_id: 0,
region_numbers: vec![0],
options: Default::default(),
created_on: Default::default(),
partition_key_indices: Default::default(),
};
// construct RawTableInfo
let raw_table_info = RawTableInfo {
ident: Default::default(),
meta: raw_table_meta,
name: Default::default(),
desc: Default::default(),
catalog_name: Default::default(),
schema_name: Default::default(),
table_type: TableType::Base,
};
// construct create table expr
let create_table_expr = CreateTableExpr {
column_defs: vec![
ColumnDef {
name: "column3".to_string(),
semantic_type: SemanticType::Tag as i32,
..Default::default()
},
ColumnDef {
name: "column1".to_string(),
semantic_type: SemanticType::Timestamp as i32,
..Default::default()
},
ColumnDef {
name: "column2".to_string(),
semantic_type: SemanticType::Field as i32,
..Default::default()
},
],
primary_keys: vec!["column3".to_string()],
..Default::default()
};
let mut create_table_task =
CreateTableTask::new(create_table_expr, Vec::new(), raw_table_info);
// Call the sort_columns method
create_table_task.sort_columns();
// Assert that the columns are sorted correctly
assert_eq!(
create_table_task.create_table.column_defs[0].name,
"column1".to_string()
);
assert_eq!(
create_table_task.create_table.column_defs[1].name,
"column2".to_string()
);
assert_eq!(
create_table_task.create_table.column_defs[2].name,
"column3".to_string()
);
// Assert that the table_info is updated correctly
assert_eq!(
create_table_task.table_info.meta.schema.timestamp_index,
Some(0)
);
assert_eq!(
create_table_task.table_info.meta.primary_key_indices,
vec![2]
);
assert_eq!(create_table_task.table_info.meta.value_indices, vec![1]);
}
}

8
src/common/meta/src/test_util.rs
View File

@@ -19,7 +19,9 @@ pub use common_base::AffectedRows;
use common_recordbatch::SendableRecordBatchStream;
use crate::cache_invalidator::DummyCacheInvalidator;
use crate::datanode_manager::{Datanode, DatanodeManager, DatanodeManagerRef, DatanodeRef};
use crate::datanode_manager::{
Datanode, DatanodeManager, DatanodeManagerRef, DatanodeRef, HandleResponse,
};
use crate::ddl::table_meta::TableMetadataAllocator;
use crate::ddl::DdlContext;
use crate::error::Result;
@@ -32,7 +34,7 @@ use crate::wal_options_allocator::WalOptionsAllocator;
#[async_trait::async_trait]
pub trait MockDatanodeHandler: Sync + Send + Clone {
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows>;
async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<HandleResponse>;
async fn handle_query(
&self,
@@ -62,7 +64,7 @@ struct MockDatanode<T> {
#[async_trait::async_trait]
impl<T: MockDatanodeHandler> Datanode for MockDatanode<T> {
async fn handle(&self, request: RegionRequest) -> Result<AffectedRows> {
async fn handle(&self, request: RegionRequest) -> Result<HandleResponse> {
self.handler.handle(&self.peer, request).await
}

6
src/common/procedure/src/lib.rs
View File

@@ -25,8 +25,8 @@ pub mod watcher;
pub use crate::error::{Error, Result};
pub use crate::procedure::{
BoxedProcedure, Context, ContextProvider, LockKey, Output, ParseIdError, Procedure,
ProcedureId, ProcedureManager, ProcedureManagerRef, ProcedureState, ProcedureWithId, Status,
StringKey,
BoxedProcedure, BoxedProcedureLoader, Context, ContextProvider, LockKey, Output, ParseIdError,
Procedure, ProcedureId, ProcedureManager, ProcedureManagerRef, ProcedureState, ProcedureWithId,
Status, StringKey,
};
pub use crate::watcher::Watcher;

8
src/common/procedure/src/local.rs
View File

@@ -799,8 +799,10 @@ mod tests {
let root_id = ProcedureId::random();
// Prepare data for the root procedure.
for step in 0..3 {
let type_name = root.type_name().to_string();
let data = root.dump().unwrap();
procedure_store
.store_procedure(root_id, step, &root, None)
.store_procedure(root_id, step, type_name, data, None)
.await
.unwrap();
}
@@ -809,8 +811,10 @@ mod tests {
let child_id = ProcedureId::random();
// Prepare data for the child procedure
for step in 0..2 {
let type_name = child.type_name().to_string();
let data = child.dump().unwrap();
procedure_store
.store_procedure(child_id, step, &child, Some(root_id))
.store_procedure(child_id, step, type_name, data, Some(root_id))
.await
.unwrap();
}

10
src/common/procedure/src/local/runner.rs
View File

@@ -385,7 +385,7 @@ impl Runner {
}
/// Extend the retry time to wait for the next retry.
async fn wait_on_err(&self, d: Duration, i: u64) {
async fn wait_on_err(&mut self, d: Duration, i: u64) {
logging::info!(
"Procedure {}-{} retry for the {} times after {} millis",
self.procedure.type_name(),
@@ -396,7 +396,7 @@ impl Runner {
time::sleep(d).await;
}
async fn on_suspended(&self, subprocedures: Vec<ProcedureWithId>) {
async fn on_suspended(&mut self, subprocedures: Vec<ProcedureWithId>) {
let has_child = !subprocedures.is_empty();
for subprocedure in subprocedures {
logging::info!(
@@ -429,11 +429,15 @@ impl Runner {
}
async fn persist_procedure(&mut self) -> Result<()> {
let type_name = self.procedure.type_name().to_string();
let data = self.procedure.dump()?;
self.store
.store_procedure(
self.meta.id,
self.step,
&self.procedure,
type_name,
data,
self.meta.parent_id,
)
.await

2
src/common/procedure/src/procedure.rs
View File

@@ -116,7 +116,7 @@ pub struct Context {
/// A `Procedure` represents an operation or a set of operations to be performed step-by-step.
#[async_trait]
pub trait Procedure: Send + Sync {
pub trait Procedure: Send {
/// Type name of the procedure.
fn type_name(&self) -> &str;

68
src/common/procedure/src/store.rs
View File

@@ -22,7 +22,7 @@ use snafu::ResultExt;
use crate::error::{Result, ToJsonSnafu};
pub(crate) use crate::store::state_store::StateStoreRef;
use crate::{BoxedProcedure, ProcedureId};
use crate::ProcedureId;
pub mod state_store;
@@ -75,14 +75,12 @@ impl ProcedureStore {
&self,
procedure_id: ProcedureId,
step: u32,
procedure: &BoxedProcedure,
type_name: String,
data: String,
parent_id: Option<ProcedureId>,
) -> Result<()> {
let type_name = procedure.type_name();
let data = procedure.dump()?;
let message = ProcedureMessage {
type_name: type_name.to_string(),
type_name,
data,
parent_id,
step,
@@ -312,6 +310,7 @@ mod tests {
use object_store::ObjectStore;
use crate::store::state_store::ObjectStateStore;
use crate::BoxedProcedure;
impl ProcedureStore {
pub(crate) fn from_object_store(store: ObjectStore) -> ProcedureStore {
@@ -481,9 +480,10 @@ mod tests {
let procedure_id = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("test store procedure"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 0, &procedure, None)
.store_procedure(procedure_id, 0, type_name, data, None)
.await
.unwrap();
@@ -507,9 +507,10 @@ mod tests {
let procedure_id = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("test store procedure"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 0, &procedure, None)
.store_procedure(procedure_id, 0, type_name, data, None)
.await
.unwrap();
store.commit_procedure(procedure_id, 1).await.unwrap();
@@ -526,9 +527,10 @@ mod tests {
let procedure_id = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("test store procedure"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 0, &procedure, None)
.store_procedure(procedure_id, 0, type_name, data, None)
.await
.unwrap();
store.rollback_procedure(procedure_id, 1).await.unwrap();
@@ -545,13 +547,16 @@ mod tests {
let procedure_id = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("test store procedure"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 0, &procedure, None)
.store_procedure(procedure_id, 0, type_name, data, None)
.await
.unwrap();
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 1, &procedure, None)
.store_procedure(procedure_id, 1, type_name, data, None)
.await
.unwrap();
@@ -570,12 +575,17 @@ mod tests {
let procedure_id = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("test store procedure"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 0, &procedure, None)
.store_procedure(procedure_id, 0, type_name, data, None)
.await
.unwrap();
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(procedure_id, 1, &procedure, None)
.store_procedure(procedure_id, 1, type_name, data, None)
.await
.unwrap();
store.commit_procedure(procedure_id, 2).await.unwrap();
@@ -595,31 +605,41 @@ mod tests {
// store 3 steps
let id0 = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id0-0"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id0, 0, &procedure, None)
.store_procedure(id0, 0, type_name, data, None)
.await
.unwrap();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id0-1"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id0, 1, &procedure, None)
.store_procedure(id0, 1, type_name, data, None)
.await
.unwrap();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id0-2"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id0, 2, &procedure, None)
.store_procedure(id0, 2, type_name, data, None)
.await
.unwrap();
// store 2 steps and then commit
let id1 = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id1-0"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id1, 0, &procedure, None)
.store_procedure(id1, 0, type_name, data, None)
.await
.unwrap();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id1-1"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id1, 1, &procedure, None)
.store_procedure(id1, 1, type_name, data, None)
.await
.unwrap();
store.commit_procedure(id1, 2).await.unwrap();
@@ -627,8 +647,10 @@ mod tests {
// store 1 step
let id2 = ProcedureId::random();
let procedure: BoxedProcedure = Box::new(MockProcedure::new("id2-0"));
let type_name = procedure.type_name().to_string();
let data = procedure.dump().unwrap();
store
.store_procedure(id2, 0, &procedure, None)
.store_procedure(id2, 0, type_name, data, None)
.await
.unwrap();

37
src/datanode/src/region_server.rs
View File

@@ -25,6 +25,7 @@ use async_trait::async_trait;
use bytes::Bytes;
use common_error::ext::BoxedError;
use common_error::status_code::StatusCode;
use common_meta::datanode_manager::HandleResponse;
use common_query::logical_plan::Expr;
use common_query::physical_plan::DfPhysicalPlanAdapter;
use common_query::{DfPhysicalPlan, OutputData};
@@ -128,7 +129,7 @@ impl RegionServer {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows> {
) -> Result<HandleResponse> {
self.inner.handle_request(region_id, request).await
}
@@ -267,11 +268,12 @@ impl RegionServerHandler for RegionServer {
results
};
// merge results by simply sum up affected rows.
// only insert/delete will have multiple results.
// merge results by sum up affected rows and merge extensions.
let mut affected_rows = 0;
let mut extension = HashMap::new();
for result in results {
affected_rows += result;
affected_rows += result.affected_rows;
extension.extend(result.extension);
}
Ok(RegionResponse {
@@ -282,6 +284,7 @@ impl RegionServerHandler for RegionServer {
}),
}),
affected_rows: affected_rows as _,
extension,
})
}
}
@@ -462,7 +465,7 @@ impl RegionServerInner {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows> {
) -> Result<HandleResponse> {
let request_type = request.request_type();
let _timer = crate::metrics::HANDLE_REGION_REQUEST_ELAPSED
.with_label_values(&[request_type])
@@ -487,7 +490,7 @@ impl RegionServerInner {
let engine = match self.get_engine(region_id, &region_change)? {
CurrentEngine::Engine(engine) => engine,
CurrentEngine::EarlyReturn(rows) => return Ok(rows),
CurrentEngine::EarlyReturn(rows) => return Ok(HandleResponse::new(rows)),
};
// Sets corresponding region status to registering/deregistering before the operation.
@@ -502,7 +505,10 @@ impl RegionServerInner {
// Sets corresponding region status to ready.
self.set_region_status_ready(region_id, engine, region_change)
.await?;
Ok(result)
Ok(HandleResponse {
affected_rows: result.affected_rows,
extension: result.extension,
})
}
Err(err) => {
// Removes the region status if the operation fails.
@@ -645,6 +651,7 @@ impl RegionServerInner {
.decode(Bytes::from(plan), catalog_list, "", "")
.await
.context(DecodeLogicalPlanSnafu)?;
let result = self
.query_engine
.execute(logical_plan.into(), ctx)
@@ -916,11 +923,11 @@ mod tests {
RegionEngineWithStatus::Registering(engine.clone()),
);
let affected_rows = mock_region_server
let response = mock_region_server
.handle_request(region_id, RegionRequest::Create(create_req))
.await
.unwrap();
assert_eq!(affected_rows, 0);
assert_eq!(response.affected_rows, 0);
let status = mock_region_server
.inner
@@ -931,7 +938,7 @@ mod tests {
assert!(matches!(status, RegionEngineWithStatus::Registering(_)));
let affected_rows = mock_region_server
let response = mock_region_server
.handle_request(
region_id,
RegionRequest::Open(RegionOpenRequest {
@@ -943,7 +950,7 @@ mod tests {
)
.await
.unwrap();
assert_eq!(affected_rows, 0);
assert_eq!(response.affected_rows, 0);
let status = mock_region_server
.inner
@@ -971,11 +978,11 @@ mod tests {
RegionEngineWithStatus::Deregistering(engine.clone()),
);
let affected_rows = mock_region_server
let response = mock_region_server
.handle_request(region_id, RegionRequest::Drop(RegionDropRequest {}))
.await
.unwrap();
assert_eq!(affected_rows, 0);
assert_eq!(response.affected_rows, 0);
let status = mock_region_server
.inner
@@ -990,11 +997,11 @@ mod tests {
RegionEngineWithStatus::Deregistering(engine.clone()),
);
let affected_rows = mock_region_server
let response = mock_region_server
.handle_request(region_id, RegionRequest::Close(RegionCloseRequest {}))
.await
.unwrap();
assert_eq!(affected_rows, 0);
assert_eq!(response.affected_rows, 0);
let status = mock_region_server
.inner

10
src/datanode/src/tests.rs
View File

@@ -31,7 +31,7 @@ use query::query_engine::DescribeResult;
use query::{QueryEngine, QueryEngineContext};
use session::context::QueryContextRef;
use store_api::metadata::RegionMetadataRef;
use store_api::region_engine::{RegionEngine, RegionRole, SetReadonlyResponse};
use store_api::region_engine::{RegionEngine, RegionHandleResult, RegionRole, SetReadonlyResponse};
use store_api::region_request::{AffectedRows, RegionRequest};
use store_api::storage::{RegionId, ScanRequest};
use table::TableRef;
@@ -166,16 +166,18 @@ impl RegionEngine for MockRegionEngine {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows, BoxedError> {
) -> Result<RegionHandleResult, BoxedError> {
if let Some(delay) = self.handle_request_delay {
tokio::time::sleep(delay).await;
}
if let Some(mock_fn) = &self.handle_request_mock_fn {
return mock_fn(region_id, request).map_err(BoxedError::new);
return mock_fn(region_id, request)
.map_err(BoxedError::new)
.map(RegionHandleResult::new);
};
let _ = self.sender.send((region_id, request)).await;
Ok(0)
Ok(RegionHandleResult::new(0))
}
async fn handle_query(

11
src/datatypes/src/schema/column_schema.rs
View File

@@ -143,11 +143,22 @@ impl ColumnSchema {
}
/// Set the nullablity to `true` of the column.
/// Similar to [set_nullable] but take the ownership and return a owned value.
///
/// [set_nullable]: Self::set_nullable
pub fn with_nullable_set(mut self) -> Self {
self.is_nullable = true;
self
}
/// Set the nullability to `true` of the column.
/// Similar to [with_nullable_set] but don't take the ownership
///
/// [with_nullable_set]: Self::with_nullable_set
pub fn set_nullable(&mut self) {
self.is_nullable = true;
}
/// Creates a new [`ColumnSchema`] with given metadata.
pub fn with_metadata(mut self, metadata: Metadata) -> Self {
self.metadata = metadata;

11
src/file-engine/src/engine.rs
View File

@@ -24,7 +24,7 @@ use common_telemetry::{error, info};
use object_store::ObjectStore;
use snafu::{ensure, OptionExt};
use store_api::metadata::RegionMetadataRef;
use store_api::region_engine::{RegionEngine, RegionRole, SetReadonlyResponse};
use store_api::region_engine::{RegionEngine, RegionHandleResult, RegionRole, SetReadonlyResponse};
use store_api::region_request::{
AffectedRows, RegionCloseRequest, RegionCreateRequest, RegionDropRequest, RegionOpenRequest,
RegionRequest,
@@ -60,7 +60,7 @@ impl RegionEngine for FileRegionEngine {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows, BoxedError> {
) -> Result<RegionHandleResult, BoxedError> {
self.inner
.handle_request(region_id, request)
.await
@@ -154,8 +154,8 @@ impl EngineInner {
&self,
region_id: RegionId,
request: RegionRequest,
) -> EngineResult<AffectedRows> {
match request {
) -> EngineResult<RegionHandleResult> {
let result = match request {
RegionRequest::Create(req) => self.handle_create(region_id, req).await,
RegionRequest::Drop(req) => self.handle_drop(region_id, req).await,
RegionRequest::Open(req) => self.handle_open(region_id, req).await,
@@ -164,7 +164,8 @@ impl EngineInner {
operation: request.to_string(),
}
.fail(),
}
};
result.map(RegionHandleResult::new)
}
async fn stop(&self) -> EngineResult<()> {

8
src/frontend/src/instance/standalone.rs
View File

@@ -18,7 +18,7 @@ use api::v1::region::{QueryRequest, RegionRequest, RegionResponse};
use async_trait::async_trait;
use client::region::check_response_header;
use common_error::ext::BoxedError;
use common_meta::datanode_manager::{AffectedRows, Datanode, DatanodeManager, DatanodeRef};
use common_meta::datanode_manager::{Datanode, DatanodeManager, DatanodeRef, HandleResponse};
use common_meta::error::{self as meta_error, Result as MetaResult};
use common_meta::peer::Peer;
use common_recordbatch::SendableRecordBatchStream;
@@ -63,7 +63,7 @@ impl RegionInvoker {
#[async_trait]
impl Datanode for RegionInvoker {
async fn handle(&self, request: RegionRequest) -> MetaResult<AffectedRows> {
async fn handle(&self, request: RegionRequest) -> MetaResult<HandleResponse> {
let span = request
.header
.as_ref()
@@ -76,10 +76,10 @@ impl Datanode for RegionInvoker {
.await
.map_err(BoxedError::new)
.context(meta_error::ExternalSnafu)?;
check_response_header(response.header)
check_response_header(&response.header)
.map_err(BoxedError::new)
.context(meta_error::ExternalSnafu)?;
Ok(response.affected_rows as _)
Ok(HandleResponse::from_region_response(response))
}
async fn handle_query(&self, request: QueryRequest) -> MetaResult<SendableRecordBatchStream> {

2
src/index/src/inverted_index/error.rs
View File

@@ -113,7 +113,7 @@ pub enum Error {
#[snafu(display("Failed to parse regex DFA"))]
ParseDFA {
#[snafu(source)]
error: Box<regex_automata::dfa::Error>,
error: Box<regex_automata::dfa::dense::BuildError>,
location: Location,
},

70
src/index/src/inverted_index/search/fst_apply/intersection_apply.rs
View File

@@ -17,6 +17,10 @@ use std::mem::size_of;
use fst::map::OpBuilder;
use fst::{IntoStreamer, Streamer};
use regex_automata::dfa::dense::DFA;
use regex_automata::dfa::Automaton;
use regex_automata::util::primitives::StateID;
use regex_automata::util::start::Config;
use regex_automata::Anchored;
use snafu::{ensure, ResultExt};
use crate::inverted_index::error::{
@@ -32,7 +36,53 @@ pub struct IntersectionFstApplier {
ranges: Vec<Range>,
/// A list of `Dfa` compiled from regular expression patterns.
dfas: Vec<DFA<Vec<u32>>>,
dfas: Vec<DfaFstAutomaton>,
}
#[derive(Debug)]
struct DfaFstAutomaton(DFA<Vec<u32>>);
impl fst::Automaton for DfaFstAutomaton {
type State = StateID;
#[inline]
fn start(&self) -> Self::State {
let config = Config::new().anchored(Anchored::No);
self.0.start_state(&config).unwrap()
}
#[inline]
fn is_match(&self, state: &Self::State) -> bool {
self.0.is_match_state(*state)
}
#[inline]
fn can_match(&self, state: &Self::State) -> bool {
!self.0.is_dead_state(*state)
}
#[inline]
fn accept_eof(&self, state: &StateID) -> Option<StateID> {
if self.0.is_match_state(*state) {
return Some(*state);
}
Some(self.0.next_eoi_state(*state))
}
#[inline]
fn accept(&self, state: &Self::State, byte: u8) -> Self::State {
if self.0.is_match_state(*state) {
return *state;
}
self.0.next_state(*state, byte)
}
}
impl IntersectionFstApplier {
fn new(ranges: Vec<Range>, dfas: Vec<DFA<Vec<u32>>>) -> Self {
let dfas = dfas.into_iter().map(DfaFstAutomaton).collect();
Self { ranges, dfas }
}
}
impl FstApplier for IntersectionFstApplier {
@@ -86,7 +136,7 @@ impl FstApplier for IntersectionFstApplier {
size += self.dfas.capacity() * size_of::<DFA<Vec<u32>>>();
for dfa in &self.dfas {
size += dfa.memory_usage();
size += dfa.0.memory_usage();
}
size
}
@@ -119,7 +169,7 @@ impl IntersectionFstApplier {
}
}
Ok(Self { dfas, ranges })
Ok(Self::new(ranges, dfas))
}
}
@@ -365,18 +415,15 @@ mod tests {
#[test]
fn test_intersection_fst_applier_memory_usage() {
let applier = IntersectionFstApplier {
ranges: vec![],
dfas: vec![],
};
let applier = IntersectionFstApplier::new(vec![], vec![]);
assert_eq!(applier.memory_usage(), 0);
let dfa = DFA::new("^abc$").unwrap();
assert_eq!(dfa.memory_usage(), 320);
let applier = IntersectionFstApplier {
ranges: vec![Range {
let applier = IntersectionFstApplier::new(
vec![Range {
lower: Some(Bound {
value: b"aa".to_vec(),
inclusive: true,
@@ -386,9 +433,8 @@ mod tests {
inclusive: true,
}),
}],
dfas: vec![dfa],
};
vec![dfa],
);
assert_eq!(
applier.memory_usage(),
size_of::<Range>() + 4 + size_of::<DFA<Vec<u32>>>() + 320

2
src/meta-srv/src/procedure/tests.rs
View File

@@ -252,7 +252,7 @@ async fn test_on_datanode_create_logical_regions() {
let region_routes = test_data::new_region_routes();
let datanode_manager = new_datanode_manager(&region_server, &region_routes).await;
let physical_table_route = TableRouteValue::physical(region_routes);
let physical_table_id = 111;
let physical_table_id = 1;
let task1 = create_table_task(Some("my_table1"));
let task2 = create_table_task(Some("my_table2"));

1
src/meta-srv/src/procedure/utils.rs
View File

@@ -93,6 +93,7 @@ pub mod mock {
}),
}),
affected_rows: 0,
extension: Default::default(),
})
}
}

9
src/meta-srv/src/service/admin/meta.rs
View File

@@ -52,8 +52,7 @@ impl HttpHandler for CatalogsHandler {
let stream = self
.table_metadata_manager
.catalog_manager()
.catalog_names()
.await;
.catalog_names();
let keys = stream
.try_collect::<Vec<_>>()
@@ -84,8 +83,7 @@ impl HttpHandler for SchemasHandler {
let stream = self
.table_metadata_manager
.schema_manager()
.schema_names(catalog)
.await;
.schema_names(catalog);
let keys = stream
.try_collect::<Vec<_>>()
@@ -118,8 +116,7 @@ impl HttpHandler for TablesHandler {
let stream = self
.table_metadata_manager
.table_name_manager()
.tables(catalog, schema)
.await;
.tables(catalog, schema);
let tables = stream
.try_collect::<Vec<_>>()
.await

36
src/metric-engine/src/data_region.rs
View File

@@ -58,18 +58,19 @@ impl DataRegion {
/// Invoker don't need to set up or verify the column id. This method will adjust
/// it using underlying schema.
///
/// This method will also set the nullable marker to true.
/// This method will also set the nullable marker to true. All of those change are applies
/// to `columns` in-place.
pub async fn add_columns(
&self,
region_id: RegionId,
columns: Vec<ColumnMetadata>,
columns: &mut [ColumnMetadata],
) -> Result<()> {
let region_id = utils::to_data_region_id(region_id);
let mut retries = 0;
// submit alter request
while retries < MAX_RETRIES {
let request = self.assemble_alter_request(region_id, &columns).await?;
let request = self.assemble_alter_request(region_id, columns).await?;
let _timer = MITO_DDL_DURATION.start_timer();
@@ -90,10 +91,12 @@ impl DataRegion {
Ok(())
}
/// Generate warpped [RegionAlterRequest] with given [ColumnMetadata].
/// This method will modify `columns` in-place.
async fn assemble_alter_request(
&self,
region_id: RegionId,
columns: &[ColumnMetadata],
columns: &mut [ColumnMetadata],
) -> Result<RegionRequest> {
// retrieve underlying version
let region_metadata = self
@@ -118,15 +121,14 @@ impl DataRegion {
.unwrap_or(0);
// overwrite semantic type
let columns = columns
.iter()
let new_columns = columns
.iter_mut()
.enumerate()
.map(|(delta, c)| {
let mut c = c.clone();
if c.semantic_type == SemanticType::Tag {
if !c.column_schema.data_type.is_string() {
return ColumnTypeMismatchSnafu {
column_type: c.column_schema.data_type,
column_type: c.column_schema.data_type.clone(),
}
.fail();
}
@@ -138,11 +140,10 @@ impl DataRegion {
};
c.column_id = new_column_id_start + delta as u32;
c.column_schema = c.column_schema.with_nullable_set();
c.column_schema.set_nullable();
Ok(AddColumn {
column_metadata: c,
column_metadata: c.clone(),
location: None,
})
})
@@ -151,7 +152,9 @@ impl DataRegion {
// assemble alter request
let alter_request = RegionRequest::Alter(RegionAlterRequest {
schema_version: version,
kind: AlterKind::AddColumns { columns },
kind: AlterKind::AddColumns {
columns: new_columns,
},
});
Ok(alter_request)
@@ -167,6 +170,7 @@ impl DataRegion {
.handle_request(region_id, RegionRequest::Put(request))
.await
.context(MitoWriteOperationSnafu)
.map(|result| result.affected_rows)
}
pub async fn physical_columns(
@@ -205,7 +209,7 @@ mod test {
// TestEnv will create a logical region which changes the version to 1.
assert_eq!(current_version, 1);
let new_columns = vec![
let mut new_columns = vec![
ColumnMetadata {
column_id: 0,
semantic_type: SemanticType::Tag,
@@ -226,7 +230,7 @@ mod test {
},
];
env.data_region()
.add_columns(env.default_physical_region_id(), new_columns)
.add_columns(env.default_physical_region_id(), &mut new_columns)
.await
.unwrap();
@@ -258,14 +262,14 @@ mod test {
let env = TestEnv::new().await;
env.init_metric_region().await;
let new_columns = vec![ColumnMetadata {
let mut new_columns = vec![ColumnMetadata {
column_id: 0,
semantic_type: SemanticType::Tag,
column_schema: ColumnSchema::new("tag2", ConcreteDataType::int64_datatype(), false),
}];
let result = env
.data_region()
.add_columns(env.default_physical_region_id(), new_columns)
.add_columns(env.default_physical_region_id(), &mut new_columns)
.await;
assert!(result.is_err());
}

38
src/metric-engine/src/engine.rs
View File

@@ -24,6 +24,7 @@ mod region_metadata;
mod state;
use std::any::Any;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use async_trait::async_trait;
@@ -33,13 +34,13 @@ use common_recordbatch::SendableRecordBatchStream;
use mito2::engine::MitoEngine;
use store_api::metadata::RegionMetadataRef;
use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
use store_api::region_engine::{RegionEngine, RegionRole, SetReadonlyResponse};
use store_api::region_request::{AffectedRows, RegionRequest};
use store_api::region_engine::{RegionEngine, RegionHandleResult, RegionRole, SetReadonlyResponse};
use store_api::region_request::RegionRequest;
use store_api::storage::{RegionId, ScanRequest};
use self::state::MetricEngineState;
use crate::data_region::DataRegion;
use crate::error::Result;
use crate::error::{Result, UnsupportedRegionRequestSnafu};
use crate::metadata_region::MetadataRegion;
use crate::utils;
@@ -121,23 +122,38 @@ impl RegionEngine for MetricEngine {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows, BoxedError> {
) -> Result<RegionHandleResult, BoxedError> {
let mut extension_return_value = HashMap::new();
let result = match request {
RegionRequest::Put(put) => self.inner.put_region(region_id, put).await,
RegionRequest::Delete(_) => todo!(),
RegionRequest::Create(create) => self.inner.create_region(region_id, create).await,
RegionRequest::Create(create) => {
self.inner
.create_region(region_id, create, &mut extension_return_value)
.await
}
RegionRequest::Drop(drop) => self.inner.drop_region(region_id, drop).await,
RegionRequest::Open(open) => self.inner.open_region(region_id, open).await,
RegionRequest::Close(close) => self.inner.close_region(region_id, close).await,
RegionRequest::Alter(alter) => self.inner.alter_region(region_id, alter).await,
RegionRequest::Flush(_) => todo!(),
RegionRequest::Compact(_) => todo!(),
RegionRequest::Truncate(_) => todo!(),
RegionRequest::Alter(alter) => {
self.inner
.alter_region(region_id, alter, &mut extension_return_value)
.await
}
RegionRequest::Delete(_)
| RegionRequest::Flush(_)
| RegionRequest::Compact(_)
| RegionRequest::Truncate(_) => UnsupportedRegionRequestSnafu { request }.fail(),
// It always Ok(0), all data is the latest.
RegionRequest::Catchup(_) => Ok(0),
};
result.map_err(BoxedError::new)
result
.map_err(BoxedError::new)
.map(|rows| RegionHandleResult {
affected_rows: rows,
extension: extension_return_value,
})
}
/// Handles substrait query and return a stream of record batches

34
src/metric-engine/src/engine/alter.rs
View File

@@ -12,13 +12,19 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use common_telemetry::{error, info};
use snafu::OptionExt;
use snafu::{OptionExt, ResultExt};
use store_api::metadata::ColumnMetadata;
use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
use store_api::region_request::{AffectedRows, AlterKind, RegionAlterRequest};
use store_api::storage::RegionId;
use crate::engine::MetricEngineInner;
use crate::error::{ForbiddenPhysicalAlterSnafu, LogicalRegionNotFoundSnafu, Result};
use crate::error::{
ForbiddenPhysicalAlterSnafu, LogicalRegionNotFoundSnafu, Result, SerializeColumnMetadataSnafu,
};
use crate::metrics::FORBIDDEN_OPERATION_COUNT;
use crate::utils::{to_data_region_id, to_metadata_region_id};
@@ -28,23 +34,39 @@ impl MetricEngineInner {
&self,
region_id: RegionId,
request: RegionAlterRequest,
extension_return_value: &mut HashMap<String, Vec<u8>>,
) -> Result<AffectedRows> {
let is_altering_physical_region = self.is_physical_region(region_id);
let result = if is_altering_physical_region {
self.alter_physical_region(region_id, request).await
} else {
self.alter_logical_region(region_id, request).await
let physical_region_id = self.alter_logical_region(region_id, request).await?;
// Add physical table's column to extension map.
// It's ok to overwrite existing key, as the latter come schema is more up-to-date
let physical_columns = self
.data_region
.physical_columns(physical_region_id)
.await?;
extension_return_value.insert(
ALTER_PHYSICAL_EXTENSION_KEY.to_string(),
ColumnMetadata::encode_list(&physical_columns)
.context(SerializeColumnMetadataSnafu)?,
);
Ok(())
};
result.map(|_| 0)
}
/// Return the physical region id behind this logical region
async fn alter_logical_region(
&self,
region_id: RegionId,
request: RegionAlterRequest,
) -> Result<()> {
) -> Result<RegionId> {
let physical_region_id = {
let state = &self.state.read().unwrap();
state.get_physical_region_id(region_id).with_context(|| {
@@ -55,7 +77,7 @@ impl MetricEngineInner {
// only handle adding column
let AlterKind::AddColumns { columns } = request.kind else {
return Ok(());
return Ok(physical_region_id);
};
let metadata_region_id = to_metadata_region_id(physical_region_id);
@@ -92,7 +114,7 @@ impl MetricEngineInner {
.await?;
}
Ok(())
Ok(physical_region_id)
}
async fn alter_physical_region(

81
src/metric-engine/src/engine/create.rs
View File

@@ -15,6 +15,7 @@
use std::collections::{HashMap, HashSet};
use api::v1::SemanticType;
use common_error::ext::BoxedError;
use common_telemetry::info;
use common_time::Timestamp;
use datatypes::data_type::ConcreteDataType;
@@ -25,11 +26,12 @@ use object_store::util::join_dir;
use snafu::{ensure, OptionExt, ResultExt};
use store_api::metadata::ColumnMetadata;
use store_api::metric_engine_consts::{
DATA_REGION_SUBDIR, DATA_SCHEMA_TABLE_ID_COLUMN_NAME, DATA_SCHEMA_TSID_COLUMN_NAME,
LOGICAL_TABLE_METADATA_KEY, METADATA_REGION_SUBDIR, METADATA_SCHEMA_KEY_COLUMN_INDEX,
METADATA_SCHEMA_KEY_COLUMN_NAME, METADATA_SCHEMA_TIMESTAMP_COLUMN_INDEX,
METADATA_SCHEMA_TIMESTAMP_COLUMN_NAME, METADATA_SCHEMA_VALUE_COLUMN_INDEX,
METADATA_SCHEMA_VALUE_COLUMN_NAME, PHYSICAL_TABLE_METADATA_KEY,
ALTER_PHYSICAL_EXTENSION_KEY, DATA_REGION_SUBDIR, DATA_SCHEMA_TABLE_ID_COLUMN_NAME,
DATA_SCHEMA_TSID_COLUMN_NAME, LOGICAL_TABLE_METADATA_KEY, METADATA_REGION_SUBDIR,
METADATA_SCHEMA_KEY_COLUMN_INDEX, METADATA_SCHEMA_KEY_COLUMN_NAME,
METADATA_SCHEMA_TIMESTAMP_COLUMN_INDEX, METADATA_SCHEMA_TIMESTAMP_COLUMN_NAME,
METADATA_SCHEMA_VALUE_COLUMN_INDEX, METADATA_SCHEMA_VALUE_COLUMN_NAME,
PHYSICAL_TABLE_METADATA_KEY,
};
use store_api::region_engine::RegionEngine;
use store_api::region_request::{AffectedRows, RegionCreateRequest, RegionRequest};
@@ -41,8 +43,9 @@ use crate::engine::options::{
};
use crate::engine::MetricEngineInner;
use crate::error::{
ConflictRegionOptionSnafu, CreateMitoRegionSnafu, InternalColumnOccupiedSnafu,
MissingRegionOptionSnafu, ParseRegionIdSnafu, PhysicalRegionNotFoundSnafu, Result,
ColumnNotFoundSnafu, ConflictRegionOptionSnafu, CreateMitoRegionSnafu,
InternalColumnOccupiedSnafu, MissingRegionOptionSnafu, MitoReadOperationSnafu,
ParseRegionIdSnafu, PhysicalRegionNotFoundSnafu, Result, SerializeColumnMetadataSnafu,
};
use crate::metrics::{LOGICAL_REGION_COUNT, PHYSICAL_COLUMN_COUNT, PHYSICAL_REGION_COUNT};
use crate::utils::{to_data_region_id, to_metadata_region_id};
@@ -53,13 +56,28 @@ impl MetricEngineInner {
&self,
region_id: RegionId,
request: RegionCreateRequest,
extension_return_value: &mut HashMap<String, Vec<u8>>,
) -> Result<AffectedRows> {
Self::verify_region_create_request(&request)?;
let result = if request.options.contains_key(PHYSICAL_TABLE_METADATA_KEY) {
self.create_physical_region(region_id, request).await
} else if request.options.contains_key(LOGICAL_TABLE_METADATA_KEY) {
self.create_logical_region(region_id, request).await
let physical_region_id = self.create_logical_region(region_id, request).await?;
// Add physical table's column to extension map.
// It's ok to overwrite existing key, as the latter come schema is more up-to-date
let physical_columns = self
.data_region
.physical_columns(physical_region_id)
.await?;
extension_return_value.insert(
ALTER_PHYSICAL_EXTENSION_KEY.to_string(),
ColumnMetadata::encode_list(&physical_columns)
.context(SerializeColumnMetadataSnafu)?,
);
Ok(())
} else {
MissingRegionOptionSnafu {}.fail()
};
@@ -126,11 +144,16 @@ impl MetricEngineInner {
/// This method will alter the data region to add columns if necessary.
///
/// If the logical region to create already exists, this method will do nothing.
///
/// `alter_request` is a hashmap that stores the alter requests that were executed
/// to the physical region.
///
/// Return the physical region id of this logical region
async fn create_logical_region(
&self,
logical_region_id: RegionId,
request: RegionCreateRequest,
) -> Result<()> {
) -> Result<RegionId> {
// transform IDs
let physical_region_id_raw = request
.options
@@ -151,11 +174,12 @@ impl MetricEngineInner {
.await?
{
info!("Create a existing logical region {logical_region_id}. Skipped");
return Ok(());
return Ok(data_region_id);
}
// find new columns to add
let mut new_columns = vec![];
let mut existing_columns = vec![];
{
let state = &self.state.read().unwrap();
let physical_columns =
@@ -168,6 +192,8 @@ impl MetricEngineInner {
for col in &request.column_metadatas {
if !physical_columns.contains(&col.column_schema.name) {
new_columns.push(col.clone());
} else {
existing_columns.push(col.column_schema.name.clone());
}
}
}
@@ -188,9 +214,28 @@ impl MetricEngineInner {
self.metadata_region
.add_logical_region(metadata_region_id, logical_region_id)
.await?;
for col in &request.column_metadatas {
// register existing physical column to this new logical region.
let physical_schema = self
.data_region
.physical_columns(data_region_id)
.await
.map_err(BoxedError::new)
.context(MitoReadOperationSnafu)?;
let physical_schema_map = physical_schema
.into_iter()
.map(|metadata| (metadata.column_schema.name.clone(), metadata))
.collect::<HashMap<_, _>>();
for col in &existing_columns {
let column_metadata = physical_schema_map
.get(col)
.with_context(|| ColumnNotFoundSnafu {
name: col,
region_id: physical_region_id,
})?
.clone();
self.metadata_region
.add_column(metadata_region_id, logical_region_id, col)
.add_column(metadata_region_id, logical_region_id, &column_metadata)
.await?;
}
@@ -203,19 +248,21 @@ impl MetricEngineInner {
info!("Created new logical region {logical_region_id} on physical region {data_region_id}");
LOGICAL_REGION_COUNT.inc();
Ok(())
Ok(data_region_id)
}
/// Execute corresponding alter requests to mito region. New added columns' [ColumnMetadata] will be
/// cloned into `added_columns`.
pub(crate) async fn add_columns_to_physical_data_region(
&self,
data_region_id: RegionId,
metadata_region_id: RegionId,
logical_region_id: RegionId,
new_columns: Vec<ColumnMetadata>,
mut new_columns: Vec<ColumnMetadata>,
) -> Result<()> {
// alter data region
self.data_region
.add_columns(data_region_id, new_columns.clone())
.add_columns(data_region_id, &mut new_columns)
.await?;
// register columns to metadata region
@@ -362,13 +409,13 @@ impl MetricEngineInner {
// concat region dir
data_region_request.region_dir = join_dir(&request.region_dir, DATA_REGION_SUBDIR);
// convert semantic type
// change nullability for tag columns
data_region_request
.column_metadatas
.iter_mut()
.for_each(|metadata| {
if metadata.semantic_type == SemanticType::Tag {
metadata.semantic_type = SemanticType::Field;
metadata.column_schema.set_nullable();
}
});

8
src/metric-engine/src/engine/put.rs
View File

@@ -215,12 +215,12 @@ mod tests {
// write data
let logical_region_id = env.default_logical_region_id();
let count = env
let result = env
.metric()
.handle_request(logical_region_id, request)
.await
.unwrap();
assert_eq!(count, 5);
assert_eq!(result.affected_rows, 5);
// read data from physical region
let physical_region_id = env.default_physical_region_id();
@@ -287,11 +287,11 @@ mod tests {
});
// write data
let count = engine
let result = engine
.handle_request(logical_region_id, request)
.await
.unwrap();
assert_eq!(100, count);
assert_eq!(100, result.affected_rows);
}
#[tokio::test]

6
src/metric-engine/src/engine/read.rs
View File

@@ -143,6 +143,7 @@ impl MetricEngineInner {
self.default_projection(physical_region_id, logical_region_id)
.await?
};
request.projection = Some(physical_projection);
// add table filter
@@ -186,6 +187,7 @@ impl MetricEngineInner {
.get_metadata(data_region_id)
.await
.context(MitoReadOperationSnafu)?;
for name in projected_logical_names {
// Safety: logical columns is a strict subset of physical columns
physical_projection.push(physical_metadata.column_index_by_name(&name).unwrap());
@@ -301,7 +303,7 @@ mod test {
.await
.unwrap();
assert_eq!(scan_req.projection.unwrap(), vec![0, 1, 4, 8, 9, 10, 11]);
assert_eq!(scan_req.projection.unwrap(), vec![11, 10, 9, 8, 0, 1, 4]);
assert_eq!(scan_req.filters.len(), 1);
assert_eq!(
scan_req.filters[0],
@@ -318,6 +320,6 @@ mod test {
.transform_request(physical_region_id, logical_region_id, scan_req)
.await
.unwrap();
assert_eq!(scan_req.projection.unwrap(), vec![0, 1, 4, 8, 9, 10, 11]);
assert_eq!(scan_req.projection.unwrap(), vec![11, 10, 9, 8, 0, 1, 4]);
}
}

3
src/metric-engine/src/engine/region_metadata.rs
View File

@@ -39,7 +39,8 @@ impl MetricEngineInner {
.collect::<Vec<_>>();
// sort columns on column id to ensure the order
logical_column_metadata.sort_unstable_by_key(|col| col.column_id);
logical_column_metadata
.sort_unstable_by(|c1, c2| c1.column_schema.name.cmp(&c2.column_schema.name));
Ok(logical_column_metadata)
}

19
src/metric-engine/src/error.rs
View File

@@ -19,6 +19,7 @@ use common_error::status_code::StatusCode;
use common_macro::stack_trace_debug;
use datatypes::prelude::ConcreteDataType;
use snafu::{Location, Snafu};
use store_api::region_request::RegionRequest;
use store_api::storage::RegionId;
#[derive(Snafu)]
@@ -71,6 +72,13 @@ pub enum Error {
location: Location,
},
#[snafu(display("Failed to serialize column metadata"))]
SerializeColumnMetadata {
#[snafu(source)]
error: serde_json::Error,
location: Location,
},
#[snafu(display("Failed to decode base64 column value"))]
DecodeColumnValue {
#[snafu(source)]
@@ -155,6 +163,12 @@ pub enum Error {
region_id: RegionId,
location: Location,
},
#[snafu(display("Unsupported region request: {}", request))]
UnsupportedRegionRequest {
request: RegionRequest,
location: Location,
},
}
pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -170,11 +184,14 @@ impl ErrorExt for Error {
| ColumnTypeMismatch { .. }
| PhysicalRegionBusy { .. } => StatusCode::InvalidArguments,
ForbiddenPhysicalAlter { .. } => StatusCode::Unsupported,
ForbiddenPhysicalAlter { .. } | UnsupportedRegionRequest { .. } => {
StatusCode::Unsupported
}
MissingInternalColumn { .. }
| DeserializeSemanticType { .. }
| DeserializeColumnMetadata { .. }
| SerializeColumnMetadata { .. }
| DecodeColumnValue { .. }
| ParseRegionId { .. }
| InvalidMetadata { .. } => StatusCode::Unexpected,

2
src/metric-engine/src/metadata_region.rs
View File

@@ -167,7 +167,7 @@ impl MetadataRegion {
// TODO(ruihang): avoid using `get_all`
/// Get all the columns of a given logical region.
/// Return a list of (column_name, semantic_type).
/// Return a list of (column_name, column_metadata).
pub async fn logical_columns(
&self,
physical_region_id: RegionId,

5
src/mito2/src/engine.rs
View File

@@ -57,7 +57,7 @@ use object_store::manager::ObjectStoreManagerRef;
use snafu::{ensure, OptionExt, ResultExt};
use store_api::logstore::LogStore;
use store_api::metadata::RegionMetadataRef;
use store_api::region_engine::{RegionEngine, RegionRole, SetReadonlyResponse};
use store_api::region_engine::{RegionEngine, RegionHandleResult, RegionRole, SetReadonlyResponse};
use store_api::region_request::{AffectedRows, RegionRequest};
use store_api::storage::{RegionId, ScanRequest};
use tokio::sync::oneshot;
@@ -290,10 +290,11 @@ impl RegionEngine for MitoEngine {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows, BoxedError> {
) -> Result<RegionHandleResult, BoxedError> {
self.inner
.handle_request(region_id, request)
.await
.map(RegionHandleResult::new)
.map_err(BoxedError::new)
}

4
src/mito2/src/engine/basic_test.rs
View File

@@ -111,7 +111,7 @@ async fn test_region_replay() {
let engine = env.reopen_engine(engine, MitoConfig::default()).await;
let rows = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Open(RegionOpenRequest {
@@ -123,7 +123,7 @@ async fn test_region_replay() {
)
.await
.unwrap();
assert_eq!(0, rows);
assert_eq!(0, result.affected_rows);
let request = ScanRequest::default();
let stream = engine.handle_query(region_id, request).await.unwrap();

16
src/mito2/src/engine/compaction_test.rs
View File

@@ -42,7 +42,7 @@ async fn put_and_flush(
};
put_rows(engine, region_id, rows).await;
let rows = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Flush(RegionFlushRequest {
@@ -51,7 +51,7 @@ async fn put_and_flush(
)
.await
.unwrap();
assert_eq!(0, rows);
assert_eq!(0, result.affected_rows);
}
async fn delete_and_flush(
@@ -66,16 +66,16 @@ async fn delete_and_flush(
rows: build_rows_for_key("a", rows.start, rows.end, 0),
};
let rows_affected = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Delete(RegionDeleteRequest { rows }),
)
.await
.unwrap();
assert_eq!(row_cnt, rows_affected);
assert_eq!(row_cnt, result.affected_rows);
let rows = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Flush(RegionFlushRequest {
@@ -84,7 +84,7 @@ async fn delete_and_flush(
)
.await
.unwrap();
assert_eq!(0, rows);
assert_eq!(0, result.affected_rows);
}
async fn collect_stream_ts(stream: SendableRecordBatchStream) -> Vec<i64> {
@@ -127,11 +127,11 @@ async fn test_compaction_region() {
delete_and_flush(&engine, region_id, &column_schemas, 15..30).await;
put_and_flush(&engine, region_id, &column_schemas, 15..25).await;
let output = engine
let result = engine
.handle_request(region_id, RegionRequest::Compact(RegionCompactRequest {}))
.await
.unwrap();
assert_eq!(output, 0);
assert_eq!(result.affected_rows, 0);
let scanner = engine.scanner(region_id, ScanRequest::default()).unwrap();
assert_eq!(

12
src/mito2/src/test_util.rs
View File

@@ -712,11 +712,11 @@ pub fn delete_rows_schema(request: &RegionCreateRequest) -> Vec<api::v1::ColumnS
/// Put rows into the engine.
pub async fn put_rows(engine: &MitoEngine, region_id: RegionId, rows: Rows) {
let num_rows = rows.rows.len();
let rows_inserted = engine
let result = engine
.handle_request(region_id, RegionRequest::Put(RegionPutRequest { rows }))
.await
.unwrap();
assert_eq!(num_rows, rows_inserted);
assert_eq!(num_rows, result.affected_rows);
}
/// Build rows to put for specific `key`.
@@ -758,26 +758,26 @@ pub fn build_delete_rows_for_key(key: &str, start: usize, end: usize) -> Vec<Row
/// Delete rows from the engine.
pub async fn delete_rows(engine: &MitoEngine, region_id: RegionId, rows: Rows) {
let num_rows = rows.rows.len();
let rows_inserted = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Delete(RegionDeleteRequest { rows }),
)
.await
.unwrap();
assert_eq!(num_rows, rows_inserted);
assert_eq!(num_rows, result.affected_rows);
}
/// Flush a region manually.
pub async fn flush_region(engine: &MitoEngine, region_id: RegionId, row_group_size: Option<usize>) {
let rows = engine
let result = engine
.handle_request(
region_id,
RegionRequest::Flush(RegionFlushRequest { row_group_size }),
)
.await
.unwrap();
assert_eq!(0, rows);
assert_eq!(0, result.affected_rows);
}
/// Reopen a region.

5
src/operator/src/delete.rs
View File

@@ -144,7 +144,10 @@ impl Deleter {
});
let results = future::try_join_all(tasks).await.context(JoinTaskSnafu)?;
let affected_rows = results.into_iter().sum::<Result<AffectedRows>>()?;
let affected_rows = results
.into_iter()
.map(|resp| resp.map(|r| r.affected_rows))
.sum::<Result<AffectedRows>>()?;
crate::metrics::DIST_DELETE_ROW_COUNT.inc_by(affected_rows as u64);
Ok(affected_rows)
}

5
src/operator/src/insert.rs
View File

@@ -216,7 +216,10 @@ impl Inserter {
});
let results = future::try_join_all(tasks).await.context(JoinTaskSnafu)?;
let affected_rows = results.into_iter().sum::<Result<AffectedRows>>()?;
let affected_rows = results
.into_iter()
.map(|resp| resp.map(|r| r.affected_rows))
.sum::<Result<AffectedRows>>()?;
crate::metrics::DIST_INGEST_ROW_COUNT.inc_by(affected_rows as u64);
Ok(Output::new(
OutputData::AffectedRows(affected_rows),

5
src/operator/src/request.rs
View File

@@ -181,7 +181,10 @@ impl Requester {
});
let results = future::try_join_all(tasks).await.context(JoinTaskSnafu)?;
let affected_rows = results.into_iter().sum::<Result<AffectedRows>>()?;
let affected_rows = results
.into_iter()
.map(|resp| resp.map(|r| r.affected_rows))
.sum::<Result<AffectedRows>>()?;
Ok(affected_rows)
}

13
src/operator/src/statement.rs
View File

@@ -168,12 +168,13 @@ impl StatementExecutor {
let table_name = TableName::new(catalog, schema, table);
self.drop_table(table_name, stmt.drop_if_exists()).await
}
Statement::DropDatabase(_stmt) => {
// TODO(weny): implement the drop database procedure
error::NotSupportedSnafu {
feat: "Drop Database",
}
.fail()
Statement::DropDatabase(stmt) => {
self.drop_database(
query_ctx.current_catalog().to_string(),
format_raw_object_name(stmt.name()),
stmt.drop_if_exists(),
)
.await
}
Statement::TruncateTable(stmt) => {
let (catalog, schema, table) =

61
src/operator/src/statement/ddl.rs
View File

@@ -48,6 +48,7 @@ use sql::statements::alter::AlterTable;
use sql::statements::create::{CreateExternalTable, CreateTable, CreateTableLike, Partitions};
use sql::statements::sql_value_to_value;
use sqlparser::ast::{Expr, Ident, Value as ParserValue};
use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
use table::dist_table::DistTable;
use table::metadata::{self, RawTableInfo, RawTableMeta, TableId, TableInfo, TableType};
use table::requests::{AlterKind, AlterTableRequest, TableOptions};
@@ -138,6 +139,22 @@ impl StatementExecutor {
partitions: Option<Partitions>,
query_ctx: &QueryContextRef,
) -> Result<TableRef> {
// Check if is creating logical table
if create_table.engine == METRIC_ENGINE_NAME
&& create_table
.table_options
.contains_key(LOGICAL_TABLE_METADATA_KEY)
{
return self
.create_logical_tables(&[create_table.clone()])
.await?
.into_iter()
.next()
.context(error::UnexpectedSnafu {
violated: "expected to create a logical table",
});
}
let _timer = crate::metrics::DIST_CREATE_TABLE.start_timer();
let schema = self
.table_metadata_manager
@@ -352,6 +369,34 @@ impl StatementExecutor {
}
}
#[tracing::instrument(skip_all)]
pub async fn drop_database(
&self,
catalog: String,
schema: String,
drop_if_exists: bool,
) -> Result<Output> {
if self
.catalog_manager
.schema_exists(&catalog, &schema)
.await
.context(CatalogSnafu)?
{
self.drop_database_procedure(catalog, schema, drop_if_exists)
.await?;
Ok(Output::new_with_affected_rows(0))
} else if drop_if_exists {
// DROP TABLE IF EXISTS meets table not found - ignored
Ok(Output::new_with_affected_rows(0))
} else {
Err(SchemaNotFoundSnafu {
schema_info: schema,
}
.into_error(snafu::NoneError))
}
}
#[tracing::instrument(skip_all)]
pub async fn truncate_table(&self, table_name: TableName) -> Result<Output> {
let table = self
@@ -528,6 +573,22 @@ impl StatementExecutor {
.context(error::ExecuteDdlSnafu)
}
async fn drop_database_procedure(
&self,
catalog: String,
schema: String,
drop_if_exists: bool,
) -> Result<SubmitDdlTaskResponse> {
let request = SubmitDdlTaskRequest {
task: DdlTask::new_drop_database(catalog, schema, drop_if_exists),
};
self.procedure_executor
.submit_ddl_task(&ExecutorContext::default(), request)
.await
.context(error::ExecuteDdlSnafu)
}
async fn truncate_table_procedure(
&self,
table_name: &TableName,

1
src/partition/src/lib.rs
View File

@@ -18,7 +18,6 @@ pub mod columns;
pub mod error;
pub mod expr;
pub mod manager;
pub mod metrics;
pub mod multi_dim;
pub mod partition;
pub mod range;

3
src/promql/src/functions/extrapolate_rate.rs
View File

@@ -13,6 +13,7 @@
// limitations under the License.
// This file also contains some code from prometheus project.
// Copyright 2015 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -48,7 +49,7 @@ pub type Rate = ExtrapolatedRate<true, true>;
pub type Increase = ExtrapolatedRate<true, false>;
/// Part of the `extrapolatedRate` in Promql,
/// from <https://github.com/prometheus/prometheus/blob/6bdecf377cea8e856509914f35234e948c4fcb80/promql/functions.go#L66>
/// from <https://github.com/prometheus/prometheus/blob/v0.40.1/promql/functions.go#L66>
#[derive(Debug)]
pub struct ExtrapolatedRate<const IS_COUNTER: bool, const IS_RATE: bool> {
/// Range duration in millisecond

2
src/servers/src/http/table_result.rs
View File

@@ -135,7 +135,7 @@ impl IntoResponse for TableResponse {
let mut resp = (
[(
header::CONTENT_TYPE,
HeaderValue::from_static(mime::PLAIN.as_ref()),
HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref()),
)],
self.to_string(),
)

19
src/servers/src/repeated_field.rs
View File

@@ -1,17 +1,3 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Copyright (c) 2019 Stepan Koltsov
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -32,8 +18,9 @@
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
// OR OTHER DEALINGS IN THE SOFTWARE.
/// ! The [Clear] trait and [RepeatedField] are taken from [rust-protobuf](https://github.com/stepancheg/rust-protobuf/tree/master/protobuf-examples/vs-prost)
/// to leverage the pooling mechanism to avoid frequent heap allocation/deallocation when decoding deeply nested structs.
// The [Clear] trait and [RepeatedField] are taken from [rust-protobuf](https://github.com/stepancheg/rust-protobuf/tree/master/protobuf-examples/vs-prost)
// to leverage the pooling mechanism to avoid frequent heap allocation/deallocation when decoding deeply nested structs.
use std::borrow::Borrow;
use std::cmp::Ordering;
use std::default::Default;

40
src/servers/tests/http/http_handler_test.rs
View File

@@ -46,7 +46,7 @@ async fn test_sql_not_provided() {
script_handler: None,
};
for format in ["greptimedb_v1", "influxdb_v1", "csv"] {
for format in ["greptimedb_v1", "influxdb_v1", "csv", "table"] {
let query = http_handler::SqlQuery {
db: None,
sql: None,
@@ -82,7 +82,7 @@ async fn test_sql_output_rows() {
script_handler: None,
};
for format in ["greptimedb_v1", "influxdb_v1", "csv"] {
for format in ["greptimedb_v1", "influxdb_v1", "csv", "table"] {
let query = create_query(format);
let json = http_handler::sql(
State(api_state.clone()),
@@ -154,6 +154,23 @@ async fn test_sql_output_rows() {
hyper::body::Bytes::from_static(b"4950\n"),
);
}
HttpResponse::Table(resp) => {
use http_body::Body as HttpBody;
let mut resp = resp.into_response();
assert_eq!(
resp.headers().get(header::CONTENT_TYPE),
Some(HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref())).as_ref(),
);
assert_eq!(
resp.body_mut().data().await.unwrap().unwrap(),
hyper::body::Bytes::from(
r#"┌─SUM(numbers.uint32s)─┐
│ 4950 │
└──────────────────────┘
"#
),
);
}
_ => unreachable!(),
}
}
@@ -172,7 +189,7 @@ async fn test_sql_form() {
script_handler: None,
};
for format in ["greptimedb_v1", "influxdb_v1", "csv"] {
for format in ["greptimedb_v1", "influxdb_v1", "csv", "table"] {
let form = create_form(format);
let json = http_handler::sql(
State(api_state.clone()),
@@ -244,6 +261,23 @@ async fn test_sql_form() {
hyper::body::Bytes::from_static(b"4950\n"),
);
}
HttpResponse::Table(resp) => {
use http_body::Body as HttpBody;
let mut resp = resp.into_response();
assert_eq!(
resp.headers().get(header::CONTENT_TYPE),
Some(HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref())).as_ref(),
);
assert_eq!(
resp.body_mut().data().await.unwrap().unwrap(),
hyper::body::Bytes::from(
r#"┌─SUM(numbers.uint32s)─┐
│ 4950 │
└──────────────────────┘
"#
),
);
}
_ => unreachable!(),
}
}

10
src/store-api/src/metadata.rs
View File

@@ -95,6 +95,16 @@ impl ColumnMetadata {
column_id,
})
}
/// Encodes a vector of `ColumnMetadata` into a JSON byte vector.
pub fn encode_list(columns: &[Self]) -> serde_json::Result<Vec<u8>> {
serde_json::to_vec(columns)
}
/// Decodes a JSON byte vector into a vector of `ColumnMetadata`.
pub fn decode_list(bytes: &[u8]) -> serde_json::Result<Vec<Self>> {
serde_json::from_slice(bytes)
}
}
#[cfg_attr(doc, aquamarine::aquamarine)]

4
src/store-api/src/metric_engine_consts.rs
View File

@@ -66,3 +66,7 @@ pub const PHYSICAL_TABLE_METADATA_KEY: &str = "physical_metric_table";
/// ```
/// And this key will be translated to corresponding physical **REGION** id in metasrv.
pub const LOGICAL_TABLE_METADATA_KEY: &str = "on_physical_table";
/// HashMap key to be used in the region server's extension response.
/// Represent a list of column metadata that are added to physical table.
pub const ALTER_PHYSICAL_EXTENSION_KEY: &str = "ALTER_PHYSICAL";

20
src/store-api/src/region_engine.rs
View File

@@ -15,6 +15,7 @@
//! Region Engine's definition
use std::any::Any;
use std::collections::HashMap;
use std::fmt::Display;
use std::sync::Arc;
@@ -129,7 +130,7 @@ pub trait RegionEngine: Send + Sync {
&self,
region_id: RegionId,
request: RegionRequest,
) -> Result<AffectedRows, BoxedError>;
) -> Result<RegionHandleResult, BoxedError>;
/// Handles substrait query and return a stream of record batches
async fn handle_query(
@@ -171,3 +172,20 @@ pub trait RegionEngine: Send + Sync {
}
pub type RegionEngineRef = Arc<dyn RegionEngine>;
// TODO: reorganize the dependence to merge this struct with the
// one in common_meta
#[derive(Debug)]
pub struct RegionHandleResult {
pub affected_rows: AffectedRows,
pub extension: HashMap<String, Vec<u8>>,
}
impl RegionHandleResult {
pub fn new(affected_rows: AffectedRows) -> Self {
Self {
affected_rows,
extension: Default::default(),
}
}
}

6
src/table/src/metadata.rs
View File

@@ -567,13 +567,19 @@ impl From<TableId> for TableIdent {
#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct RawTableMeta {
pub schema: RawSchema,
/// The indices of columns in primary key. Note that the index of timestamp column
/// is not included. Order matters to this array.
pub primary_key_indices: Vec<usize>,
/// The indices of columns in value. Order doesn't matter to this array.
pub value_indices: Vec<usize>,
/// Engine type of this table. Usually in small case.
pub engine: String,
/// Deprecated. See https://github.com/GreptimeTeam/greptimedb/issues/2982
pub next_column_id: ColumnId,
pub region_numbers: Vec<u32>,
pub options: TableOptions,
pub created_on: DateTime<Utc>,
/// Order doesn't matter to this array.
#[serde(default)]
pub partition_key_indices: Vec<usize>,
}