feat(rio): rio_v2 is compatible with minio for storing data. (#3115)

* Set up a compatibility layer for replacing old Rio components with new ones.

* fix(rio). compress range

* feat(rio). Add the experimental feature rio_v2 to support minio data at the binary level.

* feat(rio_v2): add sse-c test

* test compression component

* simple fix

* fix minlz encode

* fix metadata

* fix kms key cache error

* Update launch.json

* ci: set nix crate download user agent

* fix: gate obs pyroscope backend

* ignore minio test

* fix encrypt check

* fix

* fix

* fix

* Update object_usecase.rs

* Update ci.yml

* fix

* ci add rio-v2 test

* fix

* ci fix

* fix

* Reconstructed into a more reasonable compatibility mode

* fix

* fix

---------

Signed-off-by: houseme <housemecn@gmail.com>
Signed-off-by: 唐小鸭 <tangtang1251@qq.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: cxymds <Cxymds@qq.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/s3select-api/src/object_store.rs

@@ -1046,7 +1046,10 @@ where
     AsyncTryStream::<Bytes, o_Error, _>::new(|mut y| async move {
         pin_mut!(stream);
         let mut remaining: usize = content_length;
-        while let Some(result) = stream.next().await {
+        while remaining > 0 {
+            let Some(result) = stream.next().await else {
+                break;
+            };
             let mut bytes = result.map_err(|e| o_Error::Generic {
                 store: "",
                 source: Box::new(e),
@@ -1064,12 +1067,12 @@ where
 #[cfg(test)]
 mod test {
     use super::{
-        ConvertStream, SelectScanRange, convert_field_delimiter_stream, extract_json_sub_path_from_expression, find_delimiter,
-        flatten_json_document_to_ndjson, http_range_spec_from_get_range, replace_symbol, scan_range_from_bounds,
+        ConvertStream, SelectScanRange, bytes_stream, convert_field_delimiter_stream, extract_json_sub_path_from_expression,
+        find_delimiter, flatten_json_document_to_ndjson, http_range_spec_from_get_range, replace_symbol, scan_range_from_bounds,
         scan_range_read_start, scan_range_stream, select_read_headers,
     };
     use bytes::Bytes;
-    use futures::{StreamExt, stream};
+    use futures::{StreamExt, TryStreamExt, stream};
     use object_store::GetRange;
     use s3s::dto::{
         CSVInput, CSVOutput, ExpressionType, InputSerialization, OutputSerialization, SelectObjectContentInput,
@@ -1079,6 +1082,10 @@ mod test {
         X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
         X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
     };
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
     use tokio::io::AsyncReadExt;
     use tokio_util::io::StreamReader;
 
@@ -1285,6 +1292,29 @@ mod test {
         assert_eq!(output, b"a,");
     }
 
+    #[tokio::test]
+    async fn test_bytes_stream_stops_at_content_length() {
+        let poll_count = Arc::new(AtomicUsize::new(0));
+        let stream_poll_count = Arc::clone(&poll_count);
+        let source = stream::unfold(0, move |index| {
+            let stream_poll_count = Arc::clone(&stream_poll_count);
+            async move {
+                stream_poll_count.fetch_add(1, Ordering::SeqCst);
+                let bytes = match index {
+                    0 => Bytes::from_static(b"abcd"),
+                    1 => Bytes::from_static(b"efgh"),
+                    _ => return None,
+                };
+                Some((Ok::<_, std::io::Error>(bytes), index + 1))
+            }
+        });
+
+        let chunks: Vec<Bytes> = bytes_stream(source, 4).try_collect().await.unwrap();
+
+        assert_eq!(chunks, vec![Bytes::from_static(b"abcd")]);
+        assert_eq!(poll_count.load(Ordering::SeqCst), 1);
+    }
+
     /// A JSON array is split into one NDJSON line per element.
     #[test]
     fn test_flatten_array_produces_one_line_per_element() {