feat(streaming): introduce streaming AsOf JOIN executor

proto/plan_common.proto

@@ -141,6 +141,29 @@ enum JoinType {
   JOIN_TYPE_RIGHT_ANTI = 8;
 }
 
+enum AsOfJoinType {
+  AS_OF_JOIN_TYPE_UNSPECIFIED = 0;
+  AS_OF_JOIN_TYPE_INNER = 1;
+  AS_OF_JOIN_TYPE_LEFT_OUTER = 2;
+}
+
+enum AsOfJoinInequalityType {
+  AS_OF_INEQUALITY_TYPE_UNSPECIFIED = 0;
+  AS_OF_INEQUALITY_TYPE_GT = 1;
+  AS_OF_INEQUALITY_TYPE_GE = 2;
+  AS_OF_INEQUALITY_TYPE_LT = 3;
+  AS_OF_INEQUALITY_TYPE_LE = 4;
+}
+
+message AsOfJoinDesc {
+  // The index of the right side's as of column.
+  uint32 right_idx = 1;
+  // The index of the left side's as of column.
+  uint32 left_idx = 2;
+  // The type of the inequality.
+  AsOfJoinInequalityType inequality_type = 3;
+}
+
 // https://github.com/tokio-rs/prost/issues/80
 enum FormatType {
   FORMAT_TYPE_UNSPECIFIED = 0;

proto/stream_plan.proto

@@ -455,6 +455,32 @@ message HashJoinNode {
   bool is_append_only = 14;
 }
 
+message AsOfJoinNode {
+  plan_common.AsOfJoinType join_type = 1;
+  repeated int32 left_key = 2;
+  repeated int32 right_key = 3;
+  // Used for internal table states.
+  catalog.Table left_table = 4;
+  // Used for internal table states.
+  catalog.Table right_table = 5;
+  // Used for internal table states.
+  catalog.Table left_degree_table = 6;
+  // Used for internal table states.
+  catalog.Table right_degree_table = 7;
+  // The output indices of current node
+  repeated uint32 output_indices = 8;
+  // Left deduped input pk indices. The pk of the left_table and
+  // The pk of the left_table is  [left_join_key | left_inequality_key | left_deduped_input_pk_indices]
+  // left_inequality_key is not used but for forward compatibility.
+  repeated uint32 left_deduped_input_pk_indices = 9;
+  // Right deduped input pk indices.
+  // The pk of the right_table is  [right_join_key | right_inequality_key | right_deduped_input_pk_indices]
+  // right_inequality_key is not used but for forward compatibility.
+  repeated uint32 right_deduped_input_pk_indices = 10;
+  repeated bool null_safe = 11;
+  optional plan_common.AsOfJoinDesc asof_desc = 12;
+}
+
 message TemporalJoinNode {
   plan_common.JoinType join_type = 1;
   repeated int32 left_key = 2;