Merge pull request #16 from cmu-db/cost-model-refine-schema

Modify cost model stats table schema to support update_stats

Author: lanlou1554

optd-persistent/src/entities/event.rs

@@ -18,8 +18,8 @@ pub enum Relation {
     GroupWinner,
     #[sea_orm(has_many = "super::plan_cost::Entity")]
     PlanCost,
-    #[sea_orm(has_many = "super::statistic::Entity")]
-    Statistic,
+    #[sea_orm(has_many = "super::versioned_statistic::Entity")]
+    VersionedStatistic,
 }
 
 impl Related<super::group_winner::Entity> for Entity {
@@ -34,9 +34,9 @@ impl Related<super::plan_cost::Entity> for Entity {
     }
 }
 
-impl Related<super::statistic::Entity> for Entity {
+impl Related<super::versioned_statistic::Entity> for Entity {
     fn to() -> RelationDef {
-        Relation::Statistic.def()
+        Relation::VersionedStatistic.def()
     }
 }
 

optd-persistent/src/entities/mod.rs

@@ -17,9 +17,11 @@ pub mod logical_property;
 pub mod namespace_metadata;
 pub mod physical_children;
 pub mod physical_expression;
+pub mod physical_expression_to_statistic_junction;
 pub mod physical_property;
 pub mod plan_cost;
 pub mod statistic;
 pub mod statistic_to_attribute_junction;
 pub mod table_metadata;
 pub mod trigger;
+pub mod versioned_statistic;

optd-persistent/src/entities/physical_expression.rs

@@ -27,6 +27,8 @@ pub enum Relation {
     GroupWinner,
     #[sea_orm(has_many = "super::physical_children::Entity")]
     PhysicalChildren,
+    #[sea_orm(has_many = "super::physical_expression_to_statistic_junction::Entity")]
+    PhysicalExpressionToStatisticJunction,
     #[sea_orm(has_many = "super::physical_property::Entity")]
     PhysicalProperty,
     #[sea_orm(has_many = "super::plan_cost::Entity")]
@@ -45,6 +47,12 @@ impl Related<super::physical_children::Entity> for Entity {
     }
 }
 
+impl Related<super::physical_expression_to_statistic_junction::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::PhysicalExpressionToStatisticJunction.def()
+    }
+}
+
 impl Related<super::physical_property::Entity> for Entity {
     fn to() -> RelationDef {
         Relation::PhysicalProperty.def()
@@ -70,4 +78,17 @@ impl Related<super::cascades_group::Entity> for Entity {
     }
 }
 
+impl Related<super::statistic::Entity> for Entity {
+    fn to() -> RelationDef {
+        super::physical_expression_to_statistic_junction::Relation::Statistic.def()
+    }
+    fn via() -> Option<RelationDef> {
+        Some(
+            super::physical_expression_to_statistic_junction::Relation::PhysicalExpression
+                .def()
+                .rev(),
+        )
+    }
+}
+
 impl ActiveModelBehavior for ActiveModel {}

optd-persistent/src/entities/physical_expression_to_statistic_junction.rs

@@ -0,0 +1,46 @@
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
+
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "physical_expression_to_statistic_junction")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub physical_expression_id: i32,
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub statistic_id: i32,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::physical_expression::Entity",
+        from = "Column::PhysicalExpressionId",
+        to = "super::physical_expression::Column::Id",
+        on_update = "Cascade",
+        on_delete = "Cascade"
+    )]
+    PhysicalExpression,
+    #[sea_orm(
+        belongs_to = "super::statistic::Entity",
+        from = "Column::StatisticId",
+        to = "super::statistic::Column::Id",
+        on_update = "Cascade",
+        on_delete = "Cascade"
+    )]
+    Statistic,
+}
+
+impl Related<super::physical_expression::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::PhysicalExpression.def()
+    }
+}
+
+impl Related<super::statistic::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Statistic.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

optd-persistent/src/entities/prelude.rs

@@ -15,9 +15,11 @@ pub use super::logical_property::Entity as LogicalProperty;
 pub use super::namespace_metadata::Entity as NamespaceMetadata;
 pub use super::physical_children::Entity as PhysicalChildren;
 pub use super::physical_expression::Entity as PhysicalExpression;
+pub use super::physical_expression_to_statistic_junction::Entity as PhysicalExpressionToStatisticJunction;
 pub use super::physical_property::Entity as PhysicalProperty;
 pub use super::plan_cost::Entity as PlanCost;
 pub use super::statistic::Entity as Statistic;
 pub use super::statistic_to_attribute_junction::Entity as StatisticToAttributeJunction;
 pub use super::table_metadata::Entity as TableMetadata;
 pub use super::trigger::Entity as Trigger;
+pub use super::versioned_statistic::Entity as VersionedStatistic;

optd-persistent/src/entities/statistic.rs

@@ -2,31 +2,23 @@
 
 use sea_orm::entity::prelude::*;
 
-#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
 #[sea_orm(table_name = "statistic")]
 pub struct Model {
     #[sea_orm(primary_key)]
     pub id: i32,
     pub name: String,
     pub table_id: i32,
-    pub epoch_id: i32,
     pub created_time: DateTimeUtc,
     pub number_of_attributes: i32,
     pub statistic_type: i32,
-    #[sea_orm(column_type = "Float")]
-    pub statistic_value: f32,
+    pub description: String,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
 pub enum Relation {
-    #[sea_orm(
-        belongs_to = "super::event::Entity",
-        from = "Column::EpochId",
-        to = "super::event::Column::EpochId",
-        on_update = "Cascade",
-        on_delete = "Cascade"
-    )]
-    Event,
+    #[sea_orm(has_many = "super::physical_expression_to_statistic_junction::Entity")]
+    PhysicalExpressionToStatisticJunction,
     #[sea_orm(has_many = "super::statistic_to_attribute_junction::Entity")]
     StatisticToAttributeJunction,
     #[sea_orm(
@@ -37,11 +29,13 @@ pub enum Relation {
         on_delete = "Cascade"
     )]
     TableMetadata,
+    #[sea_orm(has_many = "super::versioned_statistic::Entity")]
+    VersionedStatistic,
 }
 
-impl Related<super::event::Entity> for Entity {
+impl Related<super::physical_expression_to_statistic_junction::Entity> for Entity {
     fn to() -> RelationDef {
-        Relation::Event.def()
+        Relation::PhysicalExpressionToStatisticJunction.def()
     }
 }
 
@@ -57,6 +51,12 @@ impl Related<super::table_metadata::Entity> for Entity {
     }
 }
 
+impl Related<super::versioned_statistic::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::VersionedStatistic.def()
+    }
+}
+
 impl Related<super::attribute::Entity> for Entity {
     fn to() -> RelationDef {
         super::statistic_to_attribute_junction::Relation::Attribute.def()
@@ -70,4 +70,17 @@ impl Related<super::attribute::Entity> for Entity {
     }
 }
 
+impl Related<super::physical_expression::Entity> for Entity {
+    fn to() -> RelationDef {
+        super::physical_expression_to_statistic_junction::Relation::PhysicalExpression.def()
+    }
+    fn via() -> Option<RelationDef> {
+        Some(
+            super::physical_expression_to_statistic_junction::Relation::Statistic
+                .def()
+                .rev(),
+        )
+    }
+}
+
 impl ActiveModelBehavior for ActiveModel {}

optd-persistent/src/entities/versioned_statistic.rs

@@ -0,0 +1,47 @@
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
+
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "versioned_statistic")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub epoch_id: i32,
+    pub statistic_id: i32,
+    pub statistic_value: Json,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::event::Entity",
+        from = "Column::EpochId",
+        to = "super::event::Column::EpochId",
+        on_update = "Cascade",
+        on_delete = "Cascade"
+    )]
+    Event,
+    #[sea_orm(
+        belongs_to = "super::statistic::Entity",
+        from = "Column::StatisticId",
+        to = "super::statistic::Column::Id",
+        on_update = "Cascade",
+        on_delete = "Cascade"
+    )]
+    Statistic,
+}
+
+impl Related<super::event::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Event.def()
+    }
+}
+
+impl Related<super::statistic::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Statistic.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

optd-persistent/src/migrator/cost_model/m20241029_000001_event.rs

@@ -1,3 +1,6 @@
+//! Every time we insert/update statistics, we need to insert a new
+//! row into this table to record the event.
+
 use sea_orm_migration::{prelude::*, schema::*};
 
 #[derive(Iden)]

optd-persistent/src/migrator/cost_model/m20241029_000001_physical_expression_to_statistic_junction.rs

@@ -0,0 +1,75 @@
+//! This table stores for a physical expression, which statistics are used, so we
+//! don't need to compute it again. It is especially useful for update_stats, where
+//! we need to invalidate all the costs based on the physical_expression_id, so we
+//! need to use this table to get the physical_expression_id via statistic_id.
+//!
+//! **NOTE:** When we compute the cost for a physical expression, we should also
+//! insert related mappings into this table.
+
+use crate::migrator::cost_model::statistic::Statistic;
+use crate::migrator::memo::physical_expression::PhysicalExpression;
+
+use sea_orm_migration::{prelude::*, schema::*};
+
+#[derive(Iden)]
+pub enum PhysicalExpressionToStatisticJunction {
+    Table,
+    PhysicalExpressionId,
+    StatisticId,
+}
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(PhysicalExpressionToStatisticJunction::Table)
+                    .if_not_exists()
+                    .col(integer(
+                        PhysicalExpressionToStatisticJunction::PhysicalExpressionId,
+                    ))
+                    .col(integer(PhysicalExpressionToStatisticJunction::StatisticId))
+                    .primary_key(
+                        Index::create()
+                            .col(PhysicalExpressionToStatisticJunction::PhysicalExpressionId)
+                            .col(PhysicalExpressionToStatisticJunction::StatisticId),
+                    )
+                    .foreign_key(
+                        ForeignKey::create()
+                            .from(
+                                PhysicalExpressionToStatisticJunction::Table,
+                                PhysicalExpressionToStatisticJunction::PhysicalExpressionId,
+                            )
+                            .to(PhysicalExpression::Table, PhysicalExpression::Id)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .on_update(ForeignKeyAction::Cascade),
+                    )
+                    .foreign_key(
+                        ForeignKey::create()
+                            .from(
+                                PhysicalExpressionToStatisticJunction::Table,
+                                PhysicalExpressionToStatisticJunction::StatisticId,
+                            )
+                            .to(Statistic::Table, Statistic::Id)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .on_update(ForeignKeyAction::Cascade),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(
+                Table::drop()
+                    .table(PhysicalExpressionToStatisticJunction::Table)
+                    .to_owned(),
+            )
+            .await
+    }
+}

optd-persistent/src/migrator/cost_model/m20241029_000001_plan_cost.rs

@@ -1,14 +1,6 @@
-/*
-Table plan_cost {
-  id integer PK
-  physical_expression_id integer [ref: > physical_expression.id]
-  epoch_id integer [ref: > event.epoch_id]
-  cost integer
-  // Whether the cost is valid or not. If the latest cost for an expr is invalid, then we need to recompute the cost.
-  // We need to invalidate the cost when the related stats are updated.
-  is_valid boolean
-}
-*/
+//! When a statistic is updated, then all the related costs should be invalidated. (IsValid is set to false)
+//! This design (using IsValid flag) is based on the assumption that update_stats will not be called very frequently.
+//! It favors the compute_cost performance over the update_stats performance.
 
 use crate::migrator::cost_model::event::Event;
 use crate::migrator::memo::physical_expression::PhysicalExpression;
@@ -21,6 +13,8 @@ pub enum PlanCost {
     PhysicalExpressionId,
     EpochId,
     Cost,
+    // Whether the cost is valid or not. If the latest cost for an expr is invalid, then we need to recompute the cost.
+    // We need to invalidate the cost when the related stats are updated.
     IsValid,
 }