Merge branch 'main' into refactor/functional-scalar-function-binder

src/common/src/lib.rs

@@ -13,11 +13,9 @@
 // limitations under the License.
 
 #![allow(rustdoc::private_intra_doc_links)]
-#![allow(clippy::derive_partial_eq_without_eq)]
 #![feature(trait_alias)]
 #![feature(binary_heap_drain_sorted)]
 #![feature(is_sorted)]
-#![feature(fn_traits)]
 #![feature(type_alias_impl_trait)]
 #![feature(test)]
 #![feature(trusted_len)]

src/common/src/types/mod.rs

@@ -28,6 +28,7 @@ use crate::error::BoxedError;
 mod native_type;
 mod ops;
 mod scalar_impl;
+mod successor;
 
 use std::fmt::Debug;
 use std::io::Cursor;
@@ -37,6 +38,7 @@ pub use native_type::*;
 use risingwave_pb::data::data_type::IntervalType::*;
 use risingwave_pb::data::data_type::{IntervalType, TypeName};
 pub use scalar_impl::*;
+pub use successor::*;
 pub mod chrono_wrapper;
 pub mod decimal;
 pub mod interval;

src/common/src/types/successor.rs

@@ -0,0 +1,84 @@
+// Copyright 2023 Singularity Data
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use chrono::Duration;
+
+use super::{NaiveDateTimeWrapper, NaiveDateWrapper, ScalarImpl};
+
+/// A successor is a term that comes right after a particular value. Suppose n is a number (where n
+/// belongs to any whole number), then the successor of n is 'n+1'. The other terminologies used for
+/// a successor are just after, immediately after, and next value.
+pub trait Successor {
+    /// Returns the successor of the current value if it exists, otherwise returns None.
+    fn successor(&self) -> Option<Self>
+    where
+        Self: Sized,
+    {
+        None
+    }
+}
+
+impl Successor for i16 {
+    fn successor(&self) -> Option<Self> {
+        self.checked_add(1)
+    }
+}
+
+impl Successor for i32 {
+    fn successor(&self) -> Option<Self> {
+        self.checked_add(1)
+    }
+}
+
+impl Successor for i64 {
+    fn successor(&self) -> Option<Self> {
+        self.checked_add(1)
+    }
+}
+
+impl Successor for NaiveDateTimeWrapper {
+    fn successor(&self) -> Option<Self> {
+        self.0
+            .checked_add_signed(Duration::nanoseconds(1))
+            .map(NaiveDateTimeWrapper)
+    }
+}
+
+impl Successor for NaiveDateWrapper {
+    fn successor(&self) -> Option<Self> {
+        self.0
+            .checked_add_signed(Duration::days(1))
+            .map(NaiveDateWrapper)
+    }
+}
+
+impl ScalarImpl {
+    /// Returns the successor of the current value if it exists.
+    ///
+    /// See also [`Successor`].
+    ///
+    /// The function may return None when:
+    /// 1. The current value is the maximum value of the type.
+    /// 2. The successor value of the type is not well-defined.
+    pub fn successor(&self) -> Option<Self> {
+        match self {
+            ScalarImpl::Int16(v) => v.successor().map(ScalarImpl::Int16),
+            ScalarImpl::Int32(v) => v.successor().map(ScalarImpl::Int32),
+            ScalarImpl::Int64(v) => v.successor().map(ScalarImpl::Int64),
+            ScalarImpl::NaiveDateTime(v) => v.successor().map(ScalarImpl::NaiveDateTime),
+            ScalarImpl::NaiveDate(v) => v.successor().map(ScalarImpl::NaiveDate),
+            _ => None,
+        }
+    }
+}

src/frontend/planner_test/tests/testdata/subquery.yaml

@@ -106,11 +106,11 @@
       ) as t0
     );
   optimized_logical_plan: |
-    LogicalJoin { type: LeftSemi, on: IsNotDistinctFrom(ab.a, bc.b), output: all }
+    LogicalJoin { type: LeftSemi, on: IsNotDistinctFrom(ab.a, t.v1), output: all }
     ├─LogicalScan { table: ab, columns: [ab.a, ab.b] }
     └─LogicalJoin { type: Inner, on: true, output: all }
-      ├─LogicalScan { table: bc, columns: [bc.b] }
-      └─LogicalScan { table: t, output_columns: [], required_columns: [t.v1], predicate: IsNotNull(t.v1) }
+      ├─LogicalScan { table: bc, columns: [] }
+      └─LogicalScan { table: t, columns: [t.v1], predicate: IsNotNull(t.v1) }
 - name: We cannot reference columns in left table if not lateral
   sql: |
     create table ab (a int, b int);
@@ -322,21 +322,20 @@
                           └─LogicalFilter { predicate: (b.b1 = CorrelatedInputRef { index: 0, correlated_id: 1 }) }
                             └─LogicalScan { table: b, columns: [b.b1, b.b2, b._row_id] }
   optimized_logical_plan: |
-    LogicalJoin { type: Inner, on: IsNotDistinctFrom(a.a1, a.a1) AND (a.a1 = min(b.b2)), output: [a.a1, a.a2] }
+    LogicalJoin { type: Inner, on: IsNotDistinctFrom(a.a1, a.a1) AND (a.a1 = min(b.b1)), output: [a.a1, a.a2] }
     ├─LogicalScan { table: a, columns: [a.a1, a.a2] }
-    └─LogicalAgg { group_key: [a.a1], aggs: [min(b.b2)] }
-      └─LogicalJoin { type: LeftOuter, on: IsNotDistinctFrom(a.a1, b.b1), output: [a.a1, b.b2] }
+    └─LogicalAgg { group_key: [a.a1], aggs: [min(b.b1)] }
+      └─LogicalJoin { type: LeftOuter, on: IsNotDistinctFrom(a.a1, a.a1), output: [a.a1, b.b1] }
         ├─LogicalAgg { group_key: [a.a1], aggs: [] }
         | └─LogicalScan { table: a, columns: [a.a1] }
-        └─LogicalJoin { type: Inner, on: (b.b2 = min(b.b1)), output: [b.b1, b.b2] }
+        └─LogicalJoin { type: Inner, on: (b.b2 = min(b.b1)), output: [a.a1, b.b1] }
           ├─LogicalScan { table: b, columns: [b.b1, b.b2] }
-          └─LogicalProject { exprs: [min(b.b1)] }
-            └─LogicalAgg { group_key: [a.a1], aggs: [min(b.b1)] }
-              └─LogicalJoin { type: LeftOuter, on: IsNotDistinctFrom(a.a1, b.b1), output: [a.a1, b.b1] }
-                ├─LogicalAgg { group_key: [a.a1], aggs: [] }
-                | └─LogicalScan { table: a, columns: [a.a1] }
-                └─LogicalProject { exprs: [b.b1, b.b1] }
-                  └─LogicalScan { table: b, columns: [b.b1], predicate: IsNotNull(b.b1) }
+          └─LogicalAgg { group_key: [a.a1], aggs: [min(b.b1)] }
+            └─LogicalJoin { type: LeftOuter, on: IsNotDistinctFrom(a.a1, b.b1), output: [a.a1, b.b1] }
+              ├─LogicalAgg { group_key: [a.a1], aggs: [] }
+              | └─LogicalScan { table: a, columns: [a.a1] }
+              └─LogicalProject { exprs: [b.b1, b.b1] }
+                └─LogicalScan { table: b, columns: [b.b1], predicate: IsNotNull(b.b1) }
 - name: test subquery in join on condition
   sql: |
     create table a (v1 int, v2 int);

src/frontend/planner_test/tests/testdata/subquery_expr_correlated.yaml

@@ -624,9 +624,9 @@
     create table t3(x int, y int);
     select * from t1 where exists(select t2.x from t2 right join t3 on t2.x = t3.x and t1.y = t2.y and t1.y = t3.y);
   optimized_logical_plan: |
-    LogicalJoin { type: LeftSemi, on: IsNotDistinctFrom(t1.y, t1.y), output: all }
+    LogicalJoin { type: LeftSemi, on: IsNotDistinctFrom(t1.y, t2.y), output: all }
     ├─LogicalScan { table: t1, columns: [t1.x, t1.y] }
-    └─LogicalJoin { type: LeftOuter, on: (t2.x = t3.x) AND (t2.y = t3.y) AND IsNotDistinctFrom(t2.y, t1.y), output: [t1.y] }
+    └─LogicalJoin { type: LeftOuter, on: (t2.x = t3.x) AND (t2.y = t3.y) AND IsNotDistinctFrom(t2.y, t1.y), output: [t2.y] }
       ├─LogicalJoin { type: Inner, on: true, output: all }
       | ├─LogicalAgg { group_key: [t1.y], aggs: [] }
       | | └─LogicalScan { table: t1, columns: [t1.y] }
@@ -737,3 +737,31 @@
             ├─LogicalAgg { group_key: [strings.v1], aggs: [] }
             | └─LogicalScan { table: strings, columns: [strings.v1] }
             └─LogicalScan { table: strings, columns: [strings.v1] }
+- name: Existential join on outer join with correlated condition
+  sql: |
+    create table t1(x int, y int);
+    create table t2(x int, y int);
+    create table t3(a varchar, z int);
+    select x from t1 where y in (select y from t3 full join t2 where t1.x = t2.x and z IS NOT DISTINCT FROM t2.x);
+  optimized_logical_plan: |
+    LogicalJoin { type: LeftSemi, on: (t1.y = t2.y) AND (t1.x = t2.x), output: [t1.x] }
+    ├─LogicalScan { table: t1, columns: [t1.x, t1.y] }
+    └─LogicalProject { exprs: [t2.y, t2.x] }
+      └─LogicalFilter { predicate: IsNotDistinctFrom(t3.z, t2.x) }
+        └─LogicalJoin { type: FullOuter, on: true, output: all }
+          ├─LogicalScan { table: t3, columns: [t3.z] }
+          └─LogicalScan { table: t2, columns: [t2.x, t2.y] }
+- name: Correlated condition in RHS of right outer join
+  sql: |
+    create table t1(x int, y int);
+    create table t2(x int, y int);
+    create table t3(a varchar, z int);
+    select x from t1 where y in (select y from t3 right join t2 where t1.x = t2.x and z IS NOT DISTINCT FROM t2.x);
+  optimized_logical_plan: |
+    LogicalJoin { type: LeftSemi, on: (t1.y = t2.y) AND (t1.x = t2.x), output: [t1.x] }
+    ├─LogicalScan { table: t1, columns: [t1.x, t1.y] }
+    └─LogicalProject { exprs: [t2.y, t2.x] }
+      └─LogicalFilter { predicate: IsNotDistinctFrom(t3.z, t2.x) }
+        └─LogicalJoin { type: LeftOuter, on: true, output: [t3.z, t2.x, t2.y] }
+          ├─LogicalScan { table: t2, columns: [t2.x, t2.y] }
+          └─LogicalScan { table: t3, columns: [t3.z] }