Skip to content

Commit

Permalink
feat(streaming): introduce streaming AsOf JOIN executor (#18242)
Browse files Browse the repository at this point in the history
  • Loading branch information
yuhao-su authored Sep 13, 2024
1 parent 9533cd9 commit bc9c2ad
Show file tree
Hide file tree
Showing 10 changed files with 2,193 additions and 28 deletions.
23 changes: 23 additions & 0 deletions proto/plan_common.proto
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
26 changes: 26 additions & 0 deletions proto/stream_plan.proto
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading

0 comments on commit bc9c2ad

Please sign in to comment.