diff --git a/src/api/src/helper.rs b/src/api/src/helper.rs index 1ff437f127..4d661a372b 100644 --- a/src/api/src/helper.rs +++ b/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> { + 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> for Value { + fn from(value: DecodedValue<'_>) -> Self { + value.into_value() + } +} + +impl<'a> From> 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::>(); - 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::>(); + .collect::>(); - 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::>(); - let fields = fields.into_iter().collect::>(); 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() + ); } } diff --git a/src/datatypes/src/json/value.rs b/src/datatypes/src/json/value.rs index b727bd8b5a..e85adbb8ca 100644 --- a/src/datatypes/src/json/value.rs +++ b/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(), diff --git a/src/datatypes/src/vectors.rs b/src/datatypes/src/vectors.rs index 5f116e0952..86f382f096 100644 --- a/src/datatypes/src/vectors.rs +++ b/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 diff --git a/src/datatypes/src/vectors/json/builder.rs b/src/datatypes/src/vectors/json/builder.rs index ebf11ab796..c8612aba28 100644 --- a/src/datatypes/src/vectors/json/builder.rs +++ b/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), diff --git a/src/flow/src/batching_mode/time_window.rs b/src/flow/src/batching_mode/time_window.rs index 1c8ebea2a8..751b53a57d 100644 --- a/src/flow/src/batching_mode/time_window.rs +++ b/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", })?; } diff --git a/src/flow/src/repr.rs b/src/flow/src/repr.rs index 715f60594b..f53599c944 100644 --- a/src/flow/src/repr.rs +++ b/src/flow/src/repr.rs @@ -194,7 +194,7 @@ impl From 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()), ) } } diff --git a/src/metric-engine/src/row_modifier.rs b/src/metric-engine/src/row_modifier.rs index 7d68f94aa0..c62f52cf8d 100644 --- a/src/metric-engine/src/row_modifier.rs +++ b/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"), ) }) } diff --git a/src/mito-codec/src/key_values.rs b/src/mito-codec/src/key_values.rs index d66110bacf..8e245b1ab6 100644 --- a/src/mito-codec/src/key_values.rs +++ b/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> { + /// Gets field columns, decoding JSON values into owned values. + pub fn fields(&self) -> impl Iterator> { 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); } } diff --git a/src/mito2/src/memtable/builder.rs b/src/mito2/src/memtable/builder.rs index 7e37077f3e..50eb06aaec 100644 --- a/src/mito2/src/memtable/builder.rs +++ b/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 { diff --git a/src/mito2/src/memtable/time_series.rs b/src/mito2/src/memtable/time_series.rs index f5b772b076..b925089ba9 100644 --- a/src/mito2/src/memtable/time_series.rs +++ b/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>, - ) -> usize { + values: impl Iterator, + ) -> usize + where + T: Into>, + { // + 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>, - ) -> usize { + fields: impl Iterator, + ) -> usize + where + T: Into>, + { + let fields = fields.map(Into::into); #[cfg(debug_assertions)] let fields = { let field_vec = fields.collect::>(); @@ -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(); diff --git a/src/partition/src/splitter.rs b/src/partition/src/splitter.rs index 176422a173..15493d156c 100644 --- a/src/partition/src/splitter.rs +++ b/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() diff --git a/src/servers/src/http/event.rs b/src/servers/src/http/event.rs index df96471157..18f0fa794f 100644 --- a/src/servers/src/http/event.rs +++ b/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(