diff --git a/sbroad-cartridge/src/cartridge/config.rs b/sbroad-cartridge/src/cartridge/config.rs
index cfd339ed55c70bcc9dce475b0dc19cd763e1f714..960f9026cf146193df36fdef84c8436ff725c64d 100644
--- a/sbroad-cartridge/src/cartridge/config.rs
+++ b/sbroad-cartridge/src/cartridge/config.rs
@@ -244,13 +244,12 @@ impl RouterConfiguration {
.iter()
.map(String::as_str)
.collect::<Vec<&str>>();
- let t = Table::new(
+ let t = Table::new_sharded(
&table_name,
fields,
shard_key_str.as_slice(),
primary_key_str.as_slice(),
engine,
- false,
)?;
self.tables.insert(table_name, t);
} else {
diff --git a/sbroad-cartridge/src/cartridge/config/tests.rs b/sbroad-cartridge/src/cartridge/config/tests.rs
index 98fbec2fd3ed2e6df42b436e7ab556e75c78389f..8bb83c9d963a7995184e07043e166fa540df6ade 100644
--- a/sbroad-cartridge/src/cartridge/config/tests.rs
+++ b/sbroad-cartridge/src/cartridge/config/tests.rs
@@ -140,7 +140,7 @@ fn test_getting_table_segment() {
s.set_sharding_column("\"bucket_id\"".into());
s.load_schema(test_schema).unwrap();
- let expected = Table::new(
+ let expected = Table::new_sharded(
"\"hash_testing\"",
vec![
Column::new(
@@ -158,7 +158,6 @@ fn test_getting_table_segment() {
&["\"identification_number\"", "\"product_code\""],
&["\"identification_number\""],
SpaceEngine::Memtx,
- false,
)
.unwrap();
diff --git a/sbroad-core/src/backend/sql/tree/tests.rs b/sbroad-core/src/backend/sql/tree/tests.rs
index c9476f0513fcb69fbd279ecbd1b9fb801bcbb3e7..77e128e8a1d617e7a6e5e30370f887857695a377 100644
--- a/sbroad-core/src/backend/sql/tree/tests.rs
+++ b/sbroad-core/src/backend/sql/tree/tests.rs
@@ -19,13 +19,12 @@ fn sql_order_selection() {
// select a from t where a = 1
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![column_user_non_null(String::from("a"), Type::Boolean)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -99,7 +98,7 @@ fn sql_order_selection() {
fn sql_arithmetic_selection_plan() {
// select a from t where a + (b/c + d*e) * f - b = 1
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_integer_user_non_null(String::from("a")),
@@ -113,7 +112,6 @@ fn sql_arithmetic_selection_plan() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -297,7 +295,7 @@ fn sql_arithmetic_selection_plan() {
fn sql_arithmetic_projection_plan() {
// select a + (b/c + d*e) * f - b from t
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_integer_user_non_null(String::from("a")),
@@ -311,7 +309,6 @@ fn sql_arithmetic_projection_plan() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -472,7 +469,7 @@ fn sql_arithmetic_projection_plan() {
fn sql_arbitrary_projection_plan() {
// select a + b > c and d is not null from t
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_integer_user_non_null(String::from("a")),
@@ -484,7 +481,6 @@ fn sql_arbitrary_projection_plan() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
diff --git a/sbroad-core/src/executor/engine/mock.rs b/sbroad-core/src/executor/engine/mock.rs
index 518ae9b9f877ab99a8d08e0e93c4be506e649996..ba42a672e7be99ea4eb72d5623ccacfee1dfa7e7 100644
--- a/sbroad-core/src/executor/engine/mock.rs
+++ b/sbroad-core/src/executor/engine/mock.rs
@@ -121,26 +121,24 @@ impl RouterConfigurationMock {
let primary_key = &["\"product_code\"", "\"identification_number\""];
tables.insert(
"\"hash_testing\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"hash_testing\"",
columns.clone(),
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
tables.insert(
"\"hash_testing_hist\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"hash_testing_hist\"",
columns.clone(),
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -148,26 +146,24 @@ impl RouterConfigurationMock {
let sharding_key = &["\"identification_number\""];
tables.insert(
"\"hash_single_testing\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"hash_single_testing\"",
columns.clone(),
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
tables.insert(
"\"hash_single_testing_hist\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"hash_single_testing_hist\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -184,26 +180,24 @@ impl RouterConfigurationMock {
tables.insert(
"\"test_space\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"test_space\"",
columns.clone(),
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
tables.insert(
"\"test_space_hist\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"test_space_hist\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -216,13 +210,12 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"id\""];
tables.insert(
"\"history\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"history\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -238,13 +231,12 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"b\""];
tables.insert(
"\"t\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"t\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -258,13 +250,12 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"a\"", "\"b\""];
tables.insert(
"\"t1\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"t1\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -280,13 +271,12 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"g\"", "\"h\""];
tables.insert(
"\"t2\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"t2\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -300,13 +290,12 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"a\""];
tables.insert(
"\"t3\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"t3\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
@@ -315,19 +304,10 @@ impl RouterConfigurationMock {
Column::new("\"a\"", Type::Integer, ColumnRole::User, false),
Column::new("\"b\"", Type::Integer, ColumnRole::User, false),
];
- let sharding_key: &[&str] = &[];
let primary_key: &[&str] = &["\"a\""];
tables.insert(
"\"global_t\"".to_string(),
- Table::new(
- "\"global_t\"",
- columns,
- sharding_key,
- primary_key,
- SpaceEngine::Memtx,
- false,
- )
- .unwrap(),
+ Table::new_global("\"global_t\"", columns, primary_key).unwrap(),
);
// Table for sbroad-benches
@@ -836,25 +816,23 @@ impl RouterConfigurationMock {
let primary_key: &[&str] = &["\"reestrid\""];
tables.insert(
"\"test__gibdd_db__vehicle_reg_and_res100_actual\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"test__gibdd_db__vehicle_reg_and_res100_actual\"",
columns.clone(),
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
tables.insert(
"\"test__gibdd_db__vehicle_reg_and_res100_history\"".to_string(),
- Table::new(
+ Table::new_sharded(
"\"test__gibdd_db__vehicle_reg_and_res100_history\"",
columns,
sharding_key,
primary_key,
SpaceEngine::Memtx,
- false,
)
.unwrap(),
);
diff --git a/sbroad-core/src/ir/distribution.rs b/sbroad-core/src/ir/distribution.rs
index 1e8b321d85b937cc27e6fe9db8f86632927d35f1..ef0c82e2542dc6ab01ee61181439ab5c48b7a4da 100644
--- a/sbroad-core/src/ir/distribution.rs
+++ b/sbroad-core/src/ir/distribution.rs
@@ -12,6 +12,7 @@ use crate::ir::transformation::redistribution::{MotionKey, Target};
use super::expression::Expression;
use super::operator::Relational;
+use super::relation::{Column, ColumnPositions};
use super::{Node, Plan};
/// Tuple columns that determinate its segment distribution.
@@ -32,6 +33,38 @@ impl Key {
Key { positions }
}
+ pub(crate) fn with_columns(
+ columns: &[Column],
+ pos_map: &ColumnPositions,
+ sharding_key: &[&str],
+ ) -> Result<Self, SbroadError> {
+ let shard_positions = sharding_key
+ .iter()
+ .map(|name| match pos_map.get(name) {
+ Some(pos) => {
+ // Check that the column type is scalar.
+ // Compound types are not supported as sharding keys.
+ let column = &columns.get(pos).ok_or_else(|| {
+ SbroadError::FailedTo(
+ Action::Create,
+ Some(Entity::Column),
+ format!("column {name} not found at position {pos}"),
+ )
+ })?;
+ if !column.r#type.is_scalar() {
+ return Err(SbroadError::Invalid(
+ Entity::Column,
+ Some(format!("column {name} at position {pos} is not scalar",)),
+ ));
+ }
+ Ok(pos)
+ }
+ None => Err(SbroadError::Invalid(Entity::ShardingKey, None)),
+ })
+ .collect::<Result<Vec<usize>, _>>()?;
+ Ok(Key::new(shard_positions))
+ }
+
#[must_use]
pub fn is_empty(&self) -> bool {
self.positions.is_empty()
diff --git a/sbroad-core/src/ir/distribution/tests.rs b/sbroad-core/src/ir/distribution/tests.rs
index 33f640631a556782c769200f90ec6c2ffe39f5f5..986f485af0b1843e147e4e7e93e4794ec0423858 100644
--- a/sbroad-core/src/ir/distribution/tests.rs
+++ b/sbroad-core/src/ir/distribution/tests.rs
@@ -12,7 +12,7 @@ use std::path::Path;
fn proj_preserve_dist_key() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -23,7 +23,6 @@ fn proj_preserve_dist_key() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
diff --git a/sbroad-core/src/ir/expression/tests.rs b/sbroad-core/src/ir/expression/tests.rs
index 10c0d6448f45feff832a5e173066158994684602..2ac4a2801163e4e66c4951d10c92245817a6d8ec 100644
--- a/sbroad-core/src/ir/expression/tests.rs
+++ b/sbroad-core/src/ir/expression/tests.rs
@@ -27,13 +27,12 @@ fn rel_nodes_from_reference_in_scan() {
// select * from t
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -50,13 +49,12 @@ fn rel_nodes_from_reference_in_proj() {
// select a from t
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
diff --git a/sbroad-core/src/ir/operator/tests.rs b/sbroad-core/src/ir/operator/tests.rs
index 9061b3f117ce708408f77ef98a036ea1620d4efe..460ecf826f09ebb0b5c185f26f11fde1ab63dea9 100644
--- a/sbroad-core/src/ir/operator/tests.rs
+++ b/sbroad-core/src/ir/operator/tests.rs
@@ -18,7 +18,7 @@ use super::*;
fn scan_rel() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -29,7 +29,6 @@ fn scan_rel() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -57,7 +56,7 @@ fn scan_rel() {
fn scan_rel_serialized() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -68,7 +67,6 @@ fn scan_rel_serialized() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -94,7 +92,7 @@ fn scan_rel_serialized() {
fn projection() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -105,7 +103,6 @@ fn projection() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -152,7 +149,7 @@ fn selection() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -163,7 +160,6 @@ fn selection() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -217,26 +213,24 @@ fn selection_serialize() {
fn except() {
let mut valid_plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_user_non_null(String::from("a"), Type::Unsigned)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
let t1_copy = t1.clone();
valid_plan.add_rel(t1);
let scan_t1_id = valid_plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_user_non_null(String::from("a"), Type::Unsigned)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
valid_plan.add_rel(t2);
@@ -250,7 +244,7 @@ fn except() {
invalid_plan.add_rel(t1_copy);
let scan_t1_id = invalid_plan.add_scan("t1", None).unwrap();
- let t3 = Table::new(
+ let t3 = Table::new_sharded(
"t3",
vec![
column_user_non_null(String::from("a"), Type::Unsigned),
@@ -259,7 +253,6 @@ fn except() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
invalid_plan.add_rel(t3);
@@ -278,20 +271,19 @@ fn except() {
fn insert() {
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_user_non_null(String::from("a"), Type::Unsigned)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_user_non_null(String::from("a"), Type::Unsigned),
@@ -301,7 +293,6 @@ fn insert() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -343,25 +334,23 @@ fn insert() {
fn union_all() {
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_user_non_null(String::from("a"), Type::Unsigned)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_user_non_null(String::from("a"), Type::Unsigned)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -374,7 +363,7 @@ fn union_all() {
fn union_all_col_amount_mismatch() {
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -383,7 +372,6 @@ fn union_all_col_amount_mismatch() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
@@ -391,13 +379,12 @@ fn union_all_col_amount_mismatch() {
let scan_t1_id = plan.add_scan("t1", None).unwrap();
// Check errors for children with different amount of column
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_user_non_null(String::from("b"), Type::Unsigned)],
&["b"],
&["b"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -416,7 +403,7 @@ fn union_all_col_amount_mismatch() {
fn sub_query() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -425,7 +412,6 @@ fn sub_query() {
&["a"],
&["b"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
@@ -466,26 +452,24 @@ fn selection_with_sub_query() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_integer_user_non_null(String::from("b"))],
&["b"],
&["b"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -524,7 +508,7 @@ fn join() {
// i.e. (a), (d) - tuples containing a single column.
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -534,13 +518,12 @@ fn join() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1 = plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_user_non_null(String::from("c"), Type::Boolean),
@@ -550,7 +533,6 @@ fn join() {
&["d"],
&["d"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -587,7 +569,7 @@ fn join_duplicate_columns() {
// select * from t1 join t2 on t1.a = t2.d
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -597,13 +579,12 @@ fn join_duplicate_columns() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1 = plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -613,7 +594,6 @@ fn join_duplicate_columns() {
&["d"],
&["d"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
diff --git a/sbroad-core/src/ir/relation.rs b/sbroad-core/src/ir/relation.rs
index a81e556f0b56bfc27a7303469a7b9170138de77b..1bc1723e607ca3ff6ab77900171dcb3f036158b8 100644
--- a/sbroad-core/src/ir/relation.rs
+++ b/sbroad-core/src/ir/relation.rs
@@ -7,6 +7,7 @@
//! * Table, representing unnamed tuples storage (`Table`)
//! * Relation, representing named tables (`Relations` as a map of { name -> table })
+use ahash::AHashMap;
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::fmt::{self, Formatter};
@@ -381,16 +382,33 @@ impl TryFrom<&str> for SpaceEngine {
}
}
-/// Table is a tuple storage in the cluster.
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
-pub struct Table {
- /// List of the columns.
- pub columns: Vec<Column>,
- /// Primary key of the table (column positions).
- pub primary_key: Key,
- /// Unique table name.
- pub name: String,
- pub kind: TableKind,
+// A helper struct to collect column positions.
+#[derive(Debug)]
+pub(crate) struct ColumnPositions<'column> {
+ // Column positions with names as keys.
+ map: AHashMap<&'column str, usize>,
+}
+
+impl<'column> ColumnPositions<'column> {
+ #[allow(clippy::uninlined_format_args)]
+ pub(crate) fn new(columns: &'column [Column], table: &str) -> Result<Self, SbroadError> {
+ let mut map = AHashMap::with_capacity(columns.len());
+ for (pos, col) in columns.iter().enumerate() {
+ let name = col.name.as_str();
+ if let Some(old_pos) = map.insert(name, pos) {
+ return Err(SbroadError::DuplicatedValue(format!(
+ r#"Table "{}" has a duplicating column "{}" at positions {} and {}"#,
+ table, name, old_pos, pos,
+ )));
+ }
+ }
+ map.shrink_to_fit();
+ Ok(Self { map })
+ }
+
+ pub(crate) fn get(&self, name: &str) -> Option<usize> {
+ self.map.get(name).copied()
+ }
}
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
@@ -403,101 +421,130 @@ pub enum TableKind {
SystemSpace,
}
+impl TableKind {
+ #[must_use]
+ pub fn new_sharded(sharding_key: Key, engine: SpaceEngine) -> Self {
+ Self::ShardedSpace {
+ sharding_key,
+ engine,
+ }
+ }
+
+ #[must_use]
+ pub fn new_global() -> Self {
+ Self::GlobalSpace
+ }
+
+ #[must_use]
+ pub fn new_system() -> Self {
+ Self::SystemSpace
+ }
+}
+
+fn table_new_impl<'column>(
+ name: &str,
+ columns: &'column [Column],
+ primary_key: &'column [&str],
+) -> Result<(ColumnPositions<'column>, Key), SbroadError> {
+ let pos_map = ColumnPositions::new(columns, name)?;
+ let primary_positions = primary_key
+ .iter()
+ .map(|name| match pos_map.get(name) {
+ Some(pos) => {
+ let _ = &columns.get(pos).ok_or_else(|| {
+ SbroadError::FailedTo(
+ Action::Create,
+ Some(Entity::Column),
+ format!("column {name} not found at position {pos}"),
+ )
+ })?;
+ Ok(pos)
+ }
+ None => Err(SbroadError::Invalid(Entity::PrimaryKey, None)),
+ })
+ .collect::<Result<Vec<usize>, _>>()?;
+ Ok((pos_map, Key::new(primary_positions)))
+}
+
+/// Table is a tuple storage in the cluster.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct Table {
+ /// List of the columns.
+ pub columns: Vec<Column>,
+ /// Primary key of the table (column positions).
+ pub primary_key: Key,
+ /// Unique table name.
+ pub name: String,
+ pub kind: TableKind,
+}
+
impl Table {
#[must_use]
pub fn name(&self) -> &str {
&self.name
}
- /// Table constructor.
+ /// Sharded table constructor.
///
/// # Errors
- /// Returns `SbroadError` when the input arguments are invalid.
- pub fn new(
+ /// - column names are duplicated;
+ /// - primary key is not found among the columns;
+ /// - sharding key is not found among the columns;
+ pub fn new_sharded(
name: &str,
columns: Vec<Column>,
sharding_key: &[&str],
primary_key: &[&str],
engine: SpaceEngine,
- is_system: bool,
) -> Result<Self, SbroadError> {
- let mut pos_map: HashMap<&str, usize> = HashMap::new();
- let no_duplicates = &columns
- .iter()
- .enumerate()
- .all(|(pos, col)| matches!(pos_map.insert(&col.name, pos), None));
-
- if !no_duplicates {
- return Err(SbroadError::DuplicatedValue(
- "Table has duplicated columns and couldn't be loaded".into(),
- ));
- }
-
- let table_kind = if sharding_key.is_empty() {
- if engine != SpaceEngine::Memtx {
- return Err(SbroadError::Unsupported(
- Entity::Table,
- Some("global table can only have memtx engine".into()),
- ));
- }
- if is_system {
- TableKind::SystemSpace
- } else {
- TableKind::GlobalSpace
- }
- } else {
- let shard_positions = sharding_key
- .iter()
- .map(|name| match pos_map.get(*name) {
- Some(pos) => {
- // Check that the column type is scalar.
- // Compound types are not supported as sharding keys.
- let column = &columns.get(*pos).ok_or_else(|| {
- SbroadError::FailedTo(
- Action::Create,
- Some(Entity::Column),
- format!("column {name} not found at position {pos}"),
- )
- })?;
- if !column.r#type.is_scalar() {
- return Err(SbroadError::Invalid(
- Entity::Column,
- Some(format!("column {name} at position {pos} is not scalar",)),
- ));
- }
- Ok(*pos)
- }
- None => Err(SbroadError::Invalid(Entity::ShardingKey, None)),
- })
- .collect::<Result<Vec<usize>, _>>()?;
- TableKind::ShardedSpace {
- sharding_key: Key::new(shard_positions),
- engine,
- }
- };
+ let (pos_map, primary_key) = table_new_impl(name, &columns, primary_key)?;
+ let sharding_key = Key::with_columns(&columns, &pos_map, sharding_key)?;
+ let kind = TableKind::new_sharded(sharding_key, engine);
+ Ok(Table {
+ name: name.into(),
+ columns,
+ primary_key,
+ kind,
+ })
+ }
- let primary_positions = primary_key
- .iter()
- .map(|name| match pos_map.get(*name) {
- Some(pos) => {
- let _ = &columns.get(*pos).ok_or_else(|| {
- SbroadError::FailedTo(
- Action::Create,
- Some(Entity::Column),
- format!("column {name} not found at position {pos}"),
- )
- })?;
- Ok(*pos)
- }
- None => Err(SbroadError::Invalid(Entity::PrimaryKey, None)),
- })
- .collect::<Result<Vec<usize>, _>>()?;
+ /// Global table constructor.
+ ///
+ /// # Errors
+ /// - column names are duplicated;
+ /// - primary key is not found among the columns;
+ pub fn new_global(
+ name: &str,
+ columns: Vec<Column>,
+ primary_key: &[&str],
+ ) -> Result<Self, SbroadError> {
+ let (_, primary_key) = table_new_impl(name, &columns, primary_key)?;
+ let kind = TableKind::new_global();
+ Ok(Table {
+ name: name.into(),
+ columns,
+ primary_key,
+ kind,
+ })
+ }
+ /// System table constructor.
+ ///
+ /// # Errors
+ /// - column names are duplicated;
+ /// - primary key is not found among the columns;
+ pub fn new_system(
+ name: &str,
+ columns: Vec<Column>,
+ primary_key: &[&str],
+ ) -> Result<Self, SbroadError> {
+ let (_, primary_key) = table_new_impl(name, &columns, primary_key)?;
+ let kind = TableKind::new_system();
Ok(Table {
name: name.into(),
columns,
- primary_key: Key::new(primary_positions),
- kind: table_kind,
+ primary_key,
+ kind,
})
}
@@ -566,9 +613,10 @@ impl Table {
Ordering::Greater => Err(SbroadError::UnexpectedNumberOfValues(
"Table has more than one bucket_id column".into(),
)),
- Ordering::Less => Err(SbroadError::UnexpectedNumberOfValues(
- "Table has no bucket_id columns".into(),
- )),
+ Ordering::Less => Err(SbroadError::UnexpectedNumberOfValues(format!(
+ "Table {} has no bucket_id columns",
+ self.name
+ ))),
}
}
diff --git a/sbroad-core/src/ir/relation/tests.rs b/sbroad-core/src/ir/relation/tests.rs
index b4c7c90f0a8069a0038eae9d827eb71a9144637f..ee9ffaa2ed28b737ed5154ee01baf94d2c49ed1d 100644
--- a/sbroad-core/src/ir/relation/tests.rs
+++ b/sbroad-core/src/ir/relation/tests.rs
@@ -16,7 +16,7 @@ fn column() {
#[test]
fn table_seg() {
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -27,7 +27,6 @@ fn table_seg() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
@@ -39,13 +38,12 @@ fn table_seg() {
#[test]
fn table_seg_name() {
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![column_user_non_null(String::from("a"), Type::Boolean)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
assert_eq!("t", t.name());
@@ -54,7 +52,7 @@ fn table_seg_name() {
#[test]
fn table_seg_duplicate_columns() {
assert_eq!(
- Table::new(
+ Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -65,16 +63,17 @@ fn table_seg_duplicate_columns() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false
)
.unwrap_err(),
- SbroadError::DuplicatedValue("Table has duplicated columns and couldn't be loaded".into())
+ SbroadError::DuplicatedValue(format!(
+ r#"Table "t" has a duplicating column "a" at positions 0 and 3"#,
+ ))
);
}
#[test]
fn table_seg_dno_bucket_id_column() {
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -84,16 +83,15 @@ fn table_seg_dno_bucket_id_column() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
assert_eq!(
- SbroadError::UnexpectedNumberOfValues("Table has no bucket_id columns".into()),
+ SbroadError::UnexpectedNumberOfValues(format!("Table {} has no bucket_id columns", "t")),
t1.get_bucket_id_position().unwrap_err()
);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -105,7 +103,6 @@ fn table_seg_dno_bucket_id_column() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
@@ -118,7 +115,7 @@ fn table_seg_dno_bucket_id_column() {
#[test]
fn table_seg_wrong_key() {
assert_eq!(
- Table::new(
+ Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -127,9 +124,8 @@ fn table_seg_wrong_key() {
column_user_non_null(String::from("d"), Type::String),
],
&["a", "e"],
- &["a", "e"],
+ &["a"],
SpaceEngine::Memtx,
- false
)
.unwrap_err(),
SbroadError::Invalid(Entity::ShardingKey, None)
@@ -139,7 +135,7 @@ fn table_seg_wrong_key() {
#[test]
fn table_seg_compound_type_in_key() {
assert_eq!(
- Table::new(
+ Table::new_sharded(
"t",
vec![
Column::new("bucket_id", Type::Unsigned, ColumnRole::Sharding, false),
@@ -148,7 +144,6 @@ fn table_seg_compound_type_in_key() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false
)
.unwrap_err(),
SbroadError::Invalid(
@@ -160,7 +155,7 @@ fn table_seg_compound_type_in_key() {
#[test]
fn table_seg_serialized() {
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -173,7 +168,6 @@ fn table_seg_serialized() {
&["a", "d"],
&["a", "d"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
let path = Path::new("")
@@ -336,7 +330,7 @@ fn column_msgpack_deserialize() {
#[test]
fn table_converting() {
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -347,7 +341,6 @@ fn table_converting() {
&["b", "a"],
&["b", "a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
diff --git a/sbroad-core/src/ir/tests.rs b/sbroad-core/src/ir/tests.rs
index f6c128800207534415fbe35bcf59eb421d9c4adf..ee2e10f4a97cd36e1db55689b9d0665d51699fd7 100644
--- a/sbroad-core/src/ir/tests.rs
+++ b/sbroad-core/src/ir/tests.rs
@@ -84,13 +84,12 @@ fn plan_oor_top() {
fn get_node() {
let mut plan = Plan::default();
- let t = Table::new(
+ let t = Table::new_sharded(
"t",
vec![Column::new("a", Type::Boolean, ColumnRole::User, false)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t);
diff --git a/sbroad-core/src/ir/transformation/redistribution/tests.rs b/sbroad-core/src/ir/transformation/redistribution/tests.rs
index 4b1012f99450116c26802647e986e56c98272d82..75902704c302a5eed80f8280266a5925733091eb 100644
--- a/sbroad-core/src/ir/transformation/redistribution/tests.rs
+++ b/sbroad-core/src/ir/transformation/redistribution/tests.rs
@@ -17,20 +17,19 @@ fn full_motion_less_for_sub_query() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Vinyl,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_integer_user_non_null(String::from("a")),
@@ -39,7 +38,6 @@ fn full_motion_less_for_sub_query() {
&["a"],
&["a"],
SpaceEngine::Vinyl,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -82,7 +80,7 @@ fn full_motion_non_segment_outer_for_sub_query() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![
column_integer_user_non_null(String::from("a")),
@@ -91,20 +89,18 @@ fn full_motion_non_segment_outer_for_sub_query() {
&["a"],
&["a"],
SpaceEngine::Vinyl,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Vinyl,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -147,26 +143,24 @@ fn local_sub_query() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -208,20 +202,19 @@ fn multiple_sub_queries() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_integer_user_non_null(String::from("a")),
@@ -230,7 +223,6 @@ fn multiple_sub_queries() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
diff --git a/sbroad-core/src/ir/transformation/redistribution/tests/segment.rs b/sbroad-core/src/ir/transformation/redistribution/tests/segment.rs
index b904f07cc2df9fe448436eec96667980ba99625b..894265a7ff96fb74ae0265d3b63b1a4f819db59d 100644
--- a/sbroad-core/src/ir/transformation/redistribution/tests/segment.rs
+++ b/sbroad-core/src/ir/transformation/redistribution/tests/segment.rs
@@ -23,20 +23,19 @@ fn sub_query1() {
let mut plan = Plan::default();
let mut children: Vec<usize> = Vec::new();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_integer_user_non_null(String::from("a"))],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
children.push(scan_t1_id);
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_integer_user_non_null(String::from("a")),
@@ -45,7 +44,6 @@ fn sub_query1() {
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
diff --git a/sbroad-core/src/ir/tree/tests.rs b/sbroad-core/src/ir/tree/tests.rs
index ed69dc4c57092a9492a5cb4e3d19bdb97ef01b4a..85ae443a13bea329a57b02c32acbdba1cdf7fde5 100644
--- a/sbroad-core/src/ir/tree/tests.rs
+++ b/sbroad-core/src/ir/tree/tests.rs
@@ -54,25 +54,23 @@ fn relational_post() {
// Initialize plan
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_user_non_null(String::from("a"), Type::Boolean)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![column_user_non_null(String::from("a"), Type::Boolean)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -120,20 +118,19 @@ fn selection_subquery_dfs_post() {
// Initialize plan
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![column_user_non_null(String::from("a"), Type::Boolean)],
&["a"],
&["a"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);
let scan_t1_id = plan.add_scan("t1", None).unwrap();
let a = plan.add_row_from_child(scan_t1_id, &["a"]).unwrap();
- let t2 = Table::new(
+ let t2 = Table::new_sharded(
"t2",
vec![
column_user_non_null(String::from("b"), Type::Boolean),
@@ -142,7 +139,6 @@ fn selection_subquery_dfs_post() {
&["b"],
&["b"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t2);
@@ -209,7 +205,7 @@ fn subtree_dfs_post() {
// Initialize plan
let mut plan = Plan::default();
- let t1 = Table::new(
+ let t1 = Table::new_sharded(
"t1",
vec![
column_user_non_null(String::from("a"), Type::Boolean),
@@ -218,7 +214,6 @@ fn subtree_dfs_post() {
&["a", "c"],
&["a", "c"],
SpaceEngine::Memtx,
- false,
)
.unwrap();
plan.add_rel(t1);