Merge branch 'main' into hot-fix-settings

Author: mergify[bot]

src/meta/api/src/schema_api_test_suite.rs

@@ -1522,7 +1522,7 @@ impl SchemaApiTestSuite {
                 let ident = TableIdent::new(tb_id, seq);
 
                 let want = TableInfo {
-                    ident: ident.clone(),
+                    ident,
                     desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                     name: tbl_name.into(),
                     meta: table_meta(created_on),
@@ -1549,7 +1549,7 @@ impl SchemaApiTestSuite {
 
             let got = mt.get_table((tenant, db_name, tbl_name).into()).await?;
             let want = TableInfo {
-                ident: tb_ident_2.clone(),
+                ident: tb_ident_2,
                 desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                 name: tbl_name.into(),
                 meta: table_meta(created_on),
@@ -1584,7 +1584,7 @@ impl SchemaApiTestSuite {
 
             let got = mt.get_table((tenant, "db1", "tb2").into()).await.unwrap();
             let want = TableInfo {
-                ident: tb_ident_2.clone(),
+                ident: tb_ident_2,
                 desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                 name: tbl_name.into(),
                 meta: table_meta(created_on),
@@ -1784,7 +1784,7 @@ impl SchemaApiTestSuite {
             let cur_db = mt.get_database(Self::req_get_db(tenant, db1_name)).await?;
             assert!(old_db.ident.seq < cur_db.ident.seq);
             let got = mt.get_table((tenant, db1_name, tb2_name).into()).await?;
-            got.ident.clone()
+            got.ident
         };
 
         info!("--- rename table, ok");
@@ -1796,7 +1796,7 @@ impl SchemaApiTestSuite {
 
             let got = mt.get_table((tenant, db1_name, tb3_name).into()).await?;
             let want = TableInfo {
-                ident: tb_ident.clone(),
+                ident: tb_ident,
                 desc: format!("'{}'.'{}'.'{}'", tenant, db1_name, tb3_name),
                 name: tb3_name.into(),
                 meta: table_meta(created_on),
@@ -1842,7 +1842,7 @@ impl SchemaApiTestSuite {
             let got = mt.get_table((tenant, db1_name, tb2_name).into()).await?;
             assert_ne!(tb_ident.table_id, got.ident.table_id);
             assert_ne!(tb_ident.seq, got.ident.seq);
-            got.ident.clone()
+            got.ident
         };
 
         info!("--- db1,tb2(no_nil) -> db1,tb3(no_nil), error");
@@ -2010,7 +2010,7 @@ impl SchemaApiTestSuite {
                 let ident = TableIdent::new(tb_id, seq);
 
                 let want = TableInfo {
-                    ident: ident.clone(),
+                    ident,
                     desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                     name: tbl_name.into(),
                     meta: table_meta(created_on),
@@ -2140,7 +2140,7 @@ impl SchemaApiTestSuite {
                 let ident = TableIdent::new(tb_id, seq);
 
                 let want = TableInfo {
-                    ident: ident.clone(),
+                    ident,
                     desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                     name: tbl_name.into(),
                     meta: table_meta(created_on),
@@ -3128,7 +3128,7 @@ impl SchemaApiTestSuite {
                 let ident = TableIdent::new(tb_id, seq);
 
                 let want = TableInfo {
-                    ident: ident.clone(),
+                    ident,
                     desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name),
                     name: tbl_name.into(),
                     meta: table_meta(created_on),

src/meta/app/src/schema/table.rs

@@ -29,7 +29,7 @@ use maplit::hashmap;
 use crate::schema::database::DatabaseNameIdent;
 
 /// Globally unique identifier of a version of TableMeta.
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq, Default)]
+#[derive(serde::Serialize, serde::Deserialize, Clone, Copy, Debug, Eq, PartialEq, Default)]
 pub struct TableIdent {
     /// Globally unique id to identify a table.
     pub table_id: u64,

src/query/legacy-planners/src/plan_delete.rs

@@ -18,16 +18,30 @@ use common_datavalues::DataSchema;
 use common_datavalues::DataSchemaRef;
 use common_meta_app::schema::TableIdent;
 
-use crate::Expression;
 use crate::Projection;
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)]
+/// # TODO
+///
+/// From @xuanwo
+///
+/// Ideally, we need to use `Scalar` in DeletePlan.selection. But we met a
+/// cycle deps here. So we have to change `selection` in String first, and
+/// change into `Scalar` when our `Planner` has been moved out.
+///
+/// At this stage, DeletePlan's selection expr will be parsed twice:
+///
+/// - Parsed during `bind` to get column index and projection index.
+/// - Parsed during `execution` to get the correct columns
+///
+/// It's an ugly but necessary price to pay. Without this, we would sink in
+/// hell forever.
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct DeletePlan {
     pub catalog_name: String,
     pub database_name: String,
     pub table_name: String,
     pub table_id: TableIdent,
-    pub selection: Option<Expression>,
+    pub selection: Option<String>,
     pub projection: Projection,
 }
 

src/query/service/src/catalogs/default/immutable_catalog.rs

@@ -134,7 +134,7 @@ impl Catalog for ImmutableCatalog {
             .get_by_id(&table_id)
             .ok_or_else(|| ErrorCode::UnknownTable(format!("Unknown table id: '{}'", table_id)))?;
         let ti = table.get_table_info();
-        Ok((ti.ident.clone(), Arc::new(ti.meta.clone())))
+        Ok((ti.ident, Arc::new(ti.meta.clone())))
     }
 
     async fn get_table(

src/query/service/src/pipelines/processors/mod.rs

@@ -54,6 +54,7 @@ pub use transforms::KeyU64HashTable;
 pub use transforms::KeyU8HashTable;
 pub use transforms::MarkJoinCompactor;
 pub use transforms::ProjectionTransform;
+pub use transforms::RightJoinCompactor;
 pub use transforms::SerializerHashTable;
 pub use transforms::SinkBuildHashTable;
 pub use transforms::SortMergeCompactor;

src/query/service/src/pipelines/processors/transforms/hash_join/desc.rs

@@ -12,17 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use common_exception::Result;
 use common_functions::scalars::FunctionFactory;
 use parking_lot::RwLock;
 
 use crate::evaluator::EvalNode;
 use crate::evaluator::Evaluator;
+use crate::pipelines::processors::transforms::hash_join::row::RowPtr;
 use crate::pipelines::processors::transforms::hash_join::MarkJoinDesc;
 use crate::sql::executor::HashJoin;
 use crate::sql::executor::PhysicalScalar;
 use crate::sql::plans::JoinType;
 
+pub struct RightJoinDesc {
+    /// Record rows in build side that are matched with rows in probe side.
+    pub(crate) build_indexes: RwLock<Vec<RowPtr>>,
+    /// Record row in build side that is matched how many rows in probe side.
+    pub(crate) row_state: RwLock<HashMap<RowPtr, usize>>,
+}
+
+impl RightJoinDesc {
+    pub fn create() -> Self {
+        RightJoinDesc {
+            build_indexes: RwLock::new(Vec::new()),
+            row_state: RwLock::new(HashMap::new()),
+        }
+    }
+}
+
 pub struct HashJoinDesc {
     pub(crate) build_keys: Vec<EvalNode>,
     pub(crate) probe_keys: Vec<EvalNode>,
@@ -31,6 +50,7 @@ pub struct HashJoinDesc {
     pub(crate) marker_join_desc: MarkJoinDesc,
     /// Whether the Join are derived from correlated subquery.
     pub(crate) from_correlated_subquery: bool,
+    pub(crate) right_join_desc: RightJoinDesc,
 }
 
 impl HashJoinDesc {
@@ -50,6 +70,7 @@ impl HashJoinDesc {
                 marker_index: join.marker_index,
             },
             from_correlated_subquery: join.from_correlated_subquery,
+            right_join_desc: RightJoinDesc::create(),
         })
     }
 

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_state.rs

@@ -44,4 +44,7 @@ pub trait HashJoinState: Send + Sync {
 
     /// Get mark join results
     fn mark_join_blocks(&self) -> Result<Vec<DataBlock>>;
+
+    /// Get right join results
+    fn right_join_blocks(&self, blocks: &[DataBlock]) -> Result<Vec<DataBlock>>;
 }