Repository: risingwavelabs/risingwave (language: python)

Issue:
bug(frontend): pgwire binary response unimplemented for list and struct
### Describe the bug

pgwire binary response unimplemented for list and struct

### To Reproduce

```python
import psycopg

conn = psycopg.connect(host='localhost', port='4566', dbname='dev', user='root')
with conn.cursor() as cur:
	cur.execute("select Array[1::bigint, 2::bigint, 3::bigint]", binary=True)
	for row in cur:
		print(row)
```

### Expected behavior

No panic in frontend.

### Additional context

_No response_
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:
Loading binary for array here: https://github.com/psycopg/psycopg/blob/0d2ea7648336712988206ac1dac28f01be775413/psycopg/psycopg/types/array.py#L434
https://github.com/risingwavelabs/risingwave/issues/3676#issuecomment-1254826683

@ZENOTME Do we have plan to support them?
It's not easy to support multiple demision list so there is no work for that now.

https://github.com/MaterializeInc/materialize/blob/61a82dfd8432b094189c4f839f67ab738f0f3bc1/src/pgrepr/src/value.rs#L390-L421

We can try to implement a single dimension list first. BTW, rust-postgres seems also only support to single dimension list. 
We should avoid its panic first 😄
> single dimension list.

Let's support the single-dim array first. I think it's doable. At least we can eliminate the majority of error cases.
+1 for https://github.com/risingwavelabs/risingwave/issues/7949#issuecomment-1978126393 & https://github.com/risingwavelabs/risingwave/issues/7949#issuecomment-1432819897. Let's finish these first.


Fix patch:
diff --git a/src/common/src/types/postgres_type.rs b/src/common/src/types/postgres_type.rs
index d85f08ed59cc3..c84f3e19f3097 100644
--- a/src/common/src/types/postgres_type.rs
+++ b/src/common/src/types/postgres_type.rs
@@ -116,7 +116,7 @@ impl DataType {
                         )*
                         DataType::Int256 => 1302,
                         DataType::Serial => 1016,
-                        DataType::Struct(_) => -1,
+                        DataType::Struct(_) => 2287, // pseudo-type of array[struct] (see `pg_type.dat`)
                         DataType::List { .. } => unreachable!("Never reach here!"),
                         DataType::Map(_) => 1304,
                     }
@@ -125,8 +125,7 @@ impl DataType {
                     DataType::Int256 => 1301,
                     DataType::Map(_) => 1303,
                     // TODO: Support to give a new oid for custom struct type. #9434
-                    // 1043 is varchar
-                    DataType::Struct(_) => 1043,
+                    DataType::Struct(_) => 2249,  // pseudo-type of struct (see `pg_type.dat`)
                 }
             }
         }
diff --git a/src/common/src/types/to_binary.rs b/src/common/src/types/to_binary.rs
index 7c5e88dbc10cf..294f96bc70453 100644
--- a/src/common/src/types/to_binary.rs
+++ b/src/common/src/types/to_binary.rs
@@ -14,12 +14,13 @@
 
 use bytes::{BufMut, Bytes, BytesMut};
 use postgres_types::{ToSql, Type};
+use rw_iter_util::ZipEqFast;
 
 use super::{
     DataType, Date, Decimal, Interval, ScalarRefImpl, Serial, Time, Timestamp, Timestamptz, F32,
     F64,
 };
-use crate::array::ListRef;
+use crate::array::{ListRef, StructRef};
 use crate::error::NotImplemented;
 
 /// Error type for [`ToBinary`] trait.
@@ -116,6 +117,29 @@ impl ToBinary for ListRef<'_> {
     }
 }
 
+impl ToBinary for StructRef<'_> {
+    fn to_binary_with_type(&self, ty: &DataType) -> Result<Bytes> {
+        // Reference: Postgres code `src/backend/utils/adt/rowtypes.c`
+        // https://github.com/postgres/postgres/blob/a3699daea2026de324ed7cc7115c36d3499010d3/src/backend/utils/adt/rowtypes.c#L687
+        let mut buf = BytesMut::new();
+        buf.put_i32(ty.as_struct().len() as i32); // number of columns
+        for (datum, field_ty) in self.iter_fields_ref().zip_eq_fast(ty.as_struct().types()) {
+            buf.put_i32(field_ty.to_oid()); // column type
+            match datum {
+                None => {
+                    buf.put_i32(-1); // -1 length means a NULL
+                }
+                Some(value) => {
+                    let data = value.to_binary_with_type(field_ty)?;
+                    buf.put_i32(data.len() as i32); // Length of element
+                    buf.put(data);
+                }
+            }
+        }
+        Ok(buf.into())
+    }
+}
+
 impl ToBinary for ScalarRefImpl<'_> {
     fn to_binary_with_type(&self, ty: &DataType) -> Result<Bytes> {
         match self {
@@ -137,7 +161,8 @@ impl ToBinary for ScalarRefImpl<'_> {
             ScalarRefImpl::Bytea(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Jsonb(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::List(v) => v.to_binary_with_type(ty),
-            ScalarRefImpl::Struct(_) | ScalarRefImpl::Map(_) => {
+            ScalarRefImpl::Struct(v) => v.to_binary_with_type(ty),
+            ScalarRefImpl::Map(_) => {
                 bail_not_implemented!(
                     issue = 7949,
                     "the pgwire extended-mode encoding for {ty} is unsupported"


Test patch:
diff --git a/ci/scripts/run-e2e-test.sh b/ci/scripts/run-e2e-test.sh
index a8601fbb0ebef..e84ead4a81df3 100755
--- a/ci/scripts/run-e2e-test.sh
+++ b/ci/scripts/run-e2e-test.sh
@@ -109,6 +109,10 @@ sqllogictest -p 4566 -d dev './e2e_test/ttl/ttl.slt'
 sqllogictest -p 4566 -d dev './e2e_test/database/prepare.slt'
 sqllogictest -p 4566 -d test './e2e_test/database/test.slt'
 
+echo "--- e2e, $mode, python_client"
+python3 -m pip install --break-system-packages psycopg
+python3 ./e2e_test/python_client/main.py
+
 echo "--- e2e, $mode, subscription"
 python3 -m pip install --break-system-packages psycopg2-binary
 sqllogictest -p 4566 -d dev './e2e_test/subscription/check_sql_statement.slt'
diff --git a/e2e_test/python_client/main.py b/e2e_test/python_client/main.py
new file mode 100644
index 0000000000000..bb41ba6c38f34
--- /dev/null
+++ b/e2e_test/python_client/main.py
@@ -0,0 +1,19 @@
+import psycopg
+
+def test_psycopg_extended_mode():
+    conn = psycopg.connect(host='localhost', port='4566', dbname='dev', user='root')
+    with conn.cursor() as cur:
+        cur.execute("select Array[1::bigint, 2::bigint, 3::bigint]", binary=True)
+        assert cur.fetchone() == ([1, 2, 3],)
+
+        cur.execute("select Array['foo', null, 'bar']", binary=True)
+        assert cur.fetchone() == (['foo', None, 'bar'],)
+
+        cur.execute("select ROW('123 Main St', 'New York', '10001')", binary=True)
+        assert cur.fetchone() == (('123 Main St', 'New York', '10001'),)
+
+        cur.execute("select array[ROW('123 Main St', 'New York', '10001'), ROW('234 Main St', null, '10001')]", binary=True)
+        assert cur.fetchone() == ([('123 Main St', 'New York', '10001'), ('234 Main St', None, '10001')],)
+
+if __name__ == '__main__':
+    test_psycopg_extended_mode()