refactor: binding struct values when they are parsed as row

src/common/src/types/mod.rs

@@ -19,7 +19,7 @@ use bytes::{Buf, BufMut};
 use risingwave_pb::data::DataType as ProstDataType;
 use serde::{Deserialize, Serialize};
 
-use crate::error::{internal_error, ErrorCode, Result, RwError};
+use crate::error::{ErrorCode, Result, RwError};
 mod native_type;
 
 mod scalar_impl;
@@ -326,16 +326,6 @@ for_all_scalar_variants! { scalar_impl_enum }
 pub type Datum = Option<ScalarImpl>;
 pub type DatumRef<'a> = Option<ScalarRefImpl<'a>>;
 
-pub fn get_data_type_from_datum(datum: &Datum) -> Result<DataType> {
-    match datum {
-        // TODO: Predicate data type from None Datum
-        None => Err(internal_error(
-            "cannot get data type from None Datum".to_string(),
-        )),
-        Some(scalar) => scalar.data_type(),
-    }
-}
-
 /// Convert a [`Datum`] to a [`DatumRef`].
 pub fn to_datum_ref(datum: &Datum) -> DatumRef<'_> {
     datum.as_ref().map(|d| d.as_scalar_ref_impl())

src/common/src/types/scalar_impl.rs

@@ -300,40 +300,6 @@ impl ScalarImpl {
         }
         for_all_scalar_variants! { impl_all_get_ident, self }
     }
-
-    pub(crate) fn data_type(&self) -> Result<DataType> {
-        let data_type = match self {
-            ScalarImpl::Int16(_) => DataType::Int16,
-            ScalarImpl::Int32(_) => DataType::Int32,
-            ScalarImpl::Int64(_) => DataType::Int64,
-            ScalarImpl::Float32(_) => DataType::Float32,
-            ScalarImpl::Float64(_) => DataType::Float64,
-            ScalarImpl::Utf8(_) => DataType::Varchar,
-            ScalarImpl::Bool(_) => DataType::Boolean,
-            ScalarImpl::Decimal(_) => DataType::Decimal,
-            ScalarImpl::Interval(_) => DataType::Interval,
-            ScalarImpl::NaiveDate(_) => DataType::Date,
-            ScalarImpl::NaiveDateTime(_) => DataType::Timestamp,
-            ScalarImpl::NaiveTime(_) => DataType::Time,
-            ScalarImpl::Struct(data) => {
-                let types = data
-                    .fields()
-                    .iter()
-                    .map(get_data_type_from_datum)
-                    .collect::<Result<Vec<_>>>()?;
-                DataType::Struct {
-                    fields: types.into(),
-                }
-            }
-            ScalarImpl::List(data) => {
-                let data = data.values().get(0).ok_or_else(|| {
-                    internal_error("cannot get data type from empty list".to_string())
-                })?;
-                get_data_type_from_datum(data)?
-            }
-        };
-        Ok(data_type)
-    }
 }
 
 impl<'scalar> ScalarRefImpl<'scalar> {

src/frontend/src/binder/values.rs

@@ -16,12 +16,12 @@ use itertools::Itertools;
 use risingwave_common::array::StructValue;
 use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::error::{ErrorCode, Result, TrackingIssue};
-use risingwave_common::types::{get_data_type_from_datum, DataType, Datum, Scalar};
+use risingwave_common::types::{DataType, Scalar};
 use risingwave_sqlparser::ast::{Expr, Values};
 
 use super::bind_context::Clause;
 use crate::binder::Binder;
-use crate::expr::{align_types, ExprImpl, Literal};
+use crate::expr::{align_types, Expr as _, ExprImpl, Literal};
 
 #[derive(Debug)]
 pub struct BoundValues {
@@ -86,27 +86,27 @@ impl Binder {
     /// e.g. Row(1,2,(1,2,3)).
     /// Only accept value and row expr in row.
     pub fn bind_row(&mut self, exprs: &[Expr]) -> Result<Literal> {
-        let datums = exprs
+        let literals = exprs
             .iter()
             .map(|e| match e {
-                Expr::Value(value) => Ok(self.bind_value(value.clone())?.get_data().clone()),
-                Expr::Row(expr) => Ok(self.bind_row(expr)?.get_data().clone()),
+                Expr::Value(value) => Ok(self.bind_value(value.clone())?),
+                Expr::Row(expr) => Ok(self.bind_row(expr)?),
                 _ => Err(ErrorCode::NotImplemented(
                     format!("unsupported expression {:?}", e),
                     TrackingIssue::none(),
                 )
                 .into()),
             })
-            .collect::<Result<Vec<Datum>>>()?;
-        let value = StructValue::new(datums);
+            .collect::<Result<Vec<_>>>()?;
         let data_type = DataType::Struct {
-            fields: value
-                .fields()
+            fields: literals
                 .iter()
-                .map(get_data_type_from_datum)
-                .collect::<Result<Vec<_>>>()?
+                .map(|l| l.return_type())
+                .collect_vec()
                 .into(),
         };
+        let value = StructValue::new(literals.iter().map(|f| f.get_data().clone()).collect_vec());
+
         Ok(Literal::new(Some(value.to_scalar_value()), data_type))
     }
 }
@@ -119,7 +119,6 @@ mod tests {
 
     use super::*;
     use crate::binder::test_utils::mock_binder;
-    use crate::expr::Expr as _;
 
     #[test]
     fn test_bind_values() {