create and apply rule

risingwavelabs · Feb 4, 2025 · 5ad81a7 · 5ad81a7
1 parent 3cd5a9a
commit 5ad81a7
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/src/frontend/src/optimizer/mod.rs b/src/frontend/src/optimizer/mod.rs
@@ -514,6 +514,16 @@ impl PlanRoot {
             ))?;
         }
 
+        if ctx.session_ctx().config().streaming_unaligned_join() {
+            // TODO: make it a logical optimization.
+            // Rewrite joins with index to delta join
+            plan = plan.optimize_by_rules(&OptimizationStage::new(
+                "Add Logstore for Unaligned join",
+                vec![AddLogstoreRule::create()],
+                ApplyOrder::BottomUp,
+            ))?;
+        }
+
         // Inline session timezone
         plan = inline_session_timezone_in_exprs(ctx.clone(), plan)?;
 

diff --git a/src/frontend/src/optimizer/rule/add_logstore_rule.rs b/src/frontend/src/optimizer/rule/add_logstore_rule.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::optimizer::plan_node::StreamSyncLogStore;
-use crate::optimizer::rule::Rule;
+use crate::optimizer::rule::{BoxedRule, Rule};
 use crate::PlanRef;
 
 pub struct AddLogstoreRule {}
@@ -25,3 +25,9 @@ impl Rule for AddLogstoreRule {
         Some(log_store_plan.into())
     }
 }
+
+impl AddLogstoreRule {
+    pub fn create() -> BoxedRule {
+        Box::new(Self {})
+    }
+}