Skip to content

Commit 1abffb2

Browse files
b41shmergify[bot]
b41sh
and
mergify[bot]
authored
feat(query): Support generate virtual columns (#11590)
* feat(query): Support generate virtual columns * fix * fix * fix taplo fmt * add tests * fix * fix * add test * fix * add tests * add alter virtual column * fix tests * add virtual block folder * fix * fix * fix meta api * fix * fix * fix --------- Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
1 parent 786ee1f commit 1abffb2

File tree

81 files changed

+3501
-55
lines changed

Some content is hidden

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

81 files changed

+3501
-55
lines changed

‎Cargo.lock

+20
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

+2
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,8 @@ build_exceptions! {
142142
LicenceDenied(1112),
143143
UnknownDatamask(1113),
144144
UnmatchColumnDataType(1114),
145+
VirtualColumnNotFound(1115),
146+
VirtualColumnAlreadyExists(1116),
145147

146148
// Data Related Errors
147149

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

+30
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;
@@ -58,8 +63,11 @@ use common_meta_app::schema::UndropTableReply;
5863
use common_meta_app::schema::UndropTableReq;
5964
use common_meta_app::schema::UpdateTableMetaReply;
6065
use common_meta_app::schema::UpdateTableMetaReq;
66+
use common_meta_app::schema::UpdateVirtualColumnReply;
67+
use common_meta_app::schema::UpdateVirtualColumnReq;
6168
use common_meta_app::schema::UpsertTableOptionReply;
6269
use common_meta_app::schema::UpsertTableOptionReq;
70+
use common_meta_app::schema::VirtualColumnMeta;
6371
use common_meta_types::GCDroppedDataReply;
6472
use common_meta_types::GCDroppedDataReq;
6573
use common_meta_types::MetaId;
@@ -111,6 +119,28 @@ pub trait SchemaApi: Send + Sync {
111119
req: ListIndexesReq,
112120
) -> Result<Vec<(u64, String, IndexMeta)>, KVAppError>;
113121

122+
// virtual column
123+
124+
async fn create_virtual_column(
125+
&self,
126+
req: CreateVirtualColumnReq,
127+
) -> Result<CreateVirtualColumnReply, KVAppError>;
128+
129+
async fn update_virtual_column(
130+
&self,
131+
req: UpdateVirtualColumnReq,
132+
) -> Result<UpdateVirtualColumnReply, KVAppError>;
133+
134+
async fn drop_virtual_column(
135+
&self,
136+
req: DropVirtualColumnReq,
137+
) -> Result<DropVirtualColumnReply, KVAppError>;
138+
139+
async fn list_virtual_columns(
140+
&self,
141+
req: ListVirtualColumnsReq,
142+
) -> Result<Vec<VirtualColumnMeta>, KVAppError>;
143+
114144
// table
115145

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

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

+206
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ use common_meta_app::app_error::UnknownDatabaseId;
4141
use common_meta_app::app_error::UnknownIndex;
4242
use common_meta_app::app_error::UnknownTable;
4343
use common_meta_app::app_error::UnknownTableId;
44+
use common_meta_app::app_error::VirtualColumnAlreadyExists;
4445
use common_meta_app::app_error::WrongShare;
4546
use common_meta_app::app_error::WrongShareObject;
4647
use common_meta_app::schema::CountTablesKey;
@@ -54,6 +55,8 @@ use common_meta_app::schema::CreateTableLockRevReply;
5455
use common_meta_app::schema::CreateTableLockRevReq;
5556
use common_meta_app::schema::CreateTableReply;
5657
use common_meta_app::schema::CreateTableReq;
58+
use common_meta_app::schema::CreateVirtualColumnReply;
59+
use common_meta_app::schema::CreateVirtualColumnReq;
5760
use common_meta_app::schema::DBIdTableName;
5861
use common_meta_app::schema::DatabaseId;
5962
use common_meta_app::schema::DatabaseIdToName;
@@ -71,6 +74,8 @@ use common_meta_app::schema::DropIndexReply;
7174
use common_meta_app::schema::DropIndexReq;
7275
use common_meta_app::schema::DropTableByIdReq;
7376
use common_meta_app::schema::DropTableReply;
77+
use common_meta_app::schema::DropVirtualColumnReply;
78+
use common_meta_app::schema::DropVirtualColumnReq;
7479
use common_meta_app::schema::EmptyProto;
7580
use common_meta_app::schema::ExtendTableLockRevReq;
7681
use common_meta_app::schema::GetDatabaseReq;
@@ -85,6 +90,7 @@ use common_meta_app::schema::ListDatabaseReq;
8590
use common_meta_app::schema::ListIndexesReq;
8691
use common_meta_app::schema::ListTableLockRevReq;
8792
use common_meta_app::schema::ListTableReq;
93+
use common_meta_app::schema::ListVirtualColumnsReq;
8894
use common_meta_app::schema::RenameDatabaseReply;
8995
use common_meta_app::schema::RenameDatabaseReq;
9096
use common_meta_app::schema::RenameTableReply;
@@ -108,9 +114,13 @@ use common_meta_app::schema::UndropTableReply;
108114
use common_meta_app::schema::UndropTableReq;
109115
use common_meta_app::schema::UpdateTableMetaReply;
110116
use common_meta_app::schema::UpdateTableMetaReq;
117+
use common_meta_app::schema::UpdateVirtualColumnReply;
118+
use common_meta_app::schema::UpdateVirtualColumnReq;
111119
use common_meta_app::schema::UpsertTableCopiedFileReq;
112120
use common_meta_app::schema::UpsertTableOptionReply;
113121
use common_meta_app::schema::UpsertTableOptionReq;
122+
use common_meta_app::schema::VirtualColumnMeta;
123+
use common_meta_app::schema::VirtualColumnNameIdent;
114124
use common_meta_app::share::ShareGrantObject;
115125
use common_meta_app::share::ShareId;
116126
use common_meta_app::share::ShareNameIdent;
@@ -163,6 +173,7 @@ use crate::util::deserialize_u64;
163173
use crate::util::get_index_metas_by_ids;
164174
use crate::util::get_table_by_id_or_err;
165175
use crate::util::get_table_names_by_ids;
176+
use crate::util::get_virtual_column_by_id_or_err;
166177
use crate::util::list_tables_from_share_db;
167178
use crate::util::list_tables_from_unshare_db;
168179
use crate::util::mget_pb_values;
@@ -1071,6 +1082,201 @@ impl<KV: kvapi::KVApi<Error = MetaError>> SchemaApi for KV {
10711082
Ok(index_metas)
10721083
}
10731084

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

0 commit comments

Comments
 (0)