x

Signed-off-by: luofucong <luofc@foxmail.com>

src/api/src/helper.rs

@@ -20,13 +20,15 @@ use common_decimal::decimal128::{DECIMAL128_DEFAULT_SCALE, DECIMAL128_MAX_PRECIS
 use common_time::time::Time;
 use common_time::timestamp::TimeUnit;
 use common_time::{Date, IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp};
-use datatypes::json::value::{JsonNumber, JsonValue, JsonValueRef, JsonVariant, JsonVariantRef};
+use datatypes::json::value::{JsonNumber, JsonValue, JsonVariant};
 use datatypes::prelude::{ConcreteDataType, ValueRef};
 use datatypes::types::json_type::JsonNativeType;
 use datatypes::types::{
     IntervalType, JsonFormat, JsonType, StructField, StructType, TimeType, TimestampType,
 };
-use datatypes::value::{ListValueRef, OrderedF32, OrderedF64, StructValueRef, Value};
+use datatypes::value::{
+    ListValue, ListValueRef, OrderedF32, OrderedF64, StructValue, StructValueRef, Value,
+};
 use datatypes::vectors::VectorRef;
 use greptime_proto::v1::column_data_type_extension::TypeExt;
 use greptime_proto::v1::ddl_request::Expr;
@@ -732,15 +734,71 @@ pub fn convert_to_pb_decimal128(v: Decimal128) -> v1::Decimal128 {
     v1::Decimal128 { hi, lo }
 }
 
+/// A protobuf value decoded without copying scalar strings and binary values.
+pub enum DecodedValue<'a> {
+    /// A value borrowing from the protobuf value.
+    Ref(ValueRef<'a>),
+    /// An owned value for representations that cannot borrow from protobuf.
+    Owned(Value),
+}
+
+impl<'a> DecodedValue<'a> {
+    /// Converts this decoded value into an owned value.
+    pub fn into_value(self) -> Value {
+        match self {
+            Self::Ref(value) => value.into(),
+            Self::Owned(value) => value,
+        }
+    }
+
+    /// Returns the decoded value ref if it does not own its representation.
+    pub fn into_value_ref(self) -> Option<ValueRef<'a>> {
+        match self {
+            Self::Ref(value) => Some(value),
+            Self::Owned(_) => None,
+        }
+    }
+
+    /// Returns whether this value is null.
+    pub fn is_null(&self) -> bool {
+        match self {
+            Self::Ref(value) => value.is_null(),
+            Self::Owned(value) => value.is_null(),
+        }
+    }
+
+    /// Returns the estimated size of the underlying data.
+    pub fn data_size(&self) -> usize {
+        match self {
+            Self::Ref(value) => value.data_size(),
+            Self::Owned(Value::Json(value)) => std::mem::size_of_val(value.as_ref()),
+            Self::Owned(value) => value.as_value_ref().data_size(),
+        }
+    }
+}
+
+impl From<DecodedValue<'_>> for Value {
+    fn from(value: DecodedValue<'_>) -> Self {
+        value.into_value()
+    }
+}
+
+impl<'a> From<ValueRef<'a>> for DecodedValue<'a> {
+    fn from(value: ValueRef<'a>) -> Self {
+        Self::Ref(value)
+    }
+}
+
+/// Decodes a protobuf value, borrowing its representation when possible.
 pub fn pb_value_to_value_ref<'a>(
     value: &'a v1::Value,
     datatype_ext: Option<&'a ColumnDataTypeExtension>,
-) -> ValueRef<'a> {
+) -> DecodedValue<'a> {
     let Some(value) = &value.value_data else {
-        return ValueRef::Null;
+        return DecodedValue::Ref(ValueRef::Null);
     };
 
-    match value {
+    let value = match value {
         ValueData::I8Value(v) => ValueRef::Int8(*v as i8),
         ValueData::I16Value(v) => ValueRef::Int16(*v as i16),
         ValueData::I32Value(v) => ValueRef::Int32(*v),
@@ -827,11 +885,21 @@ pub fn pb_value_to_value_ref<'a>(
                 })
                 .collect::<Vec<_>>();
 
-            let list_value = ListValueRef::RefList {
+            let item_datatype = Arc::new(item_type);
+            if items.iter().any(|x| matches!(x, DecodedValue::Owned(_))) {
+                return DecodedValue::Owned(Value::List(ListValue::new(
+                    items.into_iter().map(DecodedValue::into_value).collect(),
+                    item_datatype,
+                )));
+            }
+            let items = items
+                .into_iter()
+                .map(|x| x.into_value_ref().unwrap())
+                .collect();
+            ValueRef::List(ListValueRef::RefList {
                 val: items,
-                item_datatype: Arc::new(item_type.clone()),
-            };
-            ValueRef::List(list_value)
+                item_datatype,
+            })
         }
 
         ValueData::StructValue(struct_value) => {
@@ -864,20 +932,27 @@ pub fn pb_value_to_value_ref<'a>(
                 .iter()
                 .zip(struct_datatype_ext.fields.iter())
                 .map(|(item, field)| pb_value_to_value_ref(item, field.datatype_extension.as_ref()))
-                .collect::<Vec<ValueRef>>();
+                .collect::<Vec<_>>();
 
-            let struct_value_ref = StructValueRef::RefList {
-                val: items,
-                fields: StructType::new(Arc::new(struct_fields)),
-            };
-            ValueRef::Struct(struct_value_ref)
+            let fields = StructType::new(Arc::new(struct_fields));
+            if items.iter().any(|x| matches!(x, DecodedValue::Owned(_))) {
+                return DecodedValue::Owned(Value::Struct(StructValue::new(
+                    items.into_iter().map(DecodedValue::into_value).collect(),
+                    fields,
+                )));
+            }
+            let items = items
+                .into_iter()
+                .map(|x| x.into_value_ref().unwrap())
+                .collect();
+            ValueRef::Struct(StructValueRef::RefList { val: items, fields })
         }
 
-        ValueData::JsonValue(inner_value) => {
-            let value = decode_json_value(inner_value);
-            ValueRef::Json(Box::new(value))
+        ValueData::JsonValue(value) => {
+            return DecodedValue::Owned(Value::Json(Box::new(decode_json_value(value))));
         }
-    }
+    };
+    DecodedValue::Ref(value)
 }
 
 /// Returns true if the pb semantic type is valid.
@@ -930,21 +1005,21 @@ pub fn encode_json_value(value: JsonValue) -> v1::JsonValue {
     helper(value.into_variant())
 }
 
-fn decode_json_value(value: &v1::JsonValue) -> JsonValueRef<'_> {
+fn decode_json_value(value: &v1::JsonValue) -> JsonValue {
     let (variant, json_type) = decode_json_value_parts(value);
-    JsonValueRef::new_with_type(variant, json_type)
+    JsonValue::new_with_type(variant, json_type)
 }
 
-fn decode_json_value_parts(value: &v1::JsonValue) -> (JsonVariantRef<'_>, JsonNativeType) {
+fn decode_json_value_parts(value: &v1::JsonValue) -> (JsonVariant, JsonNativeType) {
     let Some(value) = &value.value else {
-        return (JsonVariantRef::Null, JsonNativeType::Null);
+        return (JsonVariant::Null, JsonNativeType::Null);
     };
     match value {
-        json_value::Value::Boolean(x) => (JsonVariantRef::Bool(*x), JsonNativeType::Bool),
+        json_value::Value::Boolean(x) => (JsonVariant::Bool(*x), JsonNativeType::Bool),
         json_value::Value::Int(x) => ((*x).into(), JsonNativeType::i64()),
         json_value::Value::Uint(x) => ((*x).into(), JsonNativeType::u64()),
         json_value::Value::Float(x) => ((*x).into(), JsonNativeType::f64()),
-        json_value::Value::Str(x) => (x.as_str().into(), JsonNativeType::String),
+        json_value::Value::Str(x) => (x.clone().into(), JsonNativeType::String),
         json_value::Value::Array(array) => {
             let mut variants = Vec::with_capacity(array.items.len());
             let mut item_type = JsonNativeType::Null;
@@ -956,34 +1031,29 @@ fn decode_json_value_parts(value: &v1::JsonValue) -> (JsonVariantRef<'_>, JsonNa
                 }
             }
             (
-                JsonVariantRef::Array(variants),
+                JsonVariant::Array(variants),
                 JsonNativeType::Array(Box::new(item_type)),
             )
         }
         json_value::Value::Object(object) => {
-            let mut variants = Vec::with_capacity(object.entries.len());
-            let mut fields = Vec::with_capacity(object.entries.len());
+            let mut variants = BTreeMap::new();
+            let mut fields = BTreeMap::new();
             for entry in &object.entries {
                 let Some(value) = &entry.value else {
                     continue;
                 };
                 let (variant, json_type) = decode_json_value_parts(value);
-                variants.push((entry.key.as_str(), variant));
-                fields.push((entry.key.clone(), json_type));
+                variants.insert(entry.key.clone(), variant);
+                fields.insert(entry.key.clone(), json_type);
             }
-            let variants = variants.into_iter().collect::<BTreeMap<_, _>>();
-            let fields = fields.into_iter().collect::<BTreeMap<_, _>>();
             let json_type = if fields.is_empty() {
                 JsonNativeType::Null
             } else {
                 JsonNativeType::Object(fields)
             };
-            (JsonVariantRef::Object(variants), json_type)
+            (JsonVariant::Object(variants), json_type)
         }
-        json_value::Value::Variant(x) => (
-            JsonVariantRef::Variant(x.as_slice()),
-            JsonNativeType::Variant,
-        ),
+        json_value::Value::Variant(x) => (JsonVariant::Variant(x.clone()), JsonNativeType::Variant),
     }
 }
 
@@ -1830,37 +1900,37 @@ mod tests {
         let proto = encode_json_value(json.clone());
         assert!(proto.value.is_none());
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = true.into();
         let proto = encode_json_value(json.clone());
         assert_eq!(proto.value, Some(json_value::Value::Boolean(true)));
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = (-1i64).into();
         let proto = encode_json_value(json.clone());
         assert_eq!(proto.value, Some(json_value::Value::Int(-1)));
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = 1u64.into();
         let proto = encode_json_value(json.clone());
         assert_eq!(proto.value, Some(json_value::Value::Uint(1)));
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = 1.0f64.into();
         let proto = encode_json_value(json.clone());
         assert_eq!(proto.value, Some(json_value::Value::Float(1.0)));
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = "s".into();
         let proto = encode_json_value(json.clone());
         assert_eq!(proto.value, Some(json_value::Value::Str("s".to_string())));
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = [1i64, 2, 3].into();
         let proto = encode_json_value(json.clone());
@@ -1885,7 +1955,7 @@ mod tests {
             decode_json_value_parts(&proto).1
         );
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let proto = v1::JsonValue {
             value: Some(json_value::Value::Array(JsonList {
@@ -1915,7 +1985,7 @@ mod tests {
             decode_json_value_parts(&proto).1
         );
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = [("k3", 3i64), ("k2", 2i64), ("k1", 1i64)].into();
         let proto = encode_json_value(json.clone());
@@ -1953,7 +2023,7 @@ mod tests {
             decode_json_value_parts(&proto).1
         );
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = [("null", ()); 0].into();
         let proto = encode_json_value(json.clone());
@@ -1963,7 +2033,7 @@ mod tests {
         );
         assert_eq!(JsonNativeType::Null, decode_json_value_parts(&proto).1);
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
 
         let json: JsonValue = [
             ("null", JsonVariant::from(())),
@@ -2050,6 +2120,13 @@ mod tests {
             }))
         );
         let value = decode_json_value(&proto);
-        assert_eq!(json.as_ref(), value);
+        assert_eq!(json, value);
+        let value = v1::Value {
+            value_data: Some(ValueData::JsonValue(proto)),
+        };
+        assert_eq!(
+            Value::Json(Box::new(json)),
+            pb_value_to_value_ref(&value, None).into_value()
+        );
     }
 }

src/datatypes/src/json/value.rs

@@ -310,6 +310,15 @@ impl JsonValue {
         ().into()
     }
 
+    /// Creates an owned JSON value with its precomputed native type.
+    /// The native type must describe `json_variant` exactly.
+    pub fn new_with_type(json_variant: JsonVariant, json_type: JsonNativeType) -> Self {
+        Self {
+            json_type: OnceLock::from(Arc::new(json_type)),
+            json_variant,
+        }
+    }
+
     pub(crate) fn new(json_variant: JsonVariant) -> Self {
         Self {
             json_type: OnceLock::new(),

src/datatypes/src/vectors.rs

@@ -200,6 +200,11 @@ pub trait MutableVector: Send + Sync {
     /// Try to push value ref to this mutable vector.
     fn try_push_value_ref(&mut self, value: &ValueRef) -> Result<()>;
 
+    /// Tries to push an owned value into this mutable vector.
+    fn try_push_value(&mut self, value: Value) -> Result<()> {
+        self.try_push_value_ref(&value.as_value_ref())
+    }
+
     /// Push value ref to this mutable vector.
     ///
     /// # Panics

src/datatypes/src/vectors/json/builder.rs

@@ -212,6 +212,28 @@ impl MutableVector for JsonVectorBuilder {
         Ok(())
     }
 
+    fn try_push_value(&mut self, value: Value) -> Result<()> {
+        let Value::Json(value) = value else {
+            return TryFromValueSnafu {
+                reason: format!("expected json value, got {value:?}"),
+            }
+            .fail();
+        };
+        let json_type = value.json_type();
+        if !matches!(json_type, JsonNativeType::Object(_) | JsonNativeType::Null) {
+            return TryFromValueSnafu {
+                reason: format!("expected json object value, got {value:?}"),
+            }
+            .fail();
+        }
+        if !is_include(&self.merged_type, json_type) {
+            self.merged_type.merge(json_type);
+        }
+
+        self.values.push(value.into_variant());
+        Ok(())
+    }
+
     fn push_null(&mut self) {
         self.values.push(JsonVariant::Null)
     }
@@ -255,7 +277,7 @@ mod tests {
         let second = parse_json_value(r#"{"id":2,"extra":true,"payload":"raw"}"#);
         builder.try_push_value_ref(&first.as_value_ref())?;
         builder.push_null();
-        builder.try_push_value_ref(&second.as_value_ref())?;
+        builder.try_push_value(second)?;
 
         let merged_type = JsonNativeType::Object(JsonObjectType::from([
             ("extra".to_string(), JsonNativeType::Bool),

src/flow/src/batching_mode/time_window.rs

@@ -18,7 +18,7 @@
 use std::collections::BTreeSet;
 use std::sync::Arc;
 
-use api::helper::pb_value_to_value_ref;
+use api::helper::{DecodedValue, pb_value_to_value_ref};
 use arrow::array::{
     TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
     TimestampSecondArray,
@@ -290,8 +290,11 @@ impl TimeWindowExpr {
 
             let mut vector = cdt.create_mutable_vector(rows.rows.len());
             for row in rows.rows {
-                let value = pb_value_to_value_ref(&row.values[ts_col_index], None);
-                vector.try_push_value_ref(&value).context(DataTypeSnafu {
+                let result = match pb_value_to_value_ref(&row.values[ts_col_index], None) {
+                    DecodedValue::Ref(value) => vector.try_push_value_ref(&value),
+                    DecodedValue::Owned(value) => vector.try_push_value(value),
+                };
+                result.context(DataTypeSnafu {
                     msg: "Failed to convert rows to columns",
                 })?;
             }

src/flow/src/repr.rs

@@ -194,7 +194,7 @@ impl From<ProtoRow> for Row {
         Row::pack(
             row.values
                 .iter()
-                .map(|pb_val| -> Value { pb_value_to_value_ref(pb_val, None).into() }),
+                .map(|pb_val| pb_value_to_value_ref(pb_val, None).into_value()),
         )
     }
 }

src/metric-engine/src/row_modifier.rs

@@ -101,11 +101,15 @@ impl RowModifier {
             let internal_columns = [
                 (
                     ReservedColumnId::table_id(),
-                    api::helper::pb_value_to_value_ref(&table_id_value, None),
+                    api::helper::pb_value_to_value_ref(&table_id_value, None)
+                        .into_value_ref()
+                        .expect("table id must have a borrowed value representation"),
                 ),
                 (
                     ReservedColumnId::tsid(),
-                    api::helper::pb_value_to_value_ref(&tsid, None),
+                    api::helper::pb_value_to_value_ref(&tsid, None)
+                        .into_value_ref()
+                        .expect("tsid must have a borrowed value representation"),
                 ),
             ];
             self.codec
@@ -404,7 +408,9 @@ impl RowIter<'_> {
                     api::helper::pb_value_to_value_ref(
                         &self.row.values[idx.index],
                         self.schema[idx.index].datatype_extension.as_ref(),
-                    ),
+                    )
+                    .into_value_ref()
+                    .expect("primary key must have a borrowed value representation"),
                 )
             })
     }

src/mito-codec/src/key_values.rs

@@ -14,6 +14,7 @@
 
 use std::collections::HashMap;
 
+use api::helper::DecodedValue;
 use api::v1::{ColumnSchema, Mutation, OpType, Row, Rows};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::value::ValueRef;
@@ -206,13 +207,15 @@ impl KeyValue<'_> {
                 Some(i) => api::helper::pb_value_to_value_ref(
                     &self.row.values[*i],
                     self.schema[*i].datatype_extension.as_ref(),
-                ),
+                )
+                .into_value_ref()
+                .expect("primary key must have a borrowed value representation"),
                 None => ValueRef::Null,
             })
     }
 
-    /// Get field columns.
-    pub fn fields(&self) -> impl Iterator<Item = ValueRef<'_>> {
+    /// Gets field columns, decoding JSON values into owned values.
+    pub fn fields(&self) -> impl Iterator<Item = DecodedValue<'_>> {
         self.helper.indices[self.helper.num_primary_key_column + 1..]
             .iter()
             .map(|idx| match idx {
@@ -220,7 +223,7 @@ impl KeyValue<'_> {
                     &self.row.values[*i],
                     self.schema[*i].datatype_extension.as_ref(),
                 ),
-                None => ValueRef::Null,
+                None => DecodedValue::Ref(ValueRef::Null),
             })
     }
 
@@ -232,6 +235,8 @@ impl KeyValue<'_> {
             &self.row.values[index],
             self.schema[index].datatype_extension.as_ref(),
         )
+        .into_value_ref()
+        .expect("timestamp must have a borrowed value representation")
     }
 
     /// Get number of primary key columns.
@@ -443,7 +448,13 @@ mod tests {
             let actual_keys: Vec<_> = kv.primary_keys().collect();
             assert_eq!(expect_keys, actual_keys);
             let expect_values: Vec<_> = values.iter().map(|v| ValueRef::from(*v)).collect();
-            let actual_values: Vec<_> = kv.fields().collect();
+            let actual_values: Vec<_> = kv
+                .fields()
+                .map(|value| match value {
+                    DecodedValue::Ref(value) => value,
+                    DecodedValue::Owned(_) => unreachable!(),
+                })
+                .collect();
             assert_eq!(expect_values, actual_values);
         }
     }

src/mito2/src/memtable/builder.rs

@@ -16,6 +16,7 @@
 
 use std::sync::Arc;
 
+use api::helper::DecodedValue;
 use datatypes::arrow;
 use datatypes::arrow::array::{
     Array, ArrayDataBuilder, BufferBuilder, GenericByteArray, NullBufferBuilder, UInt8BufferBuilder,
@@ -58,6 +59,17 @@ impl FieldBuilder {
         }
     }
 
+    /// Pushes a field value into builder.
+    pub(crate) fn push_field(&mut self, value: DecodedValue) -> datatypes::error::Result<()> {
+        match value {
+            DecodedValue::Ref(value) => self.push(value),
+            DecodedValue::Owned(value) => match self {
+                FieldBuilder::String(_) => self.push(value.as_value_ref()),
+                FieldBuilder::Other(builder) => builder.try_push_value(value),
+            },
+        }
+    }
+
     /// Push n null values into builder.
     pub(crate) fn push_nulls(&mut self, n: usize) {
         match self {

src/mito2/src/memtable/time_series.rs

@@ -20,6 +20,7 @@ use std::sync::atomic::{AtomicI64, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 
+use api::helper::DecodedValue;
 use api::v1::OpType;
 use common_recordbatch::filter::SimpleFilterEvaluator;
 use common_telemetry::{debug, error};
@@ -733,13 +734,16 @@ impl Series {
     }
 
     /// Pushes a row of values into Series. Return the size of values.
-    pub(crate) fn push<'a>(
+    pub(crate) fn push<'a, T>(
         &mut self,
         ts: ValueRef<'a>,
         sequence: u64,
         op_type: OpType,
-        values: impl Iterator<Item = ValueRef<'a>>,
-    ) -> usize {
+        values: impl Iterator<Item = T>,
+    ) -> usize
+    where
+        T: Into<DecodedValue<'a>>,
+    {
         // + 10 to avoid potential reallocation.
         if self.active.len() + 10 > self.capacity {
             let region_metadata = self.region_metadata.clone();
@@ -869,13 +873,17 @@ impl ValueBuilder {
     /// Returns the size of field values.
     ///
     /// In this method, we don't check the data type of the value, because it is already checked in the caller.
-    pub(crate) fn push<'a>(
+    pub(crate) fn push<'a, T>(
         &mut self,
         ts: ValueRef,
         sequence: u64,
         op_type: u8,
-        fields: impl Iterator<Item = ValueRef<'a>>,
-    ) -> usize {
+        fields: impl Iterator<Item = T>,
+    ) -> usize
+    where
+        T: Into<DecodedValue<'a>>,
+    {
+        let fields = fields.map(Into::into);
         #[cfg(debug_assertions)]
         let fields = {
             let field_vec = fields.collect::<Vec<_>>();
@@ -894,7 +902,7 @@ impl ValueBuilder {
             if !field_value.is_null() || self.fields[idx].is_some() {
                 if let Some(field) = self.fields[idx].as_mut() {
                     field
-                        .push(field_value)
+                        .push_field(field_value)
                         .unwrap_or_else(|e| panic!("Failed to push field value: {e:?}"));
                 } else {
                     let mut mutable_vector =
@@ -908,7 +916,7 @@ impl ValueBuilder {
                         };
                     mutable_vector.push_nulls(num_rows - 1);
                     mutable_vector
-                        .push(field_value)
+                        .push_field(field_value)
                         .unwrap_or_else(|e| panic!("unexpected field value: {e:?}"));
                     self.fields[idx] = Some(mutable_vector);
                     MEMTABLE_ACTIVE_FIELD_BUILDER_COUNT.inc();

src/partition/src/splitter.rs

@@ -121,7 +121,7 @@ impl<'a> SplitReadRowHelper<'a> {
                             &row.values[*idx],
                             self.schema[*idx].datatype_extension.as_ref(),
                         )
-                        .into()
+                        .into_value()
                     })
                 })
                 .collect()

src/servers/src/http/event.rs

@@ -493,13 +493,12 @@ async fn dryrun_pipeline_inner(
                             .enumerate()
                             .map(|(idx, v)| {
                                 let mut map = Map::new();
-                                let value_ref = pb_value_to_value_ref(
+                                let value = pb_value_to_value_ref(
                                     &v,
                                     result_schema[idx].datatype_extension.as_ref(),
-                                );
-                                let greptime_value: datatypes::value::Value = value_ref.into();
-                                let serde_json_value =
-                                    serde_json::Value::try_from(greptime_value).unwrap();
+                                )
+                                .into_value();
+                                let serde_json_value = serde_json::Value::try_from(value).unwrap();
                                 map.insert("value".to_string(), serde_json_value);
                                 map.insert("key".to_string(), schema[idx][name_key].clone());
                                 map.insert(