feat(long window): support *_where in sql engine

hybridse/examples/toydb/src/tablet/tablet_catalog.h

@@ -221,21 +221,18 @@ class TabletCatalog : public vm::Catalog {
 
     bool AddTable(std::shared_ptr<TabletTableHandler> table);
 
-    std::shared_ptr<type::Database> GetDatabase(const std::string& db);
+    std::shared_ptr<type::Database> GetDatabase(const std::string& db) override;
+
+    std::shared_ptr<vm::TableHandler> GetTable(const std::string& db, const std::string& table_name) override;
 
-    std::shared_ptr<vm::TableHandler> GetTable(const std::string& db,
-                                               const std::string& table_name);
     bool IndexSupport() override;
 
-    std::vector<vm::AggrTableInfo> GetAggrTables(
-        const std::string& base_db,
-        const std::string& base_table,
-        const std::string& aggr_func,
-        const std::string& aggr_col,
-        const std::string& partition_cols,
-        const std::string& order_col) override {
-        vm::AggrTableInfo info = {"aggr_" + base_table, "aggr_db", base_db, base_table,
-                                              aggr_func, aggr_col, partition_cols, order_col, "1000"};
+    std::vector<vm::AggrTableInfo> GetAggrTables(const std::string& base_db, const std::string& base_table,
+                                                 const std::string& aggr_func, const std::string& aggr_col,
+                                                 const std::string& partition_cols, const std::string& order_col,
+                                                 const std::string& filter_col) override {
+        vm::AggrTableInfo info = {"aggr_" + base_table, "aggr_db", base_db, base_table, aggr_func, aggr_col,
+                                  partition_cols,       order_col, "1000",  filter_col};
         return {info};
     }
 

hybridse/include/node/sql_node.h

@@ -24,6 +24,7 @@
 #include <unordered_map>
 #include <vector>
 
+#include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
 #include "boost/algorithm/string.hpp"
@@ -1074,6 +1075,33 @@ class ConstNode : public ExprNode {
         }
     }
 
+    // include 'udf/literal_traits.h' for Nullable lead to recursive include
+    // so `optional` is used for nullable info
+    template <typename T>
+    absl::StatusOr<std::optional<T>> GetAs() const {
+        if (IsNull()) {
+            return std::nullopt;
+        }
+
+        if constexpr (std::is_same_v<T, bool>) {
+            return GetBool();
+        } else if constexpr(std::is_same_v<T, int16_t>) {
+            return GetAsInt16();
+        } else if constexpr (std::is_same_v<T, int32_t>) {
+            return GetAsInt32();
+        } else if constexpr (std::is_same_v<T, int64_t>) {
+            return GetAsInt64();
+        } else if constexpr (std::is_same_v<T, float>) {
+            return GetAsFloat();
+        } else if constexpr (std::is_same_v<T, double>) {
+            return GetAsDouble();
+        } else if constexpr (std::is_same_v<T, std::string>) {
+            return GetAsString();
+        } else {
+            return absl::InvalidArgumentError("can't cast as T");
+        }
+    }
+
     Status InferAttr(ExprAnalysisContext *ctx) override;
     static ConstNode *CastFrom(ExprNode *node);
 
@@ -1618,7 +1646,7 @@ class ColumnRefNode : public ExprNode {
 
     void SetRelationName(const std::string &relation_name) { relation_name_ = relation_name; }
 
-    std::string GetColumnName() const { return column_name_; }
+    const std::string &GetColumnName() const { return column_name_; }
 
     void SetColumnName(const std::string &column_name) { column_name_ = column_name; }
 

hybridse/include/vm/catalog.h

@@ -471,6 +471,7 @@ struct AggrTableInfo {
     std::string partition_cols;
     std::string order_by_col;
     std::string bucket_size;
+    std::string filter_col;
 
     bool operator==(const AggrTableInfo& rhs) const {
         return aggr_table == rhs.aggr_table &&
@@ -481,7 +482,8 @@ struct AggrTableInfo {
             aggr_col == rhs.aggr_col &&
             partition_cols == rhs.partition_cols &&
             order_by_col == rhs.order_by_col &&
-            bucket_size == rhs.bucket_size;
+            bucket_size == rhs.bucket_size &&
+            filter_col == rhs.filter_col;
     }
 };
 
@@ -514,13 +516,10 @@ class Catalog {
         return nullptr;
     }
 
-    virtual std::vector<AggrTableInfo> GetAggrTables(
-        const std::string& base_db,
-        const std::string& base_table,
-        const std::string& aggr_func,
-        const std::string& aggr_col,
-        const std::string& partition_cols,
-        const std::string& order_col) {
+    virtual std::vector<AggrTableInfo> GetAggrTables(const std::string& base_db, const std::string& base_table,
+                                                     const std::string& aggr_func, const std::string& aggr_col,
+                                                     const std::string& partition_cols, const std::string& order_col,
+                                                     const std::string& filter_col) {
         return std::vector<AggrTableInfo>();
     }
 };

hybridse/include/vm/physical_op.h

@@ -785,7 +785,7 @@ class PhysicalReduceAggregationNode : public PhysicalProjectNode {
     }
     virtual ~PhysicalReduceAggregationNode() {}
     base::Status InitSchema(PhysicalPlanContext *) override;
-    virtual void Print(std::ostream &output, const std::string &tab) const;
+    void Print(std::ostream &output, const std::string &tab) const override;
     ConditionFilter having_condition_;
     const PhysicalAggregationNode* orig_aggr_ = nullptr;
 };
@@ -1500,26 +1500,25 @@ class PhysicalRequestAggUnionNode : public PhysicalOpNode {
     PhysicalRequestAggUnionNode(PhysicalOpNode *request, PhysicalOpNode *raw, PhysicalOpNode *aggr,
                                 const RequestWindowOp &window, const RequestWindowOp &aggr_window,
                                 bool instance_not_in_window, bool exclude_current_time, bool output_request_row,
-                                const node::FnDefNode *func, const node::ExprNode* agg_col)
+                                const node::CallExprNode *project)
         : PhysicalOpNode(kPhysicalOpRequestAggUnion, true),
           window_(window),
           agg_window_(aggr_window),
-          func_(func),
-          agg_col_(agg_col),
+          project_(project),
           instance_not_in_window_(instance_not_in_window),
           exclude_current_time_(exclude_current_time),
           output_request_row_(output_request_row) {
         output_type_ = kSchemaTypeTable;
 
-        fn_infos_.push_back(&window_.partition_.fn_info());
-        fn_infos_.push_back(&window_.sort_.fn_info());
-        fn_infos_.push_back(&window_.range_.fn_info());
-        fn_infos_.push_back(&window_.index_key_.fn_info());
+        AddFnInfo(&window_.partition_.fn_info());
+        AddFnInfo(&window_.sort_.fn_info());
+        AddFnInfo(&window_.range_.fn_info());
+        AddFnInfo(&window_.index_key_.fn_info());
 
-        fn_infos_.push_back(&agg_window_.partition_.fn_info());
-        fn_infos_.push_back(&agg_window_.sort_.fn_info());
-        fn_infos_.push_back(&agg_window_.range_.fn_info());
-        fn_infos_.push_back(&agg_window_.index_key_.fn_info());
+        AddFnInfo(&agg_window_.partition_.fn_info());
+        AddFnInfo(&agg_window_.sort_.fn_info());
+        AddFnInfo(&agg_window_.range_.fn_info());
+        AddFnInfo(&agg_window_.index_key_.fn_info());
 
         AddProducers(request, raw, aggr);
     }
@@ -1547,11 +1546,18 @@ class PhysicalRequestAggUnionNode : public PhysicalOpNode {
 
     RequestWindowOp window_;
     RequestWindowOp agg_window_;
-    const node::FnDefNode* func_ = nullptr;
-    const node::ExprNode* agg_col_;
+
+    // for long window, each node has only one projection node
+    const node::CallExprNode* project_;
     const SchemasContext* parent_schema_context_ = nullptr;
 
  private:
+    void AddProducers(PhysicalOpNode *request, PhysicalOpNode *raw, PhysicalOpNode *aggr) {
+        AddProducer(request);
+        AddProducer(raw);
+        AddProducer(aggr);
+    }
+
     const bool instance_not_in_window_;
     const bool exclude_current_time_;
 
@@ -1563,12 +1569,6 @@ class PhysicalRequestAggUnionNode : public PhysicalOpNode {
     // `EXCLUDE CURRENT_ROW`
     bool output_request_row_;
 
-    void AddProducers(PhysicalOpNode *request, PhysicalOpNode *raw, PhysicalOpNode *aggr) {
-        AddProducer(request);
-        AddProducer(raw);
-        AddProducer(aggr);
-    }
-
     Schema agg_schema_;
 };
 

hybridse/include/vm/simple_catalog.h

@@ -98,13 +98,10 @@ class SimpleCatalog : public Catalog {
     bool InsertRows(const std::string &db, const std::string &table,
                     const std::vector<Row> &row);
 
-    std::vector<AggrTableInfo> GetAggrTables(
-        const std::string& base_db,
-        const std::string& base_table,
-        const std::string& aggr_func,
-        const std::string& aggr_col,
-        const std::string& partition_cols,
-        const std::string& order_col) override;
+    std::vector<AggrTableInfo> GetAggrTables(const std::string &base_db, const std::string &base_table,
+                                             const std::string &aggr_func, const std::string &aggr_col,
+                                             const std::string &partition_cols, const std::string &order_col,
+                                             const std::string &filter_col) override;
 
  private:
     bool enable_index_;