feat: support nullable api response data (#465)

cli/test.sh

@@ -32,7 +32,7 @@ case $TEST_HANDLER in
 	;;
 "http")
 	echo "==> Testing REST API handler"
-	export BENDSQL_DSN="databend+http://${DATABEND_USER}:${DATABEND_PASSWORD}@${DATABEND_HOST}:8000/?sslmode=disable&presign=on"
+	export BENDSQL_DSN="databend+http://${DATABEND_USER}:${DATABEND_PASSWORD}@${DATABEND_HOST}:8000/?sslmode=disable&presign=on&format_null_as_str=0"
 	;;
 *)
 	echo "Usage: $0 [flight|http]"

core/src/client.rs

@@ -505,16 +505,16 @@ impl APIClient {
             ));
         }
         // resp.data[0]: [ "PUT", "{\"host\":\"s3.us-east-2.amazonaws.com\"}", "https://s3.us-east-2.amazonaws.com/query-storage-xxxxx/tnxxxxx/stage/user/xxxx/xxx?" ]
-        let method = resp.data[0][0].clone();
+        let method = resp.data[0][0].clone().unwrap_or_default();
         if method != "PUT" {
             return Err(Error::Request(format!(
                 "Invalid method for presigned upload request: {}",
                 method
             )));
         }
         let headers: BTreeMap<String, String> =
-            serde_json::from_str(resp.data[0][1].clone().as_str())?;
-        let url = resp.data[0][2].clone();
+            serde_json::from_str(resp.data[0][1].clone().unwrap_or("{}".to_string()).as_str())?;
+        let url = resp.data[0][2].clone().unwrap_or_default();
         Ok(PresignedResponse {
             method,
             headers,

core/src/response.rs

@@ -58,7 +58,7 @@ pub struct QueryResponse {
     pub session_id: Option<String>,
     pub session: Option<SessionState>,
     pub schema: Vec<SchemaField>,
-    pub data: Vec<Vec<String>>,
+    pub data: Vec<Vec<Option<String>>>,
     pub state: String,
     pub error: Option<QueryError>,
     // make it optional for backward compatibility

core/tests/core/simple.rs

@@ -21,5 +21,5 @@ async fn select_simple() {
     let dsn = option_env!("TEST_DATABEND_DSN").unwrap_or(DEFAULT_DSN);
     let client = APIClient::new(dsn, None).await.unwrap();
     let resp = client.start_query("select 15532").await.unwrap();
-    assert_eq!(resp.data, [["15532"]]);
+    assert_eq!(resp.data, [[Some("15532".to_string())]]);
 }

core/tests/core/stage.rs

@@ -79,7 +79,16 @@ async fn insert_with_stage(presign: bool) {
         ["5", "Shenzhen", "55"],
         ["6", "Beijing", "99"],
     ];
-    assert_eq!(resp.data, expect);
+    let result = resp
+        .data
+        .into_iter()
+        .map(|row| {
+            row.into_iter()
+                .map(|v| v.unwrap_or_default())
+                .collect::<Vec<_>>()
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(result, expect);
 
     let sql = format!("DROP TABLE `{}`;", table);
     client.query(&sql).await.unwrap();

driver/src/rest_api.rs

@@ -237,7 +237,7 @@ type PageFut = Pin<Box<dyn Future<Output = Result<QueryResponse>> + Send>>;
 pub struct RestAPIRows {
     client: APIClient,
     schema: SchemaRef,
-    data: VecDeque<Vec<String>>,
+    data: VecDeque<Vec<Option<String>>>,
     stats: Option<ServerStats>,
     query_id: String,
     next_uri: Option<String>,
@@ -268,7 +268,7 @@ impl Stream for RestAPIRows {
             return Poll::Ready(Some(Ok(RowWithStats::Stats(ss))));
         }
         if let Some(row) = self.data.pop_front() {
-            let row = Row::try_from((self.schema.clone(), &row))?;
+            let row = Row::try_from((self.schema.clone(), row))?;
             return Poll::Ready(Some(Ok(RowWithStats::Row(row))));
         }
         match self.next_page {

sql/src/rows.rs

@@ -86,13 +86,14 @@ impl From<databend_client::response::QueryStats> for ServerStats {
 #[derive(Clone, Debug, Default)]
 pub struct Row(Vec<Value>);
 
-impl TryFrom<(SchemaRef, &Vec<String>)> for Row {
+impl TryFrom<(SchemaRef, Vec<Option<String>>)> for Row {
     type Error = Error;
 
-    fn try_from((schema, data): (SchemaRef, &Vec<String>)) -> Result<Self> {
+    fn try_from((schema, data): (SchemaRef, Vec<Option<String>>)) -> Result<Self> {
         let mut values: Vec<Value> = Vec::new();
         for (i, field) in schema.fields().iter().enumerate() {
-            values.push(Value::try_from((&field.data_type, data[i].as_str()))?);
+            let val: Option<&str> = data.get(i).and_then(|v| v.as_deref());
+            values.push(Value::try_from((&field.data_type, val))?);
         }
         Ok(Self(values))
     }

sql/src/value.rs

@@ -147,6 +147,23 @@ impl Value {
     }
 }
 
+impl TryFrom<(&DataType, Option<&str>)> for Value {
+    type Error = Error;
+
+    fn try_from((t, v): (&DataType, Option<&str>)) -> Result<Self> {
+        match v {
+            Some(v) => Self::try_from((t, v)),
+            None => match t {
+                DataType::Null => Ok(Self::Null),
+                DataType::Nullable(_) => Ok(Self::Null),
+                _ => Err(Error::InvalidResponse(
+                    "NULL value for non-nullable field".to_string(),
+                )),
+            },
+        }
+    }
+}
+
 impl TryFrom<(&DataType, &str)> for Value {
     type Error = Error;
 

tests/Makefile

@@ -15,7 +15,7 @@ test-core: up
 
 test-driver: up
 	cargo test --test driver
-	TEST_DATABEND_DSN=databend+flight://root:@localhost:8900/default?sslmode=disable cargo test --features flight-sql --test driver
+	TEST_DATABEND_DSN=databend+flight://root:@localhost:8900/default?sslmode=disable&format_null_as_str=0 cargo test --features flight-sql --test driver
 
 test-bendsql: up
 	cd .. && ./cli/test.sh http

tests/docker-compose.yaml

@@ -1,4 +1,3 @@
-version: "3"
 services:
   minio:
     image: docker.io/minio/minio
@@ -7,7 +6,7 @@ services:
     volumes:
       - ./data:/data
   databend:
-    image: docker.io/datafuselabs/databend
+    image: docker.io/datafuselabs/databend:nightly
     environment:
       - QUERY_STORAGE_TYPE=s3
       - QUERY_DATABEND_ENTERPRISE_LICENSE