Merge branch 'main' into yuhao/handle_join_watermark

Author: mergify[bot]

src/compute/src/lib.rs

@@ -18,7 +18,7 @@
 #![feature(type_alias_impl_trait)]
 #![feature(let_chains)]
 #![feature(result_option_inspect)]
-#![feature(allocator_api)]
+#![feature(lint_reasons)]
 #![cfg_attr(coverage, feature(no_coverage))]
 
 #[macro_use]

src/compute/src/memory_management/memory_manager.rs

@@ -12,20 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
+#[cfg(target_os = "linux")]
 use std::time::Duration;
 
 use risingwave_batch::task::BatchManager;
+#[cfg(target_os = "linux")]
 use risingwave_common::util::epoch::Epoch;
 use risingwave_stream::executor::monitor::StreamingMetrics;
 use risingwave_stream::task::LocalStreamManager;
 #[cfg(target_os = "linux")]
 use tikv_jemalloc_ctl::{epoch as jemalloc_epoch, stats as jemalloc_stats};
+#[cfg(target_os = "linux")]
 use tracing;
 
 /// When `enable_managed_cache` is set, compute node will launch a [`GlobalMemoryManager`] to limit
 /// the memory usage.
+#[cfg_attr(not(target_os = "linux"), expect(dead_code))]
 pub struct GlobalMemoryManager {
     /// All cached data before the watermark should be evicted.
     watermark_epoch: Arc<AtomicU64>,
@@ -39,7 +43,9 @@ pub struct GlobalMemoryManager {
 pub type GlobalMemoryManagerRef = Arc<GlobalMemoryManager>;
 
 impl GlobalMemoryManager {
+    #[cfg(target_os = "linux")]
     const EVICTION_THRESHOLD_AGGRESSIVE: f64 = 0.9;
+    #[cfg(target_os = "linux")]
     const EVICTION_THRESHOLD_GRACEFUL: f64 = 0.7;
 
     pub fn new(
@@ -63,7 +69,10 @@ impl GlobalMemoryManager {
         self.watermark_epoch.clone()
     }
 
+    #[cfg(target_os = "linux")]
     fn set_watermark_time_ms(&self, time_ms: u64) {
+        use std::sync::atomic::Ordering;
+
         let epoch = Epoch::from_physical_time(time_ms).0;
         let watermark_epoch = self.watermark_epoch.as_ref();
         watermark_epoch.store(epoch, Ordering::Relaxed);

src/expr/src/vector_op/cast.rs

@@ -397,6 +397,40 @@ pub fn bool_out(input: bool, writer: &mut dyn Write) -> Result<()> {
     Ok(())
 }
 
+/// A lite version of casting from string to target type. Used by frontend to handle types that have
+/// to be created by casting.
+///
+/// For example, the user can input `1` or `true` directly, but they have to use
+/// `'2022-01-01'::date`.
+pub fn literal_parsing(
+    t: &DataType,
+    s: &str,
+) -> std::result::Result<ScalarImpl, Option<ExprError>> {
+    let scalar = match t {
+        DataType::Boolean => str_to_bool(s)?.into(),
+        DataType::Int16 => str_parse::<i16>(s)?.into(),
+        DataType::Int32 => str_parse::<i32>(s)?.into(),
+        DataType::Int64 => str_parse::<i64>(s)?.into(),
+        DataType::Decimal => str_parse::<Decimal>(s)?.into(),
+        DataType::Float32 => str_parse::<OrderedF32>(s)?.into(),
+        DataType::Float64 => str_parse::<OrderedF64>(s)?.into(),
+        DataType::Varchar => return Err(None),
+        DataType::Date => str_to_date(s)?.into(),
+        DataType::Timestamp => str_to_timestamp(s)?.into(),
+        // We only handle the case with timezone here, and leave the implicit session timezone case
+        // for later phase.
+        DataType::Timestamptz => str_with_time_zone_to_timestamptz(s)?.into(),
+        DataType::Time => str_to_time(s)?.into(),
+        DataType::Interval => str_parse::<IntervalUnit>(s)?.into(),
+        // Not processing list or struct literal right now. Leave it for later phase (normal backend
+        // evaluation).
+        DataType::List { .. } => return Err(None),
+        DataType::Struct(_) => return Err(None),
+        DataType::Bytea => str_to_bytea(s)?.into(),
+    };
+    Ok(scalar)
+}
+
 /// It accepts a macro whose input is `{ $input:ident, $cast:ident, $func:expr }` tuples
 ///
 /// * `$input`: input type

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

@@ -126,7 +126,7 @@
 - sql: |
     select * from generate_series('2'::INT,'10'::INT,'2'::INT);
   batch_plan: |
-    BatchTableFunction { Generate('2':Varchar::Int32, '10':Varchar::Int32, '2':Varchar::Int32) }
+    BatchTableFunction { Generate(2:Int32, 10:Int32, 2:Int32) }
 - sql: |
     select * from unnest(Array[1,2,3]);
   batch_plan: |

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

@@ -53,21 +53,19 @@
     select 1 having 'y';
   batch_plan: |
     BatchProject { exprs: [1:Int32] }
-    └─BatchFilter { predicate: 'y':Varchar::Boolean }
-      └─BatchSimpleAgg { aggs: [] }
-        └─BatchValues { rows: [[]] }
+    └─BatchSimpleAgg { aggs: [] }
+      └─BatchValues { rows: [[]] }
 - name: implicit cast boolean (WHERE with literal 'y' of unknown type)
   sql: |
     select 1 where 'y';
   batch_plan: |
     BatchProject { exprs: [1:Int32] }
-    └─BatchFilter { predicate: 'y':Varchar::Boolean }
-      └─BatchValues { rows: [[]] }
+    └─BatchValues { rows: [[]] }
 - name: implicit cast boolean (CASE with literal 'y' of unknown type)
   sql: |
     select case when 'y' then 1 end;
   batch_plan: |
-    BatchProject { exprs: [Case('y':Varchar::Boolean, 1:Int32)] }
+    BatchProject { exprs: [Case(true:Boolean, 1:Int32)] }
     └─BatchValues { rows: [[]] }
 - name: implicit cast boolean (JOIN ON with literal 'y' of unknown type)
   sql: |
@@ -77,23 +75,21 @@
   batch_plan: |
     BatchNestedLoopJoin { type: Inner, predicate: true, output: all }
     ├─BatchExchange { order: [], dist: Single }
-    | └─BatchFilter { predicate: 'y':Varchar::Boolean }
-    |   └─BatchScan { table: t1, columns: [t1.v1], distribution: SomeShard }
+    | └─BatchScan { table: t1, columns: [t1.v1], distribution: SomeShard }
     └─BatchExchange { order: [], dist: Single }
-      └─BatchFilter { predicate: 'y':Varchar::Boolean }
-        └─BatchScan { table: t2, columns: [], distribution: SomeShard }
+      └─BatchScan { table: t2, columns: [], distribution: SomeShard }
 - name: current_schemas (CURRENT_SCHEMAS with literal 'y' of unknown type)
   sql: |
     select current_schemas('y');
-  binder_error: |-
-    Feature is not yet implemented: Only boolean literals are supported in `current_schemas`.
-    No tracking issue yet. Feel free to submit a feature request at https://github.com/risingwavelabs/risingwave/issues/new?labels=type%2Ffeature&template=feature_request.yml
+  batch_plan: |
+    BatchProject { exprs: [ARRAY[pg_catalog, public]:List { datatype: Varchar }] }
+    └─BatchValues { rows: [[]] }
 - name: FILTER (FILTER with literal 'y' of unknown type)
   sql: |
     create table t(v1 int);
     select count(*) FILTER(WHERE 'y') from t;
   batch_plan: |
-    BatchSimpleAgg { aggs: [sum0(count filter('y':Varchar::Boolean))] }
+    BatchSimpleAgg { aggs: [sum0(count)] }
     └─BatchExchange { order: [], dist: Single }
-      └─BatchSimpleAgg { aggs: [count filter('y':Varchar::Boolean)] }
+      └─BatchSimpleAgg { aggs: [count] }
         └─BatchScan { table: t, columns: [], distribution: SomeShard }

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

@@ -3,13 +3,13 @@
   sql: |
     select int '1';
   logical_plan: |
-    LogicalProject { exprs: ['1':Varchar::Int32] }
+    LogicalProject { exprs: [1:Int32] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
 - name: bind typed literal - bool
   sql: |
     SELECT bool 't'
   logical_plan: |
-    LogicalProject { exprs: ['t':Varchar::Boolean] }
+    LogicalProject { exprs: [true:Boolean] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
 - sql: |
     values(must_be_unimplemented_func(1));
@@ -143,14 +143,17 @@
 - sql: |
     -- Single quoted literal can be treated as number without error.
     values(round('123'));
-    -- When it is invalid, PostgreSQL reports error during explain, but we have to wait until execution as of now.
-    -- values(round('abc'));
   batch_plan: |
-    BatchValues { rows: [[Round('123':Varchar::Float64)]] }
+    BatchValues { rows: [[Round(123:Float64)]] }
+- sql: |
+    -- When it is invalid, PostgreSQL reports error during explain, but we have to wait until execution as of now. #4235
+    values(round('abc'));
+  batch_plan: |
+    BatchValues { rows: [[Round('abc':Varchar::Float64)]] }
 - sql: |
     values(extract(hour from timestamp '2001-02-16 20:38:40'));
   batch_plan: |
-    BatchValues { rows: [[Extract('HOUR':Varchar, '2001-02-16 20:38:40':Varchar::Timestamp)]] }
+    BatchValues { rows: [[Extract('HOUR':Varchar, '2001-02-16 20:38:40':Timestamp)]] }
 - sql: |
     values('Postgres' not like 'Post%');
   batch_plan: |

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

@@ -14,7 +14,7 @@
   batch_plan: |
     BatchExchange { order: [], dist: Single }
     └─BatchInsert { table: t }
-      └─BatchValues { rows: [[22.33:Decimal::Float32, '33':Varchar::Int32], [44:Int32::Float32, 55.0:Decimal::Int32]] }
+      └─BatchValues { rows: [[22.33:Decimal::Float32, 33:Int32], [44:Int32::Float32, 55.0:Decimal::Int32]] }
 - name: insert values on non-assign-castable types
   sql: |
     create table t (v1 real, v2 int);
@@ -73,7 +73,7 @@
   batch_plan: |
     BatchExchange { order: [], dist: Single }
     └─BatchInsert { table: t }
-      └─BatchValues { rows: [['2020-01-01 01:02:03':Varchar::Timestamp::Time], ['03:04:05':Varchar::Time]] }
+      └─BatchValues { rows: [['2020-01-01 01:02:03':Timestamp::Time], ['03:04:05':Time]] }
 - name: a `VALUES` without insert context may be invalid on its own (compare with above)
   sql: |
     create table t (v1 time);
@@ -115,7 +115,7 @@
     BatchExchange { order: [], dist: Single }
     └─BatchInsert { table: t }
       └─BatchExchange { order: [], dist: Single }
-        └─BatchProject { exprs: ['2020-01-01 01:02:03':Varchar::Timestamp::Time, 11:Int32, 4.5:Decimal::Float32] }
+        └─BatchProject { exprs: ['2020-01-01 01:02:03':Timestamp::Time, 11:Int32, 4.5:Decimal::Float32] }
           └─BatchScan { table: t, columns: [], distribution: SomeShard }
 - name: insert into select with cast error
   sql: |

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

@@ -53,30 +53,30 @@
     AND window_start >= ts + interval '1' day AND window_end > ts + interval '4' day;
   logical_plan: |
     LogicalProject { exprs: [t.v1, t.v2, t.v3, t.v4, t.ts, window_start, window_end] }
-    └─LogicalFilter { predicate: (t.v1 = 10:Int32) AND (t.v2 = 20:Int32) AND (t.v3 = 30:Int32) AND (t.ts >= '1997-07-01':Varchar::Date) AND (window_start >= '1997-07-02':Varchar::Date) AND (window_end >= '1997-07-03':Varchar::Date) AND (window_start >= (t.ts + '1 day':Interval)) AND (window_end > (t.ts + '4 days':Interval)) }
+    └─LogicalFilter { predicate: (t.v1 = 10:Int32) AND (t.v2 = 20:Int32) AND (t.v3 = 30:Int32) AND (t.ts >= '1997-07-01':Date) AND (window_start >= '1997-07-02':Date) AND (window_end >= '1997-07-03':Date) AND (window_start >= (t.ts + '1 day':Interval)) AND (window_end > (t.ts + '4 days':Interval)) }
       └─LogicalShare { id = 4 }
         └─LogicalProject { exprs: [t.v1, t.v2, t.v3, t.v4, t.ts, window_start, window_end] }
           └─LogicalHopWindow { time_col: t.ts, slide: 1 day, size: 3 days, output: all }
             └─LogicalScan { table: t, columns: [t.v1, t.v2, t.v3, t.v4, t.ts, t._row_id] }
   optimized_logical_plan: |
-    LogicalFilter { predicate: (window_start >= '1997-07-02':Varchar::Date) AND (window_end >= '1997-07-03':Varchar::Date) AND (window_start >= (t.ts + '1 day':Interval)) AND (window_end > (t.ts + '4 days':Interval)) }
+    LogicalFilter { predicate: (window_start >= '1997-07-02':Date) AND (window_end >= '1997-07-03':Date) AND (window_start >= (t.ts + '1 day':Interval)) AND (window_end > (t.ts + '4 days':Interval)) }
     └─LogicalHopWindow { time_col: t.ts, slide: 1 day, size: 3 days, output: all }
-      └─LogicalScan { table: t, columns: [t.v1, t.v2, t.v3, t.v4, t.ts], predicate: (t.v1 = 10:Int32) AND (t.v2 = 20:Int32) AND (t.v3 = 30:Int32) AND (t.ts >= '1997-07-01':Varchar::Date) }
+      └─LogicalScan { table: t, columns: [t.v1, t.v2, t.v3, t.v4, t.ts], predicate: (t.v1 = 10:Int32) AND (t.v2 = 20:Int32) AND (t.v3 = 30:Int32) AND (t.ts >= '1997-07-01':Date) }
 - name: filter hop transpose with non-trivial output-indices
   sql: |
     create table t(v1 int, v2 int, v3 int, v4 int, ts date);
     with cte as (select window_end, v4, v2 from hop(t, ts, interval '1' day, interval '3' day))
     select * from cte where window_end > date '2022-01-01' AND v4=10 AND v2 > 20
   logical_plan: |
     LogicalProject { exprs: [window_end, t.v4, t.v2] }
-    └─LogicalFilter { predicate: (window_end > '2022-01-01':Varchar::Date) AND (t.v4 = 10:Int32) AND (t.v2 > 20:Int32) }
+    └─LogicalFilter { predicate: (window_end > '2022-01-01':Date) AND (t.v4 = 10:Int32) AND (t.v2 > 20:Int32) }
       └─LogicalShare { id = 4 }
         └─LogicalProject { exprs: [window_end, t.v4, t.v2] }
           └─LogicalHopWindow { time_col: t.ts, slide: 1 day, size: 3 days, output: all }
             └─LogicalScan { table: t, columns: [t.v1, t.v2, t.v3, t.v4, t.ts, t._row_id] }
   optimized_logical_plan: |
     LogicalProject { exprs: [window_end, t.v4, t.v2] }
-    └─LogicalFilter { predicate: (window_end > '2022-01-01':Varchar::Date) }
+    └─LogicalFilter { predicate: (window_end > '2022-01-01':Date) }
       └─LogicalHopWindow { time_col: t.ts, slide: 1 day, size: 3 days, output: [t.v2, t.v4, window_end] }
         └─LogicalScan { table: t, columns: [t.v2, t.v4, t.ts], predicate: (t.v4 = 10:Int32) AND (t.v2 > 20:Int32) }
 - name: filter union transpose
@@ -257,16 +257,16 @@
     select * from t1 join t2 where v1 = v2 and v1 > now() + '1 hr';
   optimized_logical_plan: |
     LogicalJoin { type: Inner, on: (t1.v1 = t2.v2), output: all }
-    ├─LogicalFilter { predicate: (t1.v1 > (Now + '1 hr':Varchar::Interval)) }
+    ├─LogicalFilter { predicate: (t1.v1 > (Now + '01:00:00':Interval)) }
     | └─LogicalScan { table: t1, columns: [t1.v1] }
     └─LogicalScan { table: t2, columns: [t2.v2] }
   stream_plan: |
     StreamMaterialize { columns: [v1, v2, t1._row_id(hidden), t2._row_id(hidden)], pk_columns: [t1._row_id, t2._row_id, v1, v2] }
     └─StreamHashJoin { type: Inner, predicate: t1.v1 = t2.v2, output: [t1.v1, t2.v2, t1._row_id, t2._row_id] }
       ├─StreamExchange { dist: HashShard(t1.v1) }
-      | └─StreamDynamicFilter { predicate: (t1.v1 > (now + '1 hr':Varchar::Interval)), output: [t1.v1, t1._row_id] }
+      | └─StreamDynamicFilter { predicate: (t1.v1 > (now + '01:00:00':Interval)), output: [t1.v1, t1._row_id] }
       |   ├─StreamTableScan { table: t1, columns: [t1.v1, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
-      |   └─StreamProject { exprs: [(now + '1 hr':Varchar::Interval)] }
+      |   └─StreamProject { exprs: [(now + '01:00:00':Interval)], watermark_columns: [(now + '01:00:00':Interval)] }
       |     └─StreamNow { output: [now] }
       └─StreamExchange { dist: HashShard(t2.v2) }
         └─StreamTableScan { table: t2, columns: [t2.v2, t2._row_id], pk: [t2._row_id], dist: UpstreamHashShard(t2._row_id) }
@@ -297,14 +297,14 @@
     create table t1(v1 timestamp with time zone, v2 int);
     select * from t1 where v1 > now() + '30 min' and v2 > 5;
   optimized_logical_plan: |
-    LogicalFilter { predicate: (t1.v1 > (Now + '30 min':Varchar::Interval)) }
+    LogicalFilter { predicate: (t1.v1 > (Now + '00:30:00':Interval)) }
     └─LogicalScan { table: t1, columns: [t1.v1, t1.v2], predicate: (t1.v2 > 5:Int32) }
   stream_plan: |
     StreamMaterialize { columns: [v1, v2, t1._row_id(hidden)], pk_columns: [t1._row_id] }
-    └─StreamDynamicFilter { predicate: (t1.v1 > (now + '30 min':Varchar::Interval)), output: [t1.v1, t1.v2, t1._row_id] }
+    └─StreamDynamicFilter { predicate: (t1.v1 > (now + '00:30:00':Interval)), output: [t1.v1, t1.v2, t1._row_id] }
       ├─StreamFilter { predicate: (t1.v2 > 5:Int32) }
       | └─StreamTableScan { table: t1, columns: [t1.v1, t1.v2, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
-      └─StreamProject { exprs: [(now + '30 min':Varchar::Interval)] }
+      └─StreamProject { exprs: [(now + '00:30:00':Interval)], watermark_columns: [(now + '00:30:00':Interval)] }
         └─StreamNow { output: [now] }
 - name: eq-predicate derived condition other side pushdown in inner join
   sql: |

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

@@ -2,8 +2,8 @@
 - sql: |
     select generate_series('2'::INT,'10'::INT,'2'::INT);
   batch_plan: |
-    BatchProject { exprs: [Generate('2':Varchar::Int32, '10':Varchar::Int32, '2':Varchar::Int32)] }
-    └─BatchProjectSet { select_list: [Generate('2':Varchar::Int32, '10':Varchar::Int32, '2':Varchar::Int32)] }
+    BatchProject { exprs: [Generate(2:Int32, 10:Int32, 2:Int32)] }
+    └─BatchProjectSet { select_list: [Generate(2:Int32, 10:Int32, 2:Int32)] }
       └─BatchValues { rows: [[]] }
 - sql: |
     select unnest(Array[1,2,3]);

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

@@ -190,7 +190,7 @@
     LogicalProject { exprs: [*VALUES*_0.column_0, TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval), (TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval) + '00:00:10':Interval)] }
     └─LogicalProject { exprs: [*VALUES*_0.column_0, TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval), (TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval) + '00:00:10':Interval)] }
       └─LogicalShare { id = 2 }
-        └─LogicalValues { rows: [['2020-01-01 12:00:00':Varchar::Timestamp]], schema: Schema { fields: [*VALUES*_0.column_0:Timestamp] } }
+        └─LogicalValues { rows: [['2020-01-01 12:00:00':Timestamp]], schema: Schema { fields: [*VALUES*_0.column_0:Timestamp] } }
   batch_plan: |
     BatchProject { exprs: [*VALUES*_0.column_0, TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval), (TumbleStart(*VALUES*_0.column_0, '00:00:10':Interval) + '00:00:10':Interval)] }
-    └─BatchValues { rows: [['2020-01-01 12:00:00':Varchar::Timestamp]] }
+    └─BatchValues { rows: [['2020-01-01 12:00:00':Timestamp]] }