small refactor

risedev.yml

@@ -36,7 +36,6 @@ profile:
       # - use: grafana     # visualization
 
       - use: meta-node
-        # meta-backend: etcd
       - use: compute-node
       - use: frontend
 

src/common/src/types/decimal.rs

@@ -21,7 +21,7 @@ use bytes::{BufMut, BytesMut};
 use num_traits::{
     CheckedAdd, CheckedDiv, CheckedMul, CheckedNeg, CheckedRem, CheckedSub, Num, One, Zero,
 };
-use postgres_types::{accepts, to_sql_checked, IsNull, ToSql};
+use postgres_types::{accepts, to_sql_checked, FromSql, IsNull, ToSql, Type};
 use risingwave_common_estimate_size::ZeroHeapSize;
 use rust_decimal::prelude::FromStr;
 use rust_decimal::{Decimal as RustDecimal, Error, MathematicalOps as _, RoundingStrategy};
@@ -96,7 +96,7 @@ impl ToSql for Decimal {
 
     fn to_sql(
         &self,
-        ty: &postgres_types::Type,
+        ty: &Type,
         out: &mut BytesMut,
     ) -> Result<IsNull, Box<dyn std::error::Error + Sync + Send>>
     where
@@ -132,9 +132,9 @@ impl ToSql for Decimal {
     }
 }
 
-impl<'a> postgres_types::FromSql<'a> for Decimal {
+impl<'a> FromSql<'a> for Decimal {
     fn from_sql(
-        ty: &postgres_types::Type,
+        ty: &Type,
         raw: &'a [u8],
     ) -> Result<Self, Box<dyn std::error::Error + 'static + Sync + Send>> {
         let mut rdr = Cursor::new(raw);
@@ -145,17 +145,15 @@ impl<'a> postgres_types::FromSql<'a> for Decimal {
             0xC000 => Ok(Self::NaN),
             0xD000 => Ok(Self::PositiveInf),
             0xF000 => Ok(Self::NegativeInf),
-            _ => <RustDecimal as postgres_types::FromSql>::from_sql(ty, raw).map(Self::Normalized),
+            _ => RustDecimal::from_sql(ty, raw).map(Self::Normalized),
         }
     }
 
-    fn accepts(ty: &postgres_types::Type) -> bool {
-        matches!(*ty, postgres_types::Type::NUMERIC)
+    fn accepts(ty: &Type) -> bool {
+        matches!(*ty, Type::NUMERIC)
     }
 }
 
-
-
 macro_rules! impl_convert_int {
     ($T:ty) => {
         impl core::convert::From<$T> for Decimal {

src/common/src/types/from_sql.rs

@@ -23,16 +23,16 @@ impl<'a> FromSql<'a> for ScalarImpl {
         raw: &'a [u8],
     ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
         Ok(match *ty {
-            Type::BOOL => ScalarImpl::from(<bool as FromSql>::from_sql(ty, raw)?),
-            Type::INT2 => ScalarImpl::from(<i16 as FromSql>::from_sql(ty, raw)?),
-            Type::INT4 => ScalarImpl::from(<i32 as FromSql>::from_sql(ty, raw)?),
-            Type::INT8 => ScalarImpl::from(<i64 as FromSql>::from_sql(ty, raw)?),
-            Type::FLOAT4 => ScalarImpl::from(<f32 as FromSql>::from_sql(ty, raw)?),
-            Type::FLOAT8 => ScalarImpl::from(<f64 as FromSql>::from_sql(ty, raw)?),
-            Type::DATE => ScalarImpl::from(<Date as FromSql>::from_sql(ty, raw)?),
-            Type::TIME => ScalarImpl::from(<Time as FromSql>::from_sql(ty, raw)?),
-            Type::TIMESTAMP => ScalarImpl::from(<Timestamp as FromSql>::from_sql(ty, raw)?),
-            Type::TIMESTAMPTZ => ScalarImpl::from(<Timestamptz as FromSql>::from_sql(ty, raw)?),
+            Type::BOOL => ScalarImpl::from(bool::from_sql(ty, raw)?),
+            Type::INT2 => ScalarImpl::from(i16::from_sql(ty, raw)?),
+            Type::INT4 => ScalarImpl::from(i32::from_sql(ty, raw)?),
+            Type::INT8 => ScalarImpl::from(i64::from_sql(ty, raw)?),
+            Type::FLOAT4 => ScalarImpl::from(f32::from_sql(ty, raw)?),
+            Type::FLOAT8 => ScalarImpl::from(f64::from_sql(ty, raw)?),
+            Type::DATE => ScalarImpl::from(Date::from_sql(ty, raw)?),
+            Type::TIME => ScalarImpl::from(Time::from_sql(ty, raw)?),
+            Type::TIMESTAMP => ScalarImpl::from(Timestamp::from_sql(ty, raw)?),
+            Type::TIMESTAMPTZ => ScalarImpl::from(Timestamptz::from_sql(ty, raw)?),
             Type::JSONB => ScalarImpl::from(JsonbVal::from_sql(ty, raw)?),
             Type::INTERVAL => ScalarImpl::from(Interval::from_sql(ty, raw)?),
             Type::BYTEA => ScalarImpl::from(Vec::<u8>::from_sql(ty, raw)?.into_boxed_slice()),

src/common/src/types/ordered_float.rs

@@ -58,7 +58,7 @@ use num_traits::{
     Bounded, CheckedAdd, CheckedDiv, CheckedMul, CheckedNeg, CheckedRem, CheckedSub, Num, One, Pow,
     Zero,
 };
-use postgres_types::{accepts, to_sql_checked, IsNull, Type};
+use postgres_types::{accepts, to_sql_checked, IsNull, ToSql, Type};
 
 // masks for the parts of the IEEE 754 float
 const SIGN_MASK: u64 = 0x8000000000000000u64;
@@ -108,7 +108,7 @@ impl<T: Float> OrderedFloat<T> {
     }
 }
 
-impl postgres_types::ToSql for OrderedFloat<f32> {
+impl ToSql for OrderedFloat<f32> {
     accepts!(FLOAT4);
 
     to_sql_checked!();
@@ -121,7 +121,7 @@ impl postgres_types::ToSql for OrderedFloat<f32> {
     }
 }
 
-impl postgres_types::ToSql for OrderedFloat<f64> {
+impl ToSql for OrderedFloat<f64> {
     accepts!(FLOAT8);
 
     to_sql_checked!();

src/common/src/types/timestamptz.rs

@@ -21,7 +21,7 @@ use byteorder::{BigEndian, ReadBytesExt};
 use bytes::BytesMut;
 use chrono::{DateTime, Datelike, TimeZone, Utc};
 use chrono_tz::Tz;
-use postgres_types::{accepts, to_sql_checked, IsNull, Type};
+use postgres_types::{accepts, to_sql_checked, IsNull, ToSql, Type};
 use risingwave_common_estimate_size::ZeroHeapSize;
 use serde::{Deserialize, Serialize};
 
@@ -38,7 +38,7 @@ pub struct Timestamptz(i64);
 
 impl ZeroHeapSize for Timestamptz {}
 
-impl postgres_types::ToSql for Timestamptz {
+impl ToSql for Timestamptz {
     accepts!(TIMESTAMPTZ);
 
     to_sql_checked!();
@@ -52,12 +52,12 @@ impl postgres_types::ToSql for Timestamptz {
     }
 }
 
-impl<'a> postgres_types::FromSql<'a> for Timestamptz {
+impl<'a> FromSql<'a> for Timestamptz {
     fn from_sql(
         ty: &Type,
         raw: &'a [u8],
     ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
-        let instant = <DateTime<Utc> as postgres_types::FromSql>::from_sql(ty, raw)?;
+        let instant = DateTime::<Utc>::from_sql(ty, raw)?;
         Ok(Self::from(instant))
     }
 

src/connector/src/parser/unified/json.rs

@@ -211,6 +211,7 @@ impl JsonParseOptions {
             got: value.value_type().to_string(),
             value: value.to_string(),
         };
+
         let v: ScalarImpl = match (type_expected, value.value_type()) {
             (_, ValueType::Null) => return Ok(DatumCow::NULL),
             // ---- Boolean -----