chore: add some logging on pgwire and frontend

src/frontend/src/session.rs

@@ -375,7 +375,10 @@ impl Session for SessionImpl {
         sql: &str,
     ) -> std::result::Result<PgResponse, Box<dyn std::error::Error + Send + Sync>> {
         // Parse sql.
-        let mut stmts = Parser::parse_sql(sql)?;
+        let mut stmts = Parser::parse_sql(sql).map_err(|e| {
+            tracing::error!("failed to parse sql:\n{}:\n{}", sql, e);
+            e
+        })?;
         // With pgwire, there would be at most 1 statement in the vec.
         assert!(stmts.len() <= 1);
         if stmts.is_empty() {
@@ -387,7 +390,10 @@ impl Session for SessionImpl {
             ));
         }
         let stmt = stmts.swap_remove(0);
-        let rsp = handle(self, stmt).await?;
+        let rsp = handle(self, stmt).await.map_err(|e| {
+            tracing::error!("failed to handle sql:\n{}:\n{}", sql, e);
+            e
+        })?;
         Ok(rsp)
     }
 }

src/utils/logging/src/lib.rs

@@ -47,6 +47,11 @@ fn configure_risingwave_targets_fmt(targets: filter::Targets) -> filter::Targets
         .with_target("risingwave_batch", Level::DEBUG)
         .with_target("risingwave_storage", Level::DEBUG)
         .with_target("risingwave_sqlparser", Level::INFO)
+        .with_target("risingwave_source", Level::INFO)
+        .with_target("risingwave_connector", Level::INFO)
+        .with_target("risingwave_frontend", Level::INFO)
+        .with_target("risingwave_meta", Level::INFO)
+        .with_target("pgwire", Level::ERROR)
         // disable events that are too verbose
         // if you want to enable any of them, find the target name and set it to `TRACE`
         // .with_target("events::stream::mview::scan", Level::TRACE)

src/utils/pgwire/src/error.rs

@@ -19,8 +19,6 @@ use thiserror::Error;
 pub enum PsqlError {
     #[error("Encode error {0}.")]
     CancelError(String),
-    #[error("Read sql error: {0}.")]
-    ReadError(String),
 }
 
 impl PsqlError {

src/utils/pgwire/src/pg_message.rs

@@ -19,7 +19,6 @@ use byteorder::{BigEndian, ByteOrder};
 use bytes::{BufMut, Bytes, BytesMut};
 use tokio::io::{AsyncRead, AsyncReadExt};
 
-use crate::error::PsqlError;
 use crate::pg_field_descriptor::PgFieldDescriptor;
 use crate::pg_response::StatementType;
 use crate::types::Row;
@@ -31,8 +30,6 @@ pub enum FeMessage {
     Query(FeQueryMessage),
     CancelQuery,
     Terminate,
-    /// For error in read function of `FeStartupMessage` and `FeMessage`.
-    ReadError(PsqlError),
 }
 
 pub struct FeStartupMessage {}
@@ -68,10 +65,10 @@ impl FeMessage {
         match val {
             b'Q' => Ok(FeMessage::Query(FeQueryMessage { sql_bytes })),
             b'X' => Ok(FeMessage::Terminate),
-            _ => Ok(FeMessage::ReadError(PsqlError::ReadError(format!(
-                "Unsupported tag of regular message: {}",
-                val
-            )))),
+            _ => Err(std::io::Error::new(
+                ErrorKind::InvalidInput,
+                format!("Unsupported tag of regular message: {}", val),
+            )),
         }
     }
 }
@@ -92,10 +89,13 @@ impl FeStartupMessage {
             80877103 => Ok(FeMessage::Ssl),
             // Cancel request code.
             80877102 => Ok(FeMessage::CancelQuery),
-            _ => Ok(FeMessage::ReadError(PsqlError::ReadError(format!(
-                "Unsupported protocol number in start up msg {:?}",
-                protocol_num
-            )))),
+            _ => Err(std::io::Error::new(
+                ErrorKind::InvalidInput,
+                format!(
+                    "Unsupported protocol number in start up msg {:?}",
+                    protocol_num
+                ),
+            )),
         }
     }
 }
@@ -118,8 +118,9 @@ pub enum BeMessage<'a> {
 
 #[derive(Debug)]
 pub enum BeParameterStatusMessage<'a> {
-    Encoding(&'a str),
+    ClientEncoding(&'a str),
     StandardConformingString(&'a str),
+    ServerVersion(&'a str),
 }
 
 #[derive(Debug)]
@@ -166,10 +167,11 @@ impl<'a> BeMessage<'a> {
             BeMessage::ParameterStatus(param) => {
                 use BeParameterStatusMessage::*;
                 let [name, value] = match param {
-                    Encoding(val) => [b"client_encoding", val.as_bytes()],
+                    ClientEncoding(val) => [b"client_encoding", val.as_bytes()],
                     StandardConformingString(val) => {
                         [b"standard_conforming_strings", val.as_bytes()]
                     }
+                    ServerVersion(val) => [b"server_version", val.as_bytes()],
                 };
 
                 // Parameter names and values are passed as null-terminated strings

src/utils/pgwire/src/pg_protocol.rs

@@ -75,13 +75,31 @@ where
     }
 
     async fn do_process(&mut self) -> Result<bool> {
-        let msg = self.read_message().await?;
+        let msg = match self.read_message().await {
+            Ok(msg) => msg,
+            Err(e) => {
+                if e.kind() == std::io::ErrorKind::UnexpectedEof {
+                    return Err(e);
+                }
+                tracing::error!("unable to read message: {}", e);
+                self.write_message_no_flush(&BeMessage::ErrorResponse(Box::new(e)))?;
+                self.write_message_no_flush(&BeMessage::ReadyForQuery)?;
+                return Ok(false);
+            }
+        };
         match msg {
             FeMessage::Ssl => {
-                self.write_message_no_flush(&BeMessage::EncryptionResponse)?;
+                self.write_message_no_flush(&BeMessage::EncryptionResponse)
+                    .map_err(|e| {
+                        tracing::error!("failed to handle ssl request: {}", e);
+                        e
+                    })?;
             }
             FeMessage::Startup(msg) => {
-                self.process_startup_msg(msg)?;
+                self.process_startup_msg(msg).map_err(|e| {
+                    tracing::error!("failed to set up pg session: {}", e);
+                    e
+                })?;
                 self.state = PgProtocolState::Regular;
             }
             FeMessage::Query(query_msg) => {
@@ -95,10 +113,6 @@ where
             FeMessage::Terminate => {
                 self.process_terminate();
             }
-            FeMessage::ReadError(err) => {
-                self.write_message_no_flush(&BeMessage::ErrorResponse(Box::new(err)))?;
-                self.write_message_no_flush(&BeMessage::ReadyForQuery)?;
-            }
         }
         self.flush().await?;
         Ok(false)
@@ -116,11 +130,14 @@ where
         self.session = Some(self.session_mgr.connect("dev").map_err(IoError::other)?);
         self.write_message_no_flush(&BeMessage::AuthenticationOk)?;
         self.write_message_no_flush(&BeMessage::ParameterStatus(
-            BeParameterStatusMessage::Encoding("utf8"),
+            BeParameterStatusMessage::ClientEncoding("utf8"),
         ))?;
         self.write_message_no_flush(&BeMessage::ParameterStatus(
             BeParameterStatusMessage::StandardConformingString("on"),
         ))?;
+        self.write_message_no_flush(&BeMessage::ParameterStatus(
+            BeParameterStatusMessage::ServerVersion("9.5.0"),
+        ))?;
         self.write_message_no_flush(&BeMessage::ReadyForQuery)?;
         Ok(())
     }

src/utils/pgwire/src/pg_server.rs

@@ -75,7 +75,7 @@ async fn pg_serve_conn(socket: TcpStream, session_mgr: Arc<dyn SessionManager>)
                 }
             }
             Err(e) => {
-                if matches!(e.kind(), ErrorKind::UnexpectedEof) {
+                if e.kind() == ErrorKind::UnexpectedEof {
                     break;
                 }
                 // Execution error should not break current connection.