feat: support asof join

src/query/ast/src/ast/format/syntax/query.rs

@@ -419,6 +419,7 @@ pub(crate) fn pretty_table(table: TableReference) -> RcDoc<'static> {
                 JoinOperator::RightAnti => RcDoc::text("RIGHT ANTI JOIN"),
                 JoinOperator::LeftSemi => RcDoc::text("LEFT SEMI JOIN"),
                 JoinOperator::RightSemi => RcDoc::text("RIGHT SEMI JOIN"),
+                JoinOperator::AsofJoin => RcDoc::text("ASOF JOIN"),
             })
             .append(RcDoc::space().append(pretty_table(*join.right)))
             .append(match &join.condition {

src/query/ast/src/ast/query.rs

@@ -800,6 +800,9 @@ impl Display for TableReference {
                     JoinOperator::CrossJoin => {
                         write!(f, " CROSS JOIN")?;
                     }
+                    JoinOperator::AsofJoin => {
+                        write!(f, " ASOF JOIN")?;
+                    }
                 }
                 write!(f, " {}", join.right)?;
                 match &join.condition {
@@ -872,6 +875,7 @@ pub enum JoinOperator {
     RightAnti,
     // CrossJoin can only work with `JoinCondition::None`
     CrossJoin,
+    AsofJoin,
 }
 
 #[derive(Debug, Clone, PartialEq, Drive, DriveMut)]

src/query/ast/src/parser/query.rs

@@ -623,6 +623,7 @@ pub fn join_operator(i: Input) -> IResult<JoinOperator> {
         value(JoinOperator::RightOuter, rule! { RIGHT ~ OUTER? }),
         value(JoinOperator::FullOuter, rule! { FULL ~ OUTER? }),
         value(JoinOperator::CrossJoin, rule! { CROSS }),
+        value(JoinOperator::AsofJoin, rule! { ASOF }),
     ))(i)
 }
 

src/query/ast/src/parser/token.rs

@@ -357,6 +357,8 @@ pub enum TokenKind {
     ARRAY,
     #[token("AS", ignore(ascii_case))]
     AS,
+    #[token("ASOF", ignore(ascii_case))]
+    ASOF,
     #[token("AST", ignore(ascii_case))]
     AST,
     #[token("AT", ignore(ascii_case))]
@@ -1590,6 +1592,7 @@ impl TokenKind {
             | TokenKind::WHEN => true,
             | TokenKind::ARRAY
             | TokenKind::AS
+            | TokenKind::ASOF
             | TokenKind::BETWEEN
             | TokenKind::CREATE
             | TokenKind::ATTACH

src/query/expression/src/types.rs

@@ -34,6 +34,7 @@ pub mod variant;
 
 use std::cmp::Ordering;
 use std::fmt::Debug;
+use std::i32::MAX;
 use std::ops::Range;
 
 use databend_common_arrow::arrow::trusted_len::TrustedLen;
@@ -65,6 +66,10 @@ use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
 use crate::ScalarRef;
+use crate::types::timestamp::TIMESTAMP_MAX;
+use crate::types::timestamp::TIMESTAMP_MIN;
+use crate::types::date::DATE_MAX;
+use crate::types::date::DATE_MIN;
 
 pub type GenericMap = [DataType];
 
@@ -173,6 +178,65 @@ impl DataType {
         }
     }
 
+    pub fn infinity(&self) -> Scalar {
+        match &self {
+             DataType::Timestamp => {
+                return  Scalar::Timestamp(TIMESTAMP_MAX)
+             },
+             DataType::Date => {
+                return  Scalar::Date(DATE_MAX)
+             }
+             DataType::Number(NumberDataType::Float32) =>{
+                return Scalar::Number(NumberScalar::Float32(ordered_float::OrderedFloat(f32::INFINITY)))
+             }
+             DataType::Number(NumberDataType::Int32) => {
+                return Scalar::Number(NumberScalar::Int32(MAX))
+             }
+             DataType::Number(NumberDataType::Int16) => {
+                return Scalar::Number(NumberScalar::Int16(MAX.try_into().unwrap()))
+             }
+             DataType::Number(NumberDataType::Int8) => {
+                return Scalar::Number(NumberScalar::Int8(MAX.try_into().unwrap()))
+             }
+             DataType::Number(NumberDataType::Float64) => {
+                return Scalar::Number(NumberScalar::Float64(ordered_float::OrderedFloat(f64::INFINITY)))
+             }
+             DataType::Number(NumberDataType::Int64) => {
+                return Scalar::Number(NumberScalar::Int64(MAX.into()))
+             }
+             _ => todo!()
+        }
+    }
+    pub fn ninfinity(&self) -> Scalar {
+        match &self {
+             DataType::Timestamp => {
+                return  Scalar::Timestamp(TIMESTAMP_MIN)
+             },
+             DataType::Date => {
+                return  Scalar::Date(DATE_MIN)
+             }
+             DataType::Number(NumberDataType::Float32) =>{
+                return Scalar::Number(NumberScalar::Float32(ordered_float::OrderedFloat(f32::NEG_INFINITY)))
+             }
+             DataType::Number(NumberDataType::Int32) => {
+                return Scalar::Number(NumberScalar::Int32(-MAX))
+             }
+             DataType::Number(NumberDataType::Int16) => {
+                return Scalar::Number(NumberScalar::Int16((-MAX).try_into().unwrap()))
+             }
+             DataType::Number(NumberDataType::Int8) => {
+                return Scalar::Number(NumberScalar::Int8((-MAX).try_into().unwrap()))
+             }
+             DataType::Number(NumberDataType::Float64) => {
+                return Scalar::Number(NumberScalar::Float64(ordered_float::OrderedFloat(f64::NEG_INFINITY)))
+             }
+             DataType::Number(NumberDataType::Int64) => {
+                return Scalar::Number(NumberScalar::Int64((-MAX).into()))
+             }
+             _ => todo!()
+        }
+    }
+
     pub fn is_unsigned_numeric(&self) -> bool {
         match self {
             DataType::Number(ty) => ALL_UNSIGNED_INTEGER_TYPES.contains(ty),

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_probe_state.rs

@@ -159,6 +159,7 @@ impl HashJoinProbeState {
     pub fn probe(&self, input: DataBlock, probe_state: &mut ProbeState) -> Result<Vec<DataBlock>> {
         match self.hash_join_state.hash_join_desc.join_type {
             JoinType::Inner
+            | JoinType::AsOf
             | JoinType::LeftSemi
             | JoinType::LeftAnti
             | JoinType::RightSemi

src/query/sql/src/executor/physical_plans/mod.rs

@@ -96,3 +96,5 @@ mod physical_udf;
 pub use physical_copy_into_location::CopyIntoLocation;
 pub use physical_udf::Udf;
 pub use physical_udf::UdfFunctionDesc;
+mod physical_asof_join;
+pub use physical_asof_join::*;