Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/storage_fix_strategy_for_append_only
risingwavelabs · Jan 22, 2025 · 60c8c79 · 60c8c79
2 parents 7e6db5b + 1fa4f93
commit 60c8c79
Show file tree

Hide file tree

Showing 21 changed files with 52 additions and 50 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/docker/docker-compose-distributed.yml b/docker/docker-compose-distributed.yml
@@ -135,7 +135,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 5s
       retries: 5

diff --git a/docker/docker-compose-with-azblob.yml b/docker/docker-compose-with-azblob.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-gcs.yml b/docker/docker-compose-with-gcs.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-hdfs.yml b/docker/docker-compose-with-hdfs.yml
@@ -108,7 +108,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 5s
       retries: 5

diff --git a/docker/docker-compose-with-local-fs.yml b/docker/docker-compose-with-local-fs.yml
@@ -67,7 +67,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-obs.yml b/docker/docker-compose-with-obs.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-oss.yml b/docker/docker-compose-with-oss.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-s3.yml b/docker/docker-compose-with-s3.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose-with-sqlite.yml b/docker/docker-compose-with-sqlite.yml
@@ -68,7 +68,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -70,7 +70,7 @@ services:
         - CMD-SHELL
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
         - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
       interval: 1s
       timeout: 5s

diff --git a/integration_tests/iceberg-sink2/docker/hive/docker-compose.yml b/integration_tests/iceberg-sink2/docker/hive/docker-compose.yml
@@ -47,7 +47,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     environment:
@@ -113,4 +113,4 @@ volumes:
 
 networks:
   iceberg_net:
-    name: iceberg
+    name: iceberg
diff --git a/integration_tests/iceberg-sink2/docker/jdbc/docker-compose.yml b/integration_tests/iceberg-sink2/docker/jdbc/docker-compose.yml
@@ -48,7 +48,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     environment:
@@ -93,4 +93,4 @@ volumes:
 
 networks:
   iceberg_net:
-    name: iceberg
+    name: iceberg
diff --git a/integration_tests/iceberg-sink2/docker/rest/docker-compose.yml b/integration_tests/iceberg-sink2/docker/rest/docker-compose.yml
@@ -55,7 +55,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     networks:
@@ -96,4 +96,4 @@ volumes:
     external: false
 
 networks:
-  iceberg_net:
+  iceberg_net:
diff --git a/integration_tests/iceberg-sink2/docker/storage/docker-compose.yml b/integration_tests/iceberg-sink2/docker/storage/docker-compose.yml
@@ -31,7 +31,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     networks:
@@ -72,4 +72,4 @@ volumes:
     external: false
 
 networks:
-  iceberg_net:
+  iceberg_net:
diff --git a/integration_tests/iceberg-source/docker/hive/docker-compose.yml b/integration_tests/iceberg-source/docker/hive/docker-compose.yml
@@ -47,7 +47,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     environment:
@@ -113,4 +113,4 @@ volumes:
 
 networks:
   iceberg_net:
-    name: iceberg
+    name: iceberg
diff --git a/integration_tests/iceberg-source/docker/jdbc/docker-compose.yml b/integration_tests/iceberg-source/docker/jdbc/docker-compose.yml
@@ -48,7 +48,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     environment:
@@ -93,4 +93,4 @@ volumes:
 
 networks:
   iceberg_net:
-    name: iceberg
+    name: iceberg
diff --git a/integration_tests/iceberg-source/docker/rest/docker-compose.yml b/integration_tests/iceberg-source/docker/rest/docker-compose.yml
@@ -55,7 +55,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     networks:
@@ -96,4 +96,4 @@ volumes:
     external: false
 
 networks:
-  iceberg_net:
+  iceberg_net:
diff --git a/integration_tests/iceberg-source/docker/storage/docker-compose.yml b/integration_tests/iceberg-source/docker/storage/docker-compose.yml
@@ -31,7 +31,7 @@ services:
     healthcheck:
       test:
         - CMD-SHELL
-        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
       interval: 1s
       timeout: 30s
     networks:
@@ -72,4 +72,4 @@ volumes:
     external: false
 
 networks:
-  iceberg_net:
+  iceberg_net:
diff --git a/src/utils/pgwire/Cargo.toml b/src/utils/pgwire/Cargo.toml
@@ -19,6 +19,7 @@ jsonwebtoken = "9"
 openssl = "0.10.66"
 panic-message = "0.3"
 parking_lot = { workspace = true }
+peekable = { version = "0.2", features = ["tokio"] }
 postgres-types = { version = "0.2.6" }
 reqwest = "0.12.2"
 risingwave_common = { workspace = true }

diff --git a/src/utils/pgwire/src/pg_message.rs b/src/utils/pgwire/src/pg_message.rs
@@ -17,9 +17,10 @@ use std::ffi::CStr;
 use std::io::{Error, ErrorKind, IoSlice, Result, Write};
 
 use anyhow::anyhow;
-use byteorder::{BigEndian, ByteOrder, NetworkEndian};
+use byteorder::{BigEndian, ByteOrder};
 /// Part of code learned from <https://github.com/zenithdb/zenith/blob/main/zenith_utils/src/pq_proto.rs>.
 use bytes::{Buf, BufMut, Bytes, BytesMut};
+use peekable::tokio::AsyncPeekable;
 use tokio::io::{AsyncRead, AsyncReadExt};
 
 use crate::error_or_notice::ErrorOrNoticeMessage;
@@ -297,31 +298,18 @@ impl FeMessage {
 impl FeStartupMessage {
     /// Read startup message from the stream.
     pub async fn read(stream: &mut (impl AsyncRead + Unpin)) -> Result<FeMessage> {
-        let mut buffer1 = vec![0; 1];
-        let result = stream.read_exact(&mut buffer1).await;
-        let filled1 = match result {
-            Ok(n) => n,
-            Err(err) => {
-                // Detect whether it is a health check.
-                if err.kind() == ErrorKind::UnexpectedEof {
-                    return Ok(FeMessage::HealthCheck);
-                } else {
-                    return Err(err);
-                }
+        let mut stream = AsyncPeekable::new(stream);
+
+        if let Err(err) = stream.peek_exact(&mut [0; 1]).await {
+            // If the stream is empty, it can be a health check. Do not return error.
+            if err.kind() == ErrorKind::UnexpectedEof {
+                return Ok(FeMessage::HealthCheck);
+            } else {
+                return Err(err);
             }
-        };
-        assert_eq!(filled1, 1);
-
-        let mut buffer2 = vec![0; 3];
-        let filled2 = stream.read_exact(&mut buffer2).await?;
-        assert_eq!(filled2, 3);
-
-        let mut buffer3 = BytesMut::with_capacity(4);
-        buffer3.put_slice(&buffer1);
-        buffer3.put_slice(&buffer2);
-
-        let len = NetworkEndian::read_i32(&buffer3);
+        }
 
+        let len = stream.read_i32().await?;
         let protocol_num = stream.read_i32().await?;
         let payload_len = (len - 8) as usize;
         if payload_len >= isize::MAX as usize {