Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/common/src/array/arrow/arrow_iceberg.rs b/src/common/src/array/arrow/arrow_iceberg.rs
index d34c2a1771eec..be6520cff94c5 100644
--- a/src/common/src/array/arrow/arrow_iceberg.rs
+++ b/src/common/src/array/arrow/arrow_iceberg.rs
@@ -44,6 +44,10 @@ impl IcebergArrowConvert {
         FromArrow::from_record_batch(self, batch)
     }
 
+    pub fn type_from_field(&self, field: &arrow_schema::Field) -> Result<DataType, ArrayError> {
+        FromArrow::from_field(self, field)
+    }
+
     pub fn to_arrow_field(
         &self,
         name: &str,
@@ -52,10 +56,6 @@ impl IcebergArrowConvert {
         ToArrow::to_arrow_field(self, name, data_type)
     }
 
-    pub fn type_from_field(&self, field: &arrow_schema::Field) -> Result<DataType, ArrayError> {
-        FromArrow::from_field(self, field)
-    }
-
     pub fn struct_from_fields(
         &self,
         fields: &arrow_schema::Fields,
@@ -81,6 +81,36 @@ impl IcebergArrowConvert {
 }
 
 impl ToArrow for IcebergArrowConvert {
+    fn to_arrow_field(
+        &self,
+        name: &str,
+        data_type: &DataType,
+    ) -> Result<arrow_schema::Field, ArrayError> {
+        let data_type = match data_type {
+            DataType::Boolean => self.bool_type_to_arrow(),
+            DataType::Int16 => self.int32_type_to_arrow(),
+            DataType::Int32 => self.int32_type_to_arrow(),
+            DataType::Int64 => self.int64_type_to_arrow(),
+            DataType::Int256 => self.int256_type_to_arrow(),
+            DataType::Float32 => self.float32_type_to_arrow(),
+            DataType::Float64 => self.float64_type_to_arrow(),
+            DataType::Date => self.date_type_to_arrow(),
+            DataType::Time => self.time_type_to_arrow(),
+            DataType::Timestamp => self.timestamp_type_to_arrow(),
+            DataType::Timestamptz => self.timestamptz_type_to_arrow(),
+            DataType::Interval => self.interval_type_to_arrow(),
+            DataType::Varchar => self.varchar_type_to_arrow(),
+            DataType::Bytea => self.bytea_type_to_arrow(),
+            DataType::Serial => self.serial_type_to_arrow(),
+            DataType::Decimal => return Ok(self.decimal_type_to_arrow(name)),
+            DataType::Jsonb => self.varchar_type_to_arrow(),
+            DataType::Struct(fields) => self.struct_type_to_arrow(fields)?,
+            DataType::List(datatype) => self.list_type_to_arrow(datatype)?,
+            DataType::Map(datatype) => self.map_type_to_arrow(datatype)?,
+        };
+        Ok(arrow_schema::Field::new(name, data_type, true))
+    }
+
     #[inline]
     fn decimal_type_to_arrow(&self, name: &str) -> arrow_schema::Field {
         // Fixed-point decimal; precision P, scale S Scale is fixed, precision must be less than 38.
@@ -207,10 +237,10 @@ impl ToArrow for IcebergCreateTableArrowConvert {
         let data_type = match value {
             // using the inline function
             DataType::Boolean => self.bool_type_to_arrow(),
-            DataType::Int16 => self.int16_type_to_arrow(),
+            DataType::Int16 => self.int32_type_to_arrow(),
             DataType::Int32 => self.int32_type_to_arrow(),
             DataType::Int64 => self.int64_type_to_arrow(),
-            DataType::Int256 => self.int256_type_to_arrow(),
+            DataType::Int256 => self.varchar_type_to_arrow(),
             DataType::Float32 => self.float32_type_to_arrow(),
             DataType::Float64 => self.float64_type_to_arrow(),
             DataType::Date => self.date_type_to_arrow(),
@@ -222,7 +252,7 @@ impl ToArrow for IcebergCreateTableArrowConvert {
             DataType::Bytea => self.bytea_type_to_arrow(),
             DataType::Serial => self.serial_type_to_arrow(),
             DataType::Decimal => return Ok(self.decimal_type_to_arrow(name)),
-            DataType::Jsonb => return Ok(self.jsonb_type_to_arrow(name)),
+            DataType::Jsonb => self.varchar_type_to_arrow(),
             DataType::Struct(fields) => self.struct_type_to_arrow(fields)?,
             DataType::List(datatype) => self.list_type_to_arrow(datatype)?,
             DataType::Map(datatype) => self.map_type_to_arrow(datatype)?,
diff --git a/src/common/src/catalog/mod.rs b/src/common/src/catalog/mod.rs
index 8ea077b5a731a..5556650bd328c 100644
--- a/src/common/src/catalog/mod.rs
+++ b/src/common/src/catalog/mod.rs
@@ -98,6 +98,7 @@ pub fn default_key_column_name_version_mapping(version: &ColumnDescVersion) -> &
 /// [this rfc](https://github.com/risingwavelabs/rfcs/pull/20).
 pub const KAFKA_TIMESTAMP_COLUMN_NAME: &str = "_rw_kafka_timestamp";
 
+pub const RISINGWAVE_ICEBERG_ROW_ID: &str = "_risingwave_iceberg_row_id";
 pub fn is_system_schema(schema_name: &str) -> bool {
     SYSTEM_SCHEMAS.iter().any(|s| *s == schema_name)
 }
diff --git a/src/expr/impl/src/scalar/cast.rs b/src/expr/impl/src/scalar/cast.rs
index 41c51d95445ec..e25ca07e3398f 100644
--- a/src/expr/impl/src/scalar/cast.rs
+++ b/src/expr/impl/src/scalar/cast.rs
@@ -86,6 +86,7 @@ pub fn jsonb_to_number<T: TryFrom<F64>>(v: JsonbRef<'_>) -> Result<T> {
 #[function("cast(int4) -> int2")]
 #[function("cast(int8) -> int2")]
 #[function("cast(int8) -> int4")]
+#[function("cast(int8) -> serial")]
 #[function("cast(serial) -> int8")]
 #[function("cast(float4) -> int2")]
 #[function("cast(float8) -> int2")]
diff --git a/src/frontend/src/expr/type_inference/cast.rs b/src/frontend/src/expr/type_inference/cast.rs
index e03ec98883ce9..cced3ffd2afce 100644
--- a/src/frontend/src/expr/type_inference/cast.rs
+++ b/src/frontend/src/expr/type_inference/cast.rs
@@ -309,7 +309,7 @@ pub static CAST_TABLE: LazyLock<CastTable> = LazyLock::new(|| {
         (". e            a ", Boolean),     // 0
         (" .iiiiii       a ", Int16),       // 1
         ("ea.iiiii       a ", Int32),       // 2
-        (" aa.iiii       a ", Int64),       // 3
+        (" aa.iiii       ae", Int64),       // 3
         (" aaa.ii        a ", Decimal),     // 4
         (" aaaa.i        a ", Float32),     // 5
         (" aaaaa.        a ", Float64),     // 6
diff --git a/src/frontend/src/handler/create_mv.rs b/src/frontend/src/handler/create_mv.rs
index 1d026b75944f0..c6bbcd2787e30 100644
--- a/src/frontend/src/handler/create_mv.rs
+++ b/src/frontend/src/handler/create_mv.rs
@@ -125,7 +125,7 @@ pub fn gen_create_mv_plan_bound(
         context.warn_to_user("EMIT ON WINDOW CLOSE is currently an experimental feature. Please use it with caution.");
     }
 
-    let mut plan_root = Planner::new(context).plan_query(query)?;
+    let mut plan_root = Planner::new_for_stream(context).plan_query(query)?;
     if let Some(col_names) = col_names {
         for name in &col_names {
             check_valid_column_name(name)?;
diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
index 5acf8e593804f..713060e849543 100644
--- a/src/frontend/src/handler/create_sink.rs
+++ b/src/frontend/src/handler/create_sink.rs
@@ -220,7 +220,7 @@ pub async fn gen_sink_plan(
     };
 
     let definition = context.normalized_sql().to_owned();
-    let mut plan_root = Planner::new(context.into()).plan_query(bound)?;
+    let mut plan_root = Planner::new_for_stream(context.into()).plan_query(bound)?;
     if let Some(col_names) = &col_names {
         plan_root.set_out_names(col_names.clone())?;
     };
diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
index 37159d142cd28..86eaba5ea1673 100644
--- a/src/frontend/src/handler/create_table.rs
+++ b/src/frontend/src/handler/create_table.rs
@@ -24,7 +24,7 @@ use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::catalog::{
     CdcTableDesc, ColumnCatalog, ColumnDesc, Engine, TableId, TableVersionId, DEFAULT_SCHEMA_NAME,
-    INITIAL_TABLE_VERSION_ID, ROWID_PREFIX,
+    INITIAL_TABLE_VERSION_ID, RISINGWAVE_ICEBERG_ROW_ID, ROWID_PREFIX,
 };
 use risingwave_common::config::MetaBackend;
 use risingwave_common::license::Feature;
@@ -1530,12 +1530,14 @@ pub async fn create_iceberg_engine_table(
 
     // For the table without primary key. We will use `_row_id` as primary key
     let sink_from = if pks.is_empty() {
-        pks = vec![ROWID_PREFIX.to_string()];
-        let [stmt]: [_; 1] =
-            Parser::parse_sql(&format!("select {}, * from {}", ROWID_PREFIX, table_name))
-                .context("unable to parse query")?
-                .try_into()
-                .unwrap();
+        pks = vec![RISINGWAVE_ICEBERG_ROW_ID.to_string()];
+        let [stmt]: [_; 1] = Parser::parse_sql(&format!(
+            "select {} as {}, * from {}",
+            ROWID_PREFIX, RISINGWAVE_ICEBERG_ROW_ID, table_name
+        ))
+        .context("unable to parse query")?
+        .try_into()
+        .unwrap();
 
         let Statement::Query(query) = &stmt else {
             panic!("unexpected statement: {:?}", stmt);
diff --git a/src/frontend/src/handler/query.rs b/src/frontend/src/handler/query.rs
index 9388f7762b08a..795ef77189d19 100644
--- a/src/frontend/src/handler/query.rs
+++ b/src/frontend/src/handler/query.rs
@@ -241,7 +241,7 @@ fn gen_batch_query_plan(
         ..
     } = bind_result;
 
-    let mut planner = Planner::new(context);
+    let mut planner = Planner::new_for_batch(context);
 
     let mut logical = planner.plan(bound)?;
     let schema = logical.schema();
diff --git a/src/frontend/src/planner/mod.rs b/src/frontend/src/planner/mod.rs
index 7ccbb905cb47d..526e6697de218 100644
--- a/src/frontend/src/planner/mod.rs
+++ b/src/frontend/src/planner/mod.rs
@@ -40,13 +40,30 @@ pub struct Planner {
     /// Mapping of `ShareId` to its share plan.
     /// The share plan can be a CTE, a source, a view and so on.
     share_cache: HashMap<ShareId, PlanRef>,
+    /// Plan for stream or batch.
+    plan_for: PlanFor,
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum PlanFor {
+    Stream,
+    Batch,
 }
 
 impl Planner {
-    pub fn new(ctx: OptimizerContextRef) -> Planner {
+    pub fn new_for_batch(ctx: OptimizerContextRef) -> Planner {
+        Planner {
+            ctx,
+            share_cache: Default::default(),
+            plan_for: PlanFor::Batch,
+        }
+    }
+
+    pub fn new_for_stream(ctx: OptimizerContextRef) -> Planner {
         Planner {
             ctx,
             share_cache: Default::default(),
+            plan_for: PlanFor::Stream,
         }
     }
 
@@ -58,4 +75,8 @@ impl Planner {
     pub fn ctx(&self) -> OptimizerContextRef {
         self.ctx.clone()
     }
+
+    pub fn plan_for(&self) -> PlanFor {
+        self.plan_for
+    }
 }
diff --git a/src/frontend/src/planner/relation.rs b/src/frontend/src/planner/relation.rs
index b5f8d39276d24..bc694042df77b 100644
--- a/src/frontend/src/planner/relation.rs
+++ b/src/frontend/src/planner/relation.rs
@@ -12,13 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+use std::ops::Deref;
 use std::rc::Rc;
 
 use either::Either;
 use itertools::Itertools;
-use risingwave_common::bail_not_implemented;
-use risingwave_common::catalog::{Field, Schema};
+use risingwave_common::catalog::{
+    ColumnCatalog, Engine, Field, Schema, RISINGWAVE_ICEBERG_ROW_ID, ROWID_PREFIX,
+};
 use risingwave_common::types::{DataType, Interval, ScalarImpl};
+use risingwave_common::{bail, bail_not_implemented};
 use risingwave_sqlparser::ast::AsOf;
 
 use crate::binder::{
@@ -26,14 +30,14 @@ use crate::binder::{
     BoundSystemTable, BoundWatermark, BoundWindowTableFunction, Relation, WindowTableFunctionKind,
 };
 use crate::error::{ErrorCode, Result};
-use crate::expr::{Expr, ExprImpl, ExprType, FunctionCall, InputRef};
+use crate::expr::{CastContext, Expr, ExprImpl, ExprType, FunctionCall, InputRef, Literal};
 use crate::optimizer::plan_node::generic::SourceNodeKind;
 use crate::optimizer::plan_node::{
     LogicalApply, LogicalCteRef, LogicalHopWindow, LogicalJoin, LogicalProject, LogicalScan,
     LogicalShare, LogicalSource, LogicalSysScan, LogicalTableFunction, LogicalValues, PlanRef,
 };
 use crate::optimizer::property::Cardinality;
-use crate::planner::Planner;
+use crate::planner::{PlanFor, Planner};
 use crate::utils::Condition;
 
 const ERROR_WINDOW_SIZE_ARG: &str =
@@ -84,7 +88,7 @@ impl Planner {
             }
         }
         let table_cardinality = base_table.table_catalog.cardinality;
-        Ok(LogicalScan::create(
+        let scan = LogicalScan::create(
             base_table.table_catalog.name().to_string(),
             base_table.table_catalog.clone(),
             base_table
@@ -93,10 +97,92 @@ impl Planner {
                 .map(|x| x.as_ref().clone().into())
                 .collect(),
             self.ctx(),
-            as_of,
+            as_of.clone(),
             table_cardinality,
-        )
-        .into())
+        );
+
+        match (base_table.table_catalog.engine, self.plan_for()) {
+            (Engine::Hummock, _) | (Engine::Iceberg, PlanFor::Stream) => Ok(scan.into()),
+            (Engine::Iceberg, PlanFor::Batch) => {
+                let opt_ctx = self.ctx();
+                let session = opt_ctx.session_ctx();
+                let db_name = session.database();
+                let catalog_reader = session.env().catalog_reader().read_guard();
+                let mut source_catalog = None;
+                for schema in catalog_reader.iter_schemas(db_name).unwrap() {
+                    if schema
+                        .get_table_by_id(&base_table.table_catalog.id)
+                        .is_some()
+                    {
+                        source_catalog = schema.get_source_by_name(
+                            &base_table.table_catalog.iceberg_source_name().unwrap(),
+                        );
+                        break;
+                    }
+                }
+                if let Some(source_catalog) = source_catalog {
+                    let column_map: HashMap<String, (usize, ColumnCatalog)> = source_catalog
+                        .columns
+                        .clone()
+                        .into_iter()
+                        .enumerate()
+                        .map(|(i, column)| (column.name().to_string(), (i, column)))
+                        .collect();
+                    let exprs = scan
+                        .table_catalog()
+                        .column_schema()
+                        .fields()
+                        .iter()
+                        .map(|field| {
+                            let source_filed_name = if field.name == ROWID_PREFIX {
+                                RISINGWAVE_ICEBERG_ROW_ID
+                            } else {
+                                &field.name
+                            };
+                            if let Some((i, source_column)) = column_map.get(source_filed_name) {
+                                if source_column.column_desc.data_type == field.data_type {
+                                    ExprImpl::InputRef(
+                                        InputRef::new(*i, field.data_type.clone()).into(),
+                                    )
+                                } else {
+                                    let mut input_ref = ExprImpl::InputRef(
+                                        InputRef::new(
+                                            *i,
+                                            source_column.column_desc.data_type.clone(),
+                                        )
+                                        .into(),
+                                    );
+                                    FunctionCall::cast_mut(
+                                        &mut input_ref,
+                                        field.data_type().clone(),
+                                        CastContext::Explicit,
+                                    )
+                                    .unwrap();
+                                    input_ref
+                                }
+                            } else {
+                                // fields like `_rw_timestamp`, would not be found in source.
+                                ExprImpl::Literal(
+                                    Literal::new(None, field.data_type.clone()).into(),
+                                )
+                            }
+                        })
+                        .collect_vec();
+                    let logical_source = LogicalSource::with_catalog(
+                        Rc::new(source_catalog.deref().clone()),
+                        SourceNodeKind::CreateMViewOrBatch,
+                        self.ctx(),
+                        as_of,
+                    )?;
+                    Ok(LogicalProject::new(logical_source.into(), exprs).into())
+                } else {
+                    bail!(
+                        "failed to plan a iceberg engine table: {}. Can't find the corresponding iceberg source. Maybe you need to recreate the table",
+                        base_table.table_catalog.name()
+                    );
+                }
+            }
+        }
     }
 
     pub(super) fn plan_source(&mut self, source: BoundSource) -> Result<PlanRef> {


Test patch:
diff --git a/e2e_test/batch/catalog/pg_cast.slt.part b/e2e_test/batch/catalog/pg_cast.slt.part
index b1558d1e144c4..00bbfbe1c0aaf 100644
--- a/e2e_test/batch/catalog/pg_cast.slt.part
+++ b/e2e_test/batch/catalog/pg_cast.slt.part
@@ -24,67 +24,68 @@ SELECT * FROM pg_catalog.pg_cast;
 20 20 701 i
 21 20 1700 i
 22 20 1043 a
-23 20 1301 i
-24 700 21 a
-25 700 23 a
-26 700 20 a
-27 700 701 i
-28 700 1700 a
-29 700 1043 a
-30 701 21 a
-31 701 23 a
-32 701 20 a
-33 701 700 a
-34 701 1700 a
-35 701 1043 a
-36 1700 21 a
-37 1700 23 a
-38 1700 20 a
-39 1700 700 i
-40 1700 701 i
-41 1700 1043 a
-42 1082 1043 a
-43 1082 1114 i
-44 1082 1184 i
-45 1043 16 e
-46 1043 21 e
-47 1043 23 e
-48 1043 20 e
-49 1043 700 e
-50 1043 701 e
-51 1043 1700 e
-52 1043 1082 e
-53 1043 1083 e
-54 1043 1114 e
-55 1043 1184 e
-56 1043 1186 e
-57 1043 17 e
-58 1043 3802 e
-59 1043 1301 e
-60 1083 1043 a
-61 1083 1186 i
-62 1114 1082 a
-63 1114 1043 a
-64 1114 1083 a
-65 1114 1184 i
-66 1184 1082 a
-67 1184 1043 a
-68 1184 1083 a
-69 1184 1114 a
-70 1186 1043 a
-71 1186 1083 a
-72 17 1043 a
-73 3802 16 e
-74 3802 21 e
-75 3802 23 e
-76 3802 20 e
-77 3802 700 e
-78 3802 701 e
-79 3802 1700 e
-80 3802 1043 a
-81 20 20 e
-82 1301 701 e
-83 1301 1043 a
+23 20 20 e
+24 20 1301 i
+25 700 21 a
+26 700 23 a
+27 700 20 a
+28 700 701 i
+29 700 1700 a
+30 700 1043 a
+31 701 21 a
+32 701 23 a
+33 701 20 a
+34 701 700 a
+35 701 1700 a
+36 701 1043 a
+37 1700 21 a
+38 1700 23 a
+39 1700 20 a
+40 1700 700 i
+41 1700 701 i
+42 1700 1043 a
+43 1082 1043 a
+44 1082 1114 i
+45 1082 1184 i
+46 1043 16 e
+47 1043 21 e
+48 1043 23 e
+49 1043 20 e
+50 1043 700 e
+51 1043 701 e
+52 1043 1700 e
+53 1043 1082 e
+54 1043 1083 e
+55 1043 1114 e
+56 1043 1184 e
+57 1043 1186 e
+58 1043 17 e
+59 1043 3802 e
+60 1043 1301 e
+61 1083 1043 a
+62 1083 1186 i
+63 1114 1082 a
+64 1114 1043 a
+65 1114 1083 a
+66 1114 1184 i
+67 1184 1082 a
+68 1184 1043 a
+69 1184 1083 a
+70 1184 1114 a
+71 1186 1043 a
+72 1186 1083 a
+73 17 1043 a
+74 3802 16 e
+75 3802 21 e
+76 3802 23 e
+77 3802 20 e
+78 3802 700 e
+79 3802 701 e
+80 3802 1700 e
+81 3802 1043 a
+82 20 20 e
+83 1301 701 e
+84 1301 1043 a
 
 query TT rowsort
 SELECT s.typname, t.typname
diff --git a/e2e_test/iceberg/test_case/iceberg_engine.slt b/e2e_test/iceberg/test_case/iceberg_engine.slt
index a954bb4018a39..0a97e4f224e8e 100644
--- a/e2e_test/iceberg/test_case/iceberg_engine.slt
+++ b/e2e_test/iceberg/test_case/iceberg_engine.slt
@@ -2,7 +2,7 @@ statement ok
 set sink_decouple = false;
 
 statement ok
-create table t(id int primary key, xxname varchar) engine = iceberg;
+create table t(id int primary key, name varchar) engine = iceberg;
 
 statement ok
 insert into t values(1, 'xxx');
@@ -10,8 +10,6 @@ insert into t values(1, 'xxx');
 statement ok
 FLUSH;
 
-sleep 5s
-
 query ??
 select * from t;
 ----
@@ -19,3 +17,48 @@ select * from t;
 
 statement ok
 DROP TABLE t;
+
+statement ok
+CREATE TABLE full_type_t (
+id bigint primary key,
+v_small_int smallint,
+v_int int,
+v_long bigint,
+v_float real,
+v_double double,
+v_varchar varchar,
+v_bool boolean,
+v_date date,
+v_timestamp timestamptz,
+v_ts_ntz timestamp,
+v_decimal decimal,
+v_map map(int, int),
+v_array int[],
+v_struct struct<a int,b int>,
+v_json jsonb,
+v_one_layer_struct struct<id bigint, v_small_int smallint, v_int int, v_long bigint, v_float real, v_double double, v_varchar varchar, v_bool boolean, v_date date, v_timestamp timestamptz, v_ts_ntz timestamp, v_decimal decimal, v_json jsonb>
+) engine = iceberg;
+
+statement ok
+INSERT INTO full_type_t VALUES
+(1, 1, 1, 1000, 1.1, 1.11, '1-1', true, '2022-03-11', '2022-03-11 01:00:00Z'::timestamptz, '2022-03-11 01:00:00',1.11, map {1:100,2:200}, array[1,2,3], row(1,2), '{"a":"foo", "b":"bar"}', row(1, 1, 1, 1000, 1.1, 1.11, '1-1', true, '2022-03-11', '2022-03-11 01:00:00Z'::timestamptz, '2022-03-11 01:00:00',1.11, '{"a":"foo", "b":"bar"}')),
+(2, 2, 2, 2000, 2.2, 2.22, '2-2', false, '2022-03-12', '2022-03-12 02:00:00Z'::timestamptz, '2022-03-12 02:00:00',2.22, map {3:300}, array[1,null,3], row(3,null), '{"k2":[2,true,4]}', row(2, 2, 2, 2000, 2.2, 2.22, '2-2', false, '2022-03-12', '2022-03-12 02:00:00Z'::timestamptz, '2022-03-12 02:00:00',2.22, '{"k2":[2,true,4]}')),
+(3, 3, 3, 3000, 3.3, 3.33, '3-3', true, '2022-03-13', '2022-03-13 03:00:00Z'::timestamptz, '2022-03-13 03:00:00','inf', null, null, null, '1', row(3, 3, 3, 3000, 3.3, 3.33, '3-3', true, '2022-03-13', '2022-03-13 03:00:00Z'::timestamptz, '2022-03-13 03:00:00','inf', '1')),
+(4, 4, 4, 4000, 4.4, 4.44, '4-4', false, '2022-03-14', '2022-03-14 04:00:00Z'::timestamptz, '2022-03-14 04:00:00','-inf', null, null, null, 'true', row(4, 4, 4, 4000, 4.4, 4.44, '4-4', false, '2022-03-14', '2022-03-14 04:00:00Z'::timestamptz, '2022-03-14 04:00:00','-inf', 'true'));
+
+statement ok
+FLUSH;
+
+query ???????????????? rowsort
+select * from full_type_t
+----
+1 1 1 1000 1.1 1.11 1-1 t 2022-03-11 2022-03-11 01:00:00+00:00 2022-03-11 01:00:00 1.1100000000 {1:100,2:200} {1,2,3} (1,2) {"a": "foo", "b": "bar"} (1,1,1,1000,1.1,1.11,1-1,t,2022-03-11,"2022-03-11 01:00:00+00:00","2022-03-11 01:00:00",1.1100000000,"{""a"": ""foo"", ""b"": ""bar""}")
+2 2 2 2000 2.2 2.22 2-2 f 2022-03-12 2022-03-12 02:00:00+00:00 2022-03-12 02:00:00 2.2200000000 {3:300} {1,NULL,3} (3,) {"k2": [2, true, 4]} (2,2,2,2000,2.2,2.22,2-2,f,2022-03-12,"2022-03-12 02:00:00+00:00","2022-03-12 02:00:00",2.2200000000,"{""k2"": [2, true, 4]}")
+3 3 3 3000 3.3 3.33 3-3 t 2022-03-13 2022-03-13 03:00:00+00:00 2022-03-13 03:00:00 999999999999999999.9999999999 NULL NULL NULL 1 (3,3,3,3000,3.3,3.33,3-3,t,2022-03-13,"2022-03-13 03:00:00+00:00","2022-03-13 03:00:00",999999999999999999.9999999999,1)
+4 4 4 4000 4.4 4.44 4-4 f 2022-03-14 2022-03-14 04:00:00+00:00 2022-03-14 04:00:00 -999999999999999999.9999999999 NULL NULL NULL true (4,4,4,4000,4.4,4.44,4-4,f,2022-03-14,"2022-03-14 04:00:00+00:00","2022-03-14 04:00:00",-999999999999999999.9999999999,true)
+
+statement ok
+DROP TABLE full_type_t;
+
+
+
diff --git a/src/expr/impl/tests/sig.rs b/src/expr/impl/tests/sig.rs
index 2dc8aacdb203f..8dc0bb29ebdf5 100644
--- a/src/expr/impl/tests/sig.rs
+++ b/src/expr/impl/tests/sig.rs
@@ -60,7 +60,7 @@ fn test_func_sig_map() {
     let expected = expect_test::expect![[r#"
         [
             "cast(anyarray) -> character varying/anyarray",
-            "cast(bigint) -> rw_int256/integer/smallint/numeric/double precision/real/character varying",
+            "cast(bigint) -> rw_int256/serial/integer/smallint/numeric/double precision/real/character varying",
             "cast(boolean) -> integer/character varying",
             "cast(character varying) -> jsonb/interval/timestamp without time zone/time without time zone/date/rw_int256/real/double precision/numeric/smallint/integer/bigint/character varying/boolean/bytea/anyarray",
             "cast(date) -> timestamp without time zone/character varying",
diff --git a/src/frontend/planner_test/src/lib.rs b/src/frontend/planner_test/src/lib.rs
index fd299f5da919b..ad2c6554a2b96 100644
--- a/src/frontend/planner_test/src/lib.rs
+++ b/src/frontend/planner_test/src/lib.rs
@@ -607,7 +607,7 @@ impl TestCase {
             }
         };
 
-        let mut planner = Planner::new(context.clone());
+        let mut planner = Planner::new_for_stream(context.clone());
 
         let plan_root = match planner.plan(bound) {
             Ok(plan_root) => {
diff --git a/src/tests/sqlsmith/tests/frontend/mod.rs b/src/tests/sqlsmith/tests/frontend/mod.rs
index 6eedb4e41876b..539c5706a33f1 100644
--- a/src/tests/sqlsmith/tests/frontend/mod.rs
+++ b/src/tests/sqlsmith/tests/frontend/mod.rs
@@ -185,7 +185,7 @@ fn run_batch_query(
     let bound = binder
         .bind(stmt)
         .map_err(|e| Failed::from(format!("Failed to bind:\nReason:\n{}", e.as_report())))?;
-    let mut planner = Planner::new(context);
+    let mut planner = Planner::new_for_batch(context);
     let mut plan_root = planner.plan(bound).map_err(|e| {
         Failed::from(format!(
             "Failed to generate logical plan:\nReason:\n{}",