Merge branch 'main' into bump-opendal

Author: mergify[bot]

src/meta/client/src/grpc_action.rs

@@ -45,20 +45,6 @@ pub trait RequestFor {
     type Reply;
 }
 
-// TODO: reduce this and MetaGrpcReadReq into one enum?
-// Action wrapper for do_action.
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
-pub enum MetaGrpcWriteReq {
-    UpsertKV(UpsertKVReq),
-}
-
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
-pub enum MetaGrpcReadReq {
-    GetKV(GetKVReq),
-    MGetKV(MGetKVReq),
-    ListKV(ListKVReq), // since 2022-05-23
-}
-
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
 pub enum MetaGrpcReq {
     UpsertKV(UpsertKVReq),
@@ -81,15 +67,6 @@ impl TryInto<MetaGrpcReq> for Request<RaftRequest> {
     }
 }
 
-impl tonic::IntoRequest<RaftRequest> for MetaGrpcWriteReq {
-    fn into_request(self) -> Request<RaftRequest> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(&self).expect("fail to serialize"),
-        };
-        tonic::Request::new(raft_request)
-    }
-}
-
 impl TryInto<Request<RaftRequest>> for MetaGrpcReq {
     type Error = serde_json::Error;
 
@@ -103,32 +80,6 @@ impl TryInto<Request<RaftRequest>> for MetaGrpcReq {
     }
 }
 
-impl TryInto<Request<RaftRequest>> for MetaGrpcWriteReq {
-    type Error = serde_json::Error;
-
-    fn try_into(self) -> Result<Request<RaftRequest>, Self::Error> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(&self)?,
-        };
-
-        let request = tonic::Request::new(raft_request);
-        Ok(request)
-    }
-}
-
-impl TryInto<Request<RaftRequest>> for MetaGrpcReadReq {
-    type Error = serde_json::Error;
-
-    fn try_into(self) -> Result<Request<RaftRequest>, Self::Error> {
-        let get_req = RaftRequest {
-            data: serde_json::to_string(&self)?,
-        };
-
-        let request = tonic::Request::new(get_req);
-        Ok(request)
-    }
-}
-
 impl RequestFor for GetKVReq {
     type Reply = GetKVReply;
 }

src/meta/client/src/grpc_client.rs

@@ -715,21 +715,22 @@ impl MetaGrpcClient {
         let resp =
             res.map_err(|status| MetaHandshakeError::new("handshake is refused", &status))?;
 
-        // backward compatibility: no version in handshake.
-        // TODO(xp): remove this when merged.
-        if resp.protocol_version > 0 {
-            let min_compatible = to_digit_ver(min_metasrv_ver);
-            if resp.protocol_version < min_compatible {
-                let invalid_err = AnyError::error(format!(
-                    "metasrv protocol_version({}) < meta-client min-compatible({})",
-                    from_digit_ver(resp.protocol_version),
-                    min_metasrv_ver,
-                ));
-                return Err(MetaHandshakeError::new(
-                    "incompatible protocol version",
-                    &invalid_err,
-                ));
-            }
+        assert!(
+            resp.protocol_version > 0,
+            "talking to a very old databend-meta: upgrade databend-meta to at least 0.8"
+        );
+
+        let min_compatible = to_digit_ver(min_metasrv_ver);
+        if resp.protocol_version < min_compatible {
+            let invalid_err = AnyError::error(format!(
+                "metasrv protocol_version({}) < meta-client min-compatible({})",
+                from_digit_ver(resp.protocol_version),
+                min_metasrv_ver,
+            ));
+            return Err(MetaHandshakeError::new(
+                "incompatible protocol version",
+                &invalid_err,
+            ));
         }
 
         let token = resp.payload;

src/meta/client/src/lib.rs

@@ -17,9 +17,7 @@ mod grpc_client;
 mod kv_api_impl;
 mod message;
 
-pub use grpc_action::MetaGrpcReadReq;
 pub use grpc_action::MetaGrpcReq;
-pub use grpc_action::MetaGrpcWriteReq;
 pub use grpc_action::RequestFor;
 pub use grpc_client::ClientHandle;
 pub use grpc_client::MetaGrpcClient;

src/meta/client/tests/it/grpc_server.rs

@@ -17,6 +17,8 @@ use std::thread::sleep;
 use std::time::Duration;
 
 use common_base::base::tokio;
+use common_meta_client::to_digit_ver;
+use common_meta_client::MIN_METASRV_SEMVER;
 use common_meta_types::protobuf::meta_service_server::MetaService;
 use common_meta_types::protobuf::meta_service_server::MetaServiceServer;
 use common_meta_types::protobuf::ClientInfo;
@@ -51,7 +53,12 @@ impl MetaService for GrpcServiceForTestImpl {
         _request: Request<Streaming<common_meta_types::protobuf::HandshakeRequest>>,
     ) -> Result<Response<Self::HandshakeStream>, Status> {
         tokio::time::sleep(Duration::from_secs(2)).await;
-        let output = futures::stream::once(async { Ok(HandshakeResponse::default()) });
+        let output = futures::stream::once(async {
+            Ok(HandshakeResponse {
+                protocol_version: to_digit_ver(&MIN_METASRV_SEMVER),
+                payload: vec![],
+            })
+        });
         Ok(Response::new(Box::pin(output)))
     }
 

src/query/service/src/pipelines/processors/transforms/hash_join/common.rs

@@ -154,6 +154,10 @@ impl JoinHashTable {
             Ok(ConstColumn::new(inner, col.len()).arc())
         } else if column.is_nullable() {
             let col: &NullableColumn = Series::check_get(column)?;
+            if col.is_empty() {
+                let ty = col.data_type();
+                return ty.create_constant_column(&DataValue::Null, validity.len());
+            }
             // It's possible validity is longer than col.
             let diff_len = validity.len() - col.ensure_validity().len();
             let mut new_validity = MutableBitmap::with_capacity(validity.len());

tests/logictest/suites/mode/cluster/broadcast_join.test

@@ -469,5 +469,45 @@ drop table z0;
 statement ok
 drop table z1;
 
+statement ok
+CREATE TABLE t0(c0BOOLEAN BOOLEAN NULL DEFAULT(false));
+
+statement ok
+CREATE TABLE t1(c0BOOLEAN BOOL NULL, c1FLOAT FLOAT NOT NULL DEFAULT(0.4661566913127899));
+
+statement ok
+CREATE TABLE t2(c0VARCHAR VARCHAR NULL, c1FLOAT DOUBLE NULL DEFAULT(0.954969048500061), c2VARCHAR VARCHAR NULL);
+
+statement ok
+INSERT INTO t0(c0boolean) VALUES (false), (true);
+
+statement ok
+INSERT INTO t0(c0boolean) VALUES (false), (false), (true);
+
+statement ok
+INSERT INTO t1(c1float) VALUES (0.43919482827186584);
+
+statement ok
+INSERT INTO t1(c1float) VALUES (0.2492278516292572);
+
+statement ok
+INSERT INTO t2(c1float) VALUES (0.9702655076980591);
+
+statement ok
+INSERT INTO t2(c1float, c2varchar) VALUES (0.5340723991394043, '02'), (0.4661566913127899, '1261837');
+
+statement ok
+SELECT t0.c0boolean, t1.c0boolean, t1.c1float FROM t0, t1 RIGHT JOIN t2 ON t1.c0boolean;
+
+
+statement ok
+drop table t0;
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;
+
 statement ok
 set prefer_broadcast_join = 0;

tests/logictest/suites/query/join.test

@@ -476,3 +476,43 @@ drop table z0;
 statement ok
 drop table z1;
 
+statement ok
+CREATE TABLE t0(c0BOOLEAN BOOLEAN NULL DEFAULT(false));
+
+statement ok
+CREATE TABLE t1(c0BOOLEAN BOOL NULL, c1FLOAT FLOAT NOT NULL DEFAULT(0.4661566913127899));
+
+statement ok
+CREATE TABLE t2(c0VARCHAR VARCHAR NULL, c1FLOAT DOUBLE NULL DEFAULT(0.954969048500061), c2VARCHAR VARCHAR NULL);
+
+statement ok
+INSERT INTO t0(c0boolean) VALUES (false), (true);
+
+statement ok
+INSERT INTO t0(c0boolean) VALUES (false), (false), (true);
+
+statement ok
+INSERT INTO t1(c1float) VALUES (0.43919482827186584);
+
+statement ok
+INSERT INTO t1(c1float) VALUES (0.2492278516292572);
+
+statement ok
+INSERT INTO t2(c1float) VALUES (0.9702655076980591);
+
+statement ok
+INSERT INTO t2(c1float, c2varchar) VALUES (0.5340723991394043, '02'), (0.4661566913127899, '1261837');
+
+statement ok
+SELECT t0.c0boolean, t1.c0boolean, t1.c1float FROM t0, t1 RIGHT JOIN t2 ON t1.c0boolean;
+
+
+statement ok
+drop table t0;
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;
+

tests/suites/1_stateful/01_load_v2/01_0004_streaming_variant_load.sh

@@ -32,8 +32,7 @@ if [ $? -ne 0 ]; then
 fi
 
 # load csv
-# todo(ariesdevil): change to new syntax when format_quote landing
-curl -H "insert_sql:insert into variant_test format Csv" -H "format_skip_header:0" -H 'format_field_delimiter: ,' -H 'format_record_delimiter: \n' -H "format_quote: \'" -F "upload=@/tmp/json_sample1.csv" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" > /dev/null 2>&1
+curl -H "insert_sql:insert into variant_test file_format = (type = 'CSV' skip_header = 0 field_delimiter = ',' record_delimiter = '\n' quote = '\'')" -F "upload=@/tmp/json_sample1.csv" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" > /dev/null 2>&1
 curl -H "insert_sql:insert into variant_test format Csv" -H "format_skip_header:0" -H 'format_field_delimiter: |' -H 'format_record_delimiter: \n' -H "format_quote: \'" -F "upload=@/tmp/json_sample2.csv" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" > /dev/null 2>&1
 echo "select * from variant_test order by Id asc;" | $MYSQL_CLIENT_CONNECT
 

tests/suites/1_stateful/05_formats/05_03_xml/05_03_01_xml_v1.sh

@@ -68,7 +68,6 @@ curl -sH "insert_sql:insert into test_xml file_format = (type = 'XML')" -F "uplo
 echo "select * from test_xml" | $MYSQL_CLIENT_CONNECT
 echo "truncate table test_xml" | $MYSQL_CLIENT_CONNECT
 
-# todo(ariesdevil): wait for row_tag landing
-curl -sH "insert_sql:insert into test_xml format XML" -F "upload=@/tmp/simple_v2.xml" -H "row_tag:'databend'" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" | grep -c "SUCCESS"
+curl -sH "insert_sql:insert into test_xml file_format = (type = 'XML' row_tag = 'databend')" -F "upload=@/tmp/simple_v2.xml" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" | grep -c "SUCCESS"
 echo "select * from test_xml" | $MYSQL_CLIENT_CONNECT
 echo "truncate table test_xml" | $MYSQL_CLIENT_CONNECT

tests/suites/1_stateful/05_formats/05_03_xml/05_03_02_xml_v2.sh

@@ -36,7 +36,6 @@ curl -sH "insert_sql:insert into test_xml file_format = (type = 'XML')" -F "uplo
 echo "select * from test_xml" | $MYSQL_CLIENT_CONNECT
 echo "truncate table test_xml" | $MYSQL_CLIENT_CONNECT
 
-# todo(ariesdevil): wait for row_tag landing
-curl -sH "insert_sql:insert into test_xml format XML" -F "upload=@/tmp/simple_v3.xml" -H "row_tag:'databend'" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" | grep -c "SUCCESS"
+curl -sH "insert_sql:insert into test_xml file_format = (type = 'XML' row_tag = 'databend')" -F "upload=@/tmp/simple_v3.xml" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" | grep -c "SUCCESS"
 echo "select * from test_xml" | $MYSQL_CLIENT_CONNECT
 echo "truncate table test_xml" | $MYSQL_CLIENT_CONNECT