fix: support view scalars and preserve null structs in scalar-to-value conversion

Address PR review:
- Utf8View/BinaryView ScalarValues now convert like their non-view forms
  instead of failing row extraction for struct columns
- a null struct scalar converts to Value::Null so a null struct inside a
  list stays null in the serialized JSON

Signed-off-by: Ning Sun <sunning@greptime.com>

src/datatypes/src/value.rs

@@ -1152,12 +1152,13 @@ impl TryFrom<ScalarValue> for Value {
             ScalarValue::UInt16(u) => Value::from(u),
             ScalarValue::UInt32(u) => Value::from(u),
             ScalarValue::UInt64(u) => Value::from(u),
-            ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) => {
+            ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
                 Value::from(s.map(StringBytes::from))
             }
             ScalarValue::Binary(b)
             | ScalarValue::LargeBinary(b)
-            | ScalarValue::FixedSizeBinary(_, b) => Value::from(b.map(Bytes::from)),
+            | ScalarValue::FixedSizeBinary(_, b)
+            | ScalarValue::BinaryView(b) => Value::from(b.map(Bytes::from)),
             ScalarValue::List(array) => {
                 // this is for item type
                 let datatype = ConcreteDataType::try_from(&array.value_type())?;
@@ -1221,6 +1222,12 @@ impl TryFrom<ScalarValue> for Value {
                 .map(|v| Value::Decimal128(Decimal128::new(v, p, s)))
                 .unwrap_or(Value::Null),
             ScalarValue::Struct(struct_array) => {
+                // A struct scalar carries a single element; a null (or empty)
+                // element means the struct itself is null, for example a null
+                // struct inside a list.
+                if struct_array.is_empty() || struct_array.is_null(0) {
+                    return Ok(Value::Null);
+                }
                 let struct_type = StructType::from(struct_array.fields());
                 let items = struct_array
                     .columns()
@@ -1244,8 +1251,6 @@ impl TryFrom<ScalarValue> for Value {
             | ScalarValue::LargeListView(_)
             | ScalarValue::Union(_, _, _)
             | ScalarValue::Float16(_)
-            | ScalarValue::Utf8View(_)
-            | ScalarValue::BinaryView(_)
             | ScalarValue::Map(_)
             | ScalarValue::Date64(_)
             | ScalarValue::RunEndEncoded(_, _, _) => {
@@ -2157,6 +2162,29 @@ pub(crate) mod tests {
             Value::Struct(struct_value),
             scalar_struct_value.try_into().unwrap()
         );
+
+        // view-typed strings and binaries convert like their non-view forms
+        assert_eq!(
+            Value::String("abc".into()),
+            ScalarValue::Utf8View(Some("abc".into()))
+                .try_into()
+                .unwrap()
+        );
+        assert_eq!(Value::Null, ScalarValue::Utf8View(None).try_into().unwrap());
+        assert_eq!(
+            Value::Binary(Bytes::from(vec![1, 2])),
+            ScalarValue::BinaryView(Some(vec![1, 2]))
+                .try_into()
+                .unwrap()
+        );
+        assert_eq!(
+            Value::Null,
+            ScalarValue::BinaryView(None).try_into().unwrap()
+        );
+
+        // a null struct scalar is a null value, not a struct of null fields
+        let null_struct = ScalarStructBuilder::new_null(build_struct_type().as_arrow_fields());
+        assert_eq!(Value::Null, null_struct.try_into().unwrap());
     }
 
     #[test]

tests-integration/tests/sql.rs

@@ -96,6 +96,7 @@ macro_rules! sql_tests {
                 test_mysql_prepare_tql_and_show,
                 test_postgres_extended_query_row_returning_statements,
                 test_postgres_native_histogram,
+                test_postgres_struct_types,
                 test_declare_fetch_close_cursor,
                 test_alter_update_on,
             );
@@ -1041,6 +1042,55 @@ pub async fn test_postgres_native_histogram(store_type: StorageType) {
     guard.remove_all().await;
 }
 
+pub async fn test_postgres_struct_types(store_type: StorageType) {
+    let (mut guard, fe_pg_server) = setup_pg_server(store_type, "test_postgres_struct_types").await;
+    let addr = fe_pg_server.bind_addr().unwrap().to_string();
+
+    let (client, connection) = tokio_postgres::connect(&format!("postgres://{addr}/public"), NoTls)
+        .await
+        .unwrap();
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    tokio::spawn(async move {
+        connection.await.unwrap();
+        tx.send(()).unwrap();
+    });
+
+    async fn query_one(client: &Client, sql: &str) -> String {
+        let messages = client.simple_query(sql).await.unwrap();
+        messages
+            .iter()
+            .filter_map(|message| match message {
+                SimpleQueryMessage::Row(row) => row.get(0).map(|v| v.to_string()),
+                _ => None,
+            })
+            .next()
+            .unwrap()
+    }
+
+    // Struct and list-of-struct columns are serialized as JSON. View-typed
+    // struct fields must not fail row extraction, and a null struct inside a
+    // list stays null instead of becoming a struct of null fields.
+    let row = query_one(&client, "SELECT struct(arrow_cast('abc', 'Utf8View'))").await;
+    assert_eq!(row, "{\"c0\":\"abc\"}");
+
+    // '0102' casts to its UTF-8 bytes when interpreted as BinaryView.
+    let row = query_one(&client, "SELECT struct(arrow_cast('0102', 'BinaryView'))").await;
+    assert_eq!(row, "{\"c0\":[48,49,48,50]}");
+
+    let row = query_one(&client, "SELECT struct([struct(1), NULL])").await;
+    assert_eq!(row, "{\"c0\":[{\"c0\":1},null]}");
+
+    // Top-level control: the same null element in a bare list stays null too.
+    let row = query_one(&client, "SELECT [struct(1), NULL]").await;
+    assert_eq!(row, "[{\"c0\":1},null]");
+
+    drop(client);
+    rx.await.unwrap();
+
+    let _ = fe_pg_server.shutdown().await;
+    guard.remove_all().await;
+}
+
 pub async fn test_postgres_slow_query(store_type: StorageType) {
     let (mut guard, fe_pg_server) = setup_pg_server_with_slow_query_threshold(
         store_type,