Skip to content

Commit fbb57fc

Browse files
b41shb41sh
committed
feat(query): Support generate virtual columns
1 parent 04a119e commit fbb57fc

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

62 files changed

+2340
-40
lines changed

‎Cargo.lock

+21
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/common/exception/src/exception_code.rs

+1
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,7 @@ build_exceptions! {
141141
EmptyShareEndpointConfig(1111),
142142
LicenceDenied(1112),
143143
UnknownDatamask(1113),
144+
VirtualColumnNotExist(1114),
144145

145146
// Data Related Errors
146147

‎src/meta/api/src/schema_api.rs

+23
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ use common_meta_app::schema::CreateTableLockRevReply;
2424
use common_meta_app::schema::CreateTableLockRevReq;
2525
use common_meta_app::schema::CreateTableReply;
2626
use common_meta_app::schema::CreateTableReq;
27+
use common_meta_app::schema::CreateVirtualColumnReply;
28+
use common_meta_app::schema::CreateVirtualColumnReq;
2729
use common_meta_app::schema::DatabaseInfo;
2830
use common_meta_app::schema::DeleteTableLockRevReq;
2931
use common_meta_app::schema::DropDatabaseReply;
@@ -32,6 +34,8 @@ use common_meta_app::schema::DropIndexReply;
3234
use common_meta_app::schema::DropIndexReq;
3335
use common_meta_app::schema::DropTableByIdReq;
3436
use common_meta_app::schema::DropTableReply;
37+
use common_meta_app::schema::DropVirtualColumnReply;
38+
use common_meta_app::schema::DropVirtualColumnReq;
3539
use common_meta_app::schema::ExtendTableLockRevReq;
3640
use common_meta_app::schema::GetDatabaseReq;
3741
use common_meta_app::schema::GetTableCopiedFileReply;
@@ -42,6 +46,7 @@ use common_meta_app::schema::ListDatabaseReq;
4246
use common_meta_app::schema::ListIndexesReq;
4347
use common_meta_app::schema::ListTableLockRevReq;
4448
use common_meta_app::schema::ListTableReq;
49+
use common_meta_app::schema::ListVirtualColumnsReq;
4550
use common_meta_app::schema::RenameDatabaseReply;
4651
use common_meta_app::schema::RenameDatabaseReq;
4752
use common_meta_app::schema::RenameTableReply;
@@ -60,6 +65,7 @@ use common_meta_app::schema::UpdateTableMetaReply;
6065
use common_meta_app::schema::UpdateTableMetaReq;
6166
use common_meta_app::schema::UpsertTableOptionReply;
6267
use common_meta_app::schema::UpsertTableOptionReq;
68+
use common_meta_app::schema::VirtualColumnMeta;
6369
use common_meta_types::GCDroppedDataReply;
6470
use common_meta_types::GCDroppedDataReq;
6571
use common_meta_types::MetaId;
@@ -111,6 +117,23 @@ pub trait SchemaApi: Send + Sync {
111117
req: ListIndexesReq,
112118
) -> Result<Vec<(u64, String, IndexMeta)>, KVAppError>;
113119

120+
// virtual column
121+
122+
async fn create_virtual_column(
123+
&self,
124+
req: CreateVirtualColumnReq,
125+
) -> Result<CreateVirtualColumnReply, KVAppError>;
126+
127+
async fn drop_virtual_column(
128+
&self,
129+
req: DropVirtualColumnReq,
130+
) -> Result<DropVirtualColumnReply, KVAppError>;
131+
132+
async fn list_virtual_columns(
133+
&self,
134+
req: ListVirtualColumnsReq,
135+
) -> Result<Vec<VirtualColumnMeta>, KVAppError>;
136+
114137
// table
115138

116139
async fn create_table(&self, req: CreateTableReq) -> Result<CreateTableReply, KVAppError>;

‎src/meta/api/src/schema_api_impl.rs

+198
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
// limitations under the License.
1414

1515
use std::collections::BTreeMap;
16+
use std::collections::HashSet;
1617
use std::fmt::Display;
1718
use std::sync::Arc;
1819

@@ -41,6 +42,7 @@ use common_meta_app::app_error::UnknownDatabaseId;
4142
use common_meta_app::app_error::UnknownIndex;
4243
use common_meta_app::app_error::UnknownTable;
4344
use common_meta_app::app_error::UnknownTableId;
45+
use common_meta_app::app_error::VirtualColumnNotExist;
4446
use common_meta_app::app_error::WrongShare;
4547
use common_meta_app::app_error::WrongShareObject;
4648
use common_meta_app::schema::CountTablesKey;
@@ -54,6 +56,8 @@ use common_meta_app::schema::CreateTableLockRevReply;
5456
use common_meta_app::schema::CreateTableLockRevReq;
5557
use common_meta_app::schema::CreateTableReply;
5658
use common_meta_app::schema::CreateTableReq;
59+
use common_meta_app::schema::CreateVirtualColumnReply;
60+
use common_meta_app::schema::CreateVirtualColumnReq;
5761
use common_meta_app::schema::DBIdTableName;
5862
use common_meta_app::schema::DatabaseId;
5963
use common_meta_app::schema::DatabaseIdToName;
@@ -71,6 +75,8 @@ use common_meta_app::schema::DropIndexReply;
7175
use common_meta_app::schema::DropIndexReq;
7276
use common_meta_app::schema::DropTableByIdReq;
7377
use common_meta_app::schema::DropTableReply;
78+
use common_meta_app::schema::DropVirtualColumnReply;
79+
use common_meta_app::schema::DropVirtualColumnReq;
7480
use common_meta_app::schema::EmptyProto;
7581
use common_meta_app::schema::ExtendTableLockRevReq;
7682
use common_meta_app::schema::GetDatabaseReq;
@@ -85,6 +91,7 @@ use common_meta_app::schema::ListDatabaseReq;
8591
use common_meta_app::schema::ListIndexesReq;
8692
use common_meta_app::schema::ListTableLockRevReq;
8793
use common_meta_app::schema::ListTableReq;
94+
use common_meta_app::schema::ListVirtualColumnsReq;
8895
use common_meta_app::schema::RenameDatabaseReply;
8996
use common_meta_app::schema::RenameDatabaseReq;
9097
use common_meta_app::schema::RenameTableReply;
@@ -111,6 +118,8 @@ use common_meta_app::schema::UpdateTableMetaReq;
111118
use common_meta_app::schema::UpsertTableCopiedFileReq;
112119
use common_meta_app::schema::UpsertTableOptionReply;
113120
use common_meta_app::schema::UpsertTableOptionReq;
121+
use common_meta_app::schema::VirtualColumnMeta;
122+
use common_meta_app::schema::VirtualColumnNameIdent;
114123
use common_meta_app::share::ShareGrantObject;
115124
use common_meta_app::share::ShareId;
116125
use common_meta_app::share::ShareNameIdent;
@@ -1071,6 +1080,195 @@ impl<KV: kvapi::KVApi<Error = MetaError>> SchemaApi for KV {
10711080
Ok(index_metas)
10721081
}
10731082

1083+
// virtual column
1084+
1085+
async fn create_virtual_column(
1086+
&self,
1087+
req: CreateVirtualColumnReq,
1088+
) -> Result<CreateVirtualColumnReply, KVAppError> {
1089+
debug!(req = debug(&req), "SchemaApi: {}", func_name!());
1090+
1091+
let mut retry = 0;
1092+
while retry < TXN_MAX_RETRY_TIMES {
1093+
retry += 1;
1094+
1095+
let (_, old_virtual_column_opt): (_, Option<VirtualColumnMeta>) =
1096+
get_pb_value(self, &req.name_ident).await?;
1097+
1098+
// If virtual columns already exist, merge them together
1099+
let virtual_column_meta = if let Some(old_virtual_column) = old_virtual_column_opt {
1100+
let mut virtual_columns = req.virtual_columns.clone();
1101+
virtual_columns.extend(old_virtual_column.virtual_columns.clone());
1102+
1103+
let virtual_columns_set: HashSet<String> =
1104+
HashSet::from_iter(virtual_columns.into_iter());
1105+
let virtual_columns = Vec::from_iter(virtual_columns_set.into_iter());
1106+
1107+
VirtualColumnMeta {
1108+
table_id: req.name_ident.table_id,
1109+
virtual_columns,
1110+
created_on: old_virtual_column.created_on.clone(),
1111+
updated_on: Some(Utc::now()),
1112+
drop_on: old_virtual_column.drop_on.clone(),
1113+
}
1114+
} else {
1115+
VirtualColumnMeta {
1116+
table_id: req.name_ident.table_id,
1117+
virtual_columns: req.virtual_columns.clone(),
1118+
created_on: Utc::now(),
1119+
updated_on: None,
1120+
drop_on: None,
1121+
}
1122+
};
1123+
1124+
// Create virtual column by inserting these record:
1125+
// (tenant, table_id) -> virtual_column_meta
1126+
{
1127+
let condition = vec![];
1128+
let if_then = vec![
1129+
txn_op_put(&req.name_ident, serialize_struct(&virtual_column_meta)?), /* (tenant, table_id) -> virtual_column_meta */
1130+
];
1131+
1132+
let txn_req = TxnRequest {
1133+
condition,
1134+
if_then,
1135+
else_then: vec![],
1136+
};
1137+
1138+
let (succ, _responses) = send_txn(self, txn_req).await?;
1139+
1140+
debug!(
1141+
req.name_ident = debug(&virtual_column_meta),
1142+
succ = display(succ),
1143+
"create_virtual_column"
1144+
);
1145+
1146+
if succ {
1147+
return Ok(CreateVirtualColumnReply {});
1148+
}
1149+
}
1150+
}
1151+
1152+
Err(KVAppError::AppError(AppError::TxnRetryMaxTimes(
1153+
TxnRetryMaxTimes::new("create_virtual_column", TXN_MAX_RETRY_TIMES),
1154+
)))
1155+
}
1156+
1157+
async fn drop_virtual_column(
1158+
&self,
1159+
req: DropVirtualColumnReq,
1160+
) -> Result<DropVirtualColumnReply, KVAppError> {
1161+
debug!(req = debug(&req), "SchemaApi: {}", func_name!());
1162+
1163+
let mut retry = 0;
1164+
while retry < TXN_MAX_RETRY_TIMES {
1165+
retry += 1;
1166+
1167+
let (_, old_virtual_column_opt): (_, Option<VirtualColumnMeta>) =
1168+
get_pb_value(self, &req.name_ident).await?;
1169+
1170+
// If virtual columns already exist, merge them together
1171+
let virtual_column_meta = if let Some(old_virtual_column) = old_virtual_column_opt {
1172+
let mut virtual_columns_set: HashSet<String> = HashSet::from_iter(
1173+
old_virtual_column
1174+
.virtual_columns
1175+
.iter()
1176+
.cloned()
1177+
.collect::<Vec<_>>(),
1178+
);
1179+
for remove_virtual_column in &req.virtual_columns {
1180+
virtual_columns_set.remove(remove_virtual_column);
1181+
}
1182+
let virtual_columns = Vec::from_iter(virtual_columns_set.into_iter());
1183+
1184+
VirtualColumnMeta {
1185+
table_id: req.name_ident.table_id,
1186+
virtual_columns,
1187+
created_on: old_virtual_column.created_on.clone(),
1188+
updated_on: old_virtual_column.updated_on.clone(),
1189+
drop_on: Some(Utc::now()),
1190+
}
1191+
} else {
1192+
return Err(KVAppError::AppError(AppError::VirtualColumnNotExist(
1193+
VirtualColumnNotExist::new(
1194+
*&req.name_ident.table_id,
1195+
format!(
1196+
"drop virtual column with tenant: {} table_id: {}",
1197+
req.name_ident.tenant, req.name_ident.table_id
1198+
),
1199+
),
1200+
)));
1201+
};
1202+
1203+
// Create virtual column by inserting these record:
1204+
// (tenant, table_id) -> virtual_column_meta
1205+
{
1206+
let condition = vec![];
1207+
let if_then = vec![
1208+
txn_op_put(&req.name_ident, serialize_struct(&virtual_column_meta)?), /* (tenant, table_id) -> virtual_column_meta */
1209+
];
1210+
1211+
let txn_req = TxnRequest {
1212+
condition,
1213+
if_then,
1214+
else_then: vec![],
1215+
};
1216+
1217+
let (succ, _responses) = send_txn(self, txn_req).await?;
1218+
1219+
debug!(
1220+
req.name_ident = debug(&virtual_column_meta),
1221+
succ = display(succ),
1222+
"drop_virtual_column"
1223+
);
1224+
1225+
if succ {
1226+
return Ok(DropVirtualColumnReply {});
1227+
}
1228+
}
1229+
}
1230+
1231+
Err(KVAppError::AppError(AppError::TxnRetryMaxTimes(
1232+
TxnRetryMaxTimes::new("drop_virtual_column", TXN_MAX_RETRY_TIMES),
1233+
)))
1234+
}
1235+
1236+
async fn list_virtual_columns(
1237+
&self,
1238+
req: ListVirtualColumnsReq,
1239+
) -> Result<Vec<VirtualColumnMeta>, KVAppError> {
1240+
debug!(req = debug(&req), "SchemaApi: {}", func_name!());
1241+
1242+
if let Some(table_id) = req.table_id {
1243+
let name_ident = VirtualColumnNameIdent {
1244+
tenant: req.tenant.clone(),
1245+
table_id,
1246+
};
1247+
let (_, virtual_column_opt): (_, Option<VirtualColumnMeta>) =
1248+
get_pb_value(self, &name_ident).await?;
1249+
1250+
if let Some(virtual_column) = virtual_column_opt {
1251+
return Ok(vec![virtual_column]);
1252+
} else {
1253+
return Ok(vec![]);
1254+
}
1255+
}
1256+
1257+
// Get virtual columns list by `prefix_list` "<prefix>/<tenant>"
1258+
let prefix_key = kvapi::KeyBuilder::new_prefixed(VirtualColumnNameIdent::PREFIX)
1259+
.push_str(&req.tenant)
1260+
.done();
1261+
1262+
let list = self.prefix_list_kv(&prefix_key).await?;
1263+
let mut virtual_column_list = Vec::with_capacity(list.len());
1264+
for (_, seq) in list.iter() {
1265+
let virtual_column_meta: VirtualColumnMeta = deserialize_struct(&seq.data)?;
1266+
virtual_column_list.push(virtual_column_meta);
1267+
}
1268+
1269+
Ok(virtual_column_list)
1270+
}
1271+
10741272
#[tracing::instrument(level = "debug", ret, err, skip_all)]
10751273
async fn create_table(&self, req: CreateTableReq) -> Result<CreateTableReply, KVAppError> {
10761274
debug!(req = debug(&req), "SchemaApi: {}", func_name!());

0 commit comments

Comments
 (0)