fix(pgwire): process empty query correctly (risingwavelabs#8535)

Author: ZENOTME

src/utils/pgwire/src/pg_protocol.rs

@@ -394,7 +394,12 @@ where
 
     async fn process_parse_msg(&mut self, msg: FeParseMessage) -> PsqlResult<()> {
         let sql = cstr_to_str(&msg.sql_bytes).unwrap();
-        tracing::trace!("(extended query)parse query: {}", sql);
+        let statement_name = cstr_to_str(&msg.statement_name).unwrap().to_string();
+        tracing::trace!(
+            "(extended query)parse query: {}, statement name: {}",
+            sql,
+            statement_name
+        );
 
         let is_query_sql = {
             let stmts = Parser::parse_sql(sql)
@@ -407,8 +412,12 @@ where
                 ));
             }
 
-            StatementType::infer_from_statement(&stmts[0])
-                .map_or(false, |stmt_type| stmt_type.is_query())
+            if stmts.is_empty() {
+                false
+            } else {
+                StatementType::infer_from_statement(&stmts[0])
+                    .map_or(false, |stmt_type| stmt_type.is_query())
+            }
         };
 
         let prepared_statement = PreparedStatement::parse_statement(sql.to_string(), msg.type_ids)?;
@@ -426,12 +435,7 @@ where
             vec![]
         };
 
-        let statement = PgStatement::new(
-            cstr_to_str(&msg.statement_name).unwrap().to_string(),
-            prepared_statement,
-            fields,
-            is_query_sql,
-        );
+        let statement = PgStatement::new(statement_name, prepared_statement, fields, is_query_sql);
 
         let name = statement.name();
         if name.is_empty() {
@@ -445,11 +449,12 @@ where
 
     fn process_bind_msg(&mut self, msg: FeBindMessage) -> PsqlResult<()> {
         let statement_name = cstr_to_str(&msg.statement_name).unwrap().to_string();
+        let portal_name = cstr_to_str(&msg.portal_name).unwrap().to_string();
         // 1. Get statement.
         trace!(
             target: "pgwire_query_log",
-            "(extended query)bind: get statement name: {}",
-            &statement_name
+            "(extended query)bind: statement name: {}, portal name: {}",
+            &statement_name,&portal_name
         );
         let statement = if statement_name.is_empty() {
             self.unnamed_statement
@@ -473,7 +478,6 @@ where
             .try_collect()?;
 
         // 2. Instance the statement to get the portal.
-        let portal_name = cstr_to_str(&msg.portal_name).unwrap().to_string();
         let portal = statement.instance(
             portal_name.clone(),
             &msg.params,
@@ -505,7 +509,7 @@ where
                 .ok_or_else(PsqlError::no_portal)?
         };
 
-        tracing::trace!(target: "pgwire_query_log", "(extended query)execute query: {}", portal.query_string());
+        tracing::trace!(target: "pgwire_query_log", "(extended query)execute query: {}, portal name: {}", portal.query_string(),portal_name);
 
         // 2. Execute instance statement using portal.
         let session = self.session.clone().unwrap();

src/utils/pgwire/tests/js/test/pgwire.test.ts

@@ -11,8 +11,6 @@ describe('PgwireTest', () => {
     try {
       const conn = await pool.connect();
       try {
-        // FIXME: RisingWave currently will fail on this test due to the lacking support of prepared statement.
-        // Related issue: https://github.com/risingwavelabs/risingwave/issues/5293
         const res = await conn.query({
           text: 'SELECT $1::int AS number',
           values: ['1'],
@@ -25,4 +23,29 @@ describe('PgwireTest', () => {
       await pool.end();
     }
   });
+
+  test('empty query', async () => {
+    const pool = new Pool({
+      host: '127.0.0.1',
+      database: 'dev',
+      port: 4566,
+      user: 'root',
+    });
+    try {
+      const conn = await pool.connect();
+      try {
+        const query = {
+          name: 'empty',
+          text: '',
+          values: [],
+        };
+        const res = await conn.query(query);
+        expect(res.rowCount).toBe(null);
+      } finally {
+        await conn.release();
+      }
+    } finally {
+      await pool.end();
+    }
+  });
 });