merge upstream/main

cli/Cargo.toml

@@ -22,6 +22,8 @@ path = "src/bin/bendctl.rs"
 databend-query = {path = "../query"}
 common-base = {path = "../common/base" }
 common-datavalues = { path = "../common/datavalues" }
+common-tracing = {path = "../common/tracing" }
+
 itertools = "0.10.3"
 databend-meta = {path = "../metasrv" }
 common-meta-raft-store= {path = "../common/meta/raft-store"}
@@ -48,7 +50,6 @@ tar = "0.4.37"
 thiserror = "1.0.30"
 ureq = { version = "2.3.1", features = ["json"] }
 nix = "0.23.0"
-log = "0.4.14"
 serde_yaml = "0.8.21"
 structopt = "0.3.25"
 structopt-toml = "0.5.0"

cli/src/cmds/status.rs

@@ -27,10 +27,10 @@ use std::time;
 
 use async_trait::async_trait;
 use common_base::tokio::time::Duration;
+use common_tracing::tracing;
 use databend_meta::configs::Config as MetaConfig;
 use databend_query::configs::Config as QueryConfig;
 use libc::pid_t;
-use log::info;
 use nix::unistd::Pid;
 use reqwest::Client;
 use serde::Deserialize;
@@ -228,7 +228,7 @@ impl LocalRuntime for LocalDashboardConfig {
             .stdout(unsafe { Stdio::from_raw_fd(out_file.into_raw_fd()) })
             .stderr(unsafe { Stdio::from_raw_fd(err_file.into_raw_fd()) });
         // logging debug
-        info!("executing command {:?}", command);
+        tracing::info!("executing command {:?}", command);
         Ok(command)
     }
 
@@ -349,7 +349,7 @@ impl LocalRuntime for LocalMetaConfig {
             .stdout(unsafe { Stdio::from_raw_fd(out_file.into_raw_fd()) })
             .stderr(unsafe { Stdio::from_raw_fd(err_file.into_raw_fd()) });
         // logging debug
-        info!("executing command {:?}", command);
+        tracing::info!("executing command {:?}", command);
         Ok(command)
     }
 
@@ -519,7 +519,7 @@ impl LocalRuntime for LocalQueryConfig {
             .stdout(unsafe { Stdio::from_raw_fd(out_file.into_raw_fd()) })
             .stderr(unsafe { Stdio::from_raw_fd(err_file.into_raw_fd()) });
         // logging debug
-        info!("executing command {:?}", command);
+        tracing::info!("executing command {:?}", command);
         Ok(command)
     }
     fn set_pid(&mut self, id: pid_t) {
@@ -555,7 +555,7 @@ impl LocalRuntime for LocalQueryConfig {
 impl Status {
     pub fn read(conf: Config) -> Result<Self> {
         let status_path = format!("{}/.status.json", conf.databend_dir);
-        log::info!("{}", status_path.as_str());
+        tracing::info!("{}", status_path.as_str());
         let local_config_dir = format!("{}/configs/local", conf.databend_dir);
         std::fs::create_dir_all(local_config_dir.as_str())
             .expect("cannot create dir to store local profile");

common/cache/Cargo.toml

@@ -15,13 +15,13 @@ heapsize = ["heapsize_"]
 amortized = ["ritelinked/ahash-amortized", "ritelinked/inline-more-amortized"]
 
 [dependencies] # In alphabetical order
+common-tracing = {path = "../tracing" }
 # Workspace dependencies
 
 # Github dependencies
 
 # Crates.io dependencies
 filetime = "0.2.15"
-log = "0.4.14"
 ritelinked  = { version = "0.3.2", default-features = false, features = ["ahash", "inline-more"] }
 walkdir = "2.3.2"
 

common/cache/src/disk_cache.rs

@@ -23,6 +23,7 @@ use std::io::prelude::*;
 use std::path::Path;
 use std::path::PathBuf;
 
+use common_tracing::tracing;
 use filetime::set_file_times;
 use filetime::FileTime;
 use ritelinked::DefaultHashBuilder;
@@ -163,14 +164,15 @@ where
         for (file, size) in get_all_files(&self.root) {
             if !self.can_store(size) {
                 fs::remove_file(file).unwrap_or_else(|e| {
-                    error!(
+                    tracing::error!(
                         "Error removing file `{}` which is too large for the cache ({} bytes)",
-                        e, size
+                        e,
+                        size
                     )
                 });
             } else {
                 self.add_file(AddFile::AbsPath(file), size)
-                    .unwrap_or_else(|e| error!("Error adding file: {}", e));
+                    .unwrap_or_else(|e| tracing::error!("Error adding file: {}", e));
             }
         }
         Ok(self)
@@ -225,7 +227,7 @@ where
         let size = size.unwrap_or(fs::metadata(path)?.len());
         self.add_file(AddFile::RelPath(rel_path), size)
             .map_err(|e| {
-                error!(
+                tracing::error!(
                     "Failed to insert file `{}`: {}",
                     rel_path.to_string_lossy(),
                     e
@@ -259,10 +261,10 @@ where
         let size = fs::metadata(path.as_ref())?.len();
         self.insert_by(key, Some(size), |new_path| {
             fs::rename(path.as_ref(), new_path).or_else(|_| {
-                warn!("fs::rename failed, falling back to copy!");
+                tracing::warn!("fs::rename failed, falling back to copy!");
                 fs::copy(path.as_ref(), new_path)?;
                 fs::remove_file(path.as_ref()).unwrap_or_else(|e| {
-                    error!("Failed to remove original file in insert_file: {}", e)
+                    tracing::error!("Failed to remove original file in insert_file: {}", e)
                 });
                 Ok(())
             })
@@ -301,7 +303,7 @@ where
             Some(_) => {
                 let path = self.rel_to_abs_path(key.as_ref());
                 fs::remove_file(&path).map_err(|e| {
-                    error!("Error removing file from cache: `{:?}`: {}", path, e);
+                    tracing::error!("Error removing file from cache: `{:?}`: {}", path, e);
                     Into::into(e)
                 })
             }

common/cache/src/lib.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#[macro_use]
-extern crate log;
 #[cfg(feature = "heapsize")]
 #[cfg(not(target_os = "macos"))]
 extern crate heapsize_;

common/clickhouse-srv/Cargo.toml

@@ -18,6 +18,7 @@ tokio_io = ["tokio"]
 
 [dependencies]
 common-io = { path = "../io" }
+common-tracing = {path = "../tracing" }
 
 lazy_static = "1.4.0"
 thiserror = "1.0.30"

common/clickhouse-srv/examples/simple.rs

@@ -25,12 +25,11 @@ use common_clickhouse_srv::types::Block;
 use common_clickhouse_srv::types::Progress;
 use common_clickhouse_srv::CHContext;
 use common_clickhouse_srv::ClickHouseServer;
+use common_tracing::tracing;
 use futures::task::Context;
 use futures::task::Poll;
 use futures::Stream;
 use futures::StreamExt;
-use log::debug;
-use log::info;
 use tokio::net::TcpListener;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
@@ -46,7 +45,7 @@ async fn main() -> std::result::Result<(), Box<dyn Error>> {
     // Note that this is the Tokio TcpListener, which is fully async.
     let listener = TcpListener::bind(host_port).await?;
 
-    info!("Server start at {}", host_port);
+    tracing::info!("Server start at {}", host_port);
 
     loop {
         // Asynchronously wait for an inbound TcpStream.
@@ -76,7 +75,7 @@ struct Session {
 impl common_clickhouse_srv::ClickHouseSession for Session {
     async fn execute_query(&self, ctx: &mut CHContext, connection: &mut Connection) -> Result<()> {
         let query = ctx.state.query.clone();
-        debug!("Receive query {}", query);
+        tracing::debug!("Receive query {}", query);
 
         let start = Instant::now();
 
@@ -125,9 +124,10 @@ impl common_clickhouse_srv::ClickHouseSession for Session {
         }
 
         let duration = start.elapsed();
-        debug!(
+        tracing::debug!(
             "ClickHouseHandler executor cost:{:?}, statistics:{:?}",
-            duration, "xxx",
+            duration,
+            "xxx",
         );
         Ok(())
     }

common/clickhouse-srv/src/lib.rs

@@ -14,8 +14,8 @@
 
 use std::sync::Arc;
 
+use common_tracing::tracing;
 use errors::Result;
-use log::debug;
 use protocols::Stage;
 use tokio::net::TcpStream;
 use tokio::sync::mpsc::Sender;
@@ -145,7 +145,7 @@ impl ClickHouseServer {
     }
 
     async fn run(&mut self, session: Arc<dyn ClickHouseSession>, stream: TcpStream) -> Result<()> {
-        debug!("Handle New session");
+        tracing::debug!("Handle New session");
         let tz = session.timezone().to_string();
         let mut ctx = CHContext::new(QueryState::default());
         let mut connection = Connection::new(stream, session, tz)?;
@@ -164,7 +164,7 @@ impl ClickHouseServer {
                     return Err(e);
                 }
                 Ok(None) => {
-                    debug!("{:?}", "none data reset");
+                    tracing::debug!("{:?}", "none data reset");
                     ctx.state.reset();
                     return Ok(());
                 }

common/datavalues/src/series/series_impl.rs

@@ -258,6 +258,7 @@ impl_from!([i64], DFInt64Array, new_from_slice);
 impl_from!([f32], DFFloat32Array, new_from_slice);
 impl_from!([f64], DFFloat64Array, new_from_slice);
 impl_from!([Vec<u8>], DFStringArray, new_from_slice);
+impl_from!([String], DFStringArray, new_from_slice);
 
 impl_from!([Option<bool>], DFBooleanArray, new_from_opt_slice);
 impl_from!([Option<u8>], DFUInt8Array, new_from_opt_slice);

common/exception/src/exception.rs

@@ -188,6 +188,7 @@ build_exceptions! {
     SHA1CheckFailed(57),
     UnknownColumn(58),
     InvalidSourceFormat(59),
+    StrParseError(60),
 
     // uncategorized
     UnexpectedResponseType(600),

common/exception/tests/it/exception.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_exception::ErrorCode;
+use common_exception::SerializedError;
 use tonic::Code;
 use tonic::Status;
 
@@ -88,6 +90,18 @@ fn test_derive_from_display() {
     );
 }
 
+#[test]
+fn test_from_and_to_serialized_error() {
+    let ec = ErrorCode::UnknownDatabase("foo");
+    let se: SerializedError = ec.clone().into();
+
+    let ec2: ErrorCode = se.into();
+    assert_eq!(ec.code(), ec2.code());
+    assert_eq!(ec.message(), ec2.message());
+    assert_eq!(format!("{}", ec), format!("{}", ec2));
+    assert_eq!(ec.backtrace_str(), ec2.backtrace_str());
+}
+
 #[test]
 fn test_from_and_to_status() -> anyhow::Result<()> {
     use common_exception::exception::*;

common/flight-rpc/Cargo.toml

@@ -15,13 +15,13 @@ test = false
 common-arrow = {path = "../arrow"}
 common-base = {path = "../base" }
 common-exception= {path = "../exception"}
+common-tracing = {path = "../tracing" }
 
 # Github dependencies
 
 # Crates.io dependencies
 futures = "0.3.18"
 jwt-simple = "0.10.7"
-log = "0.4.14"
 prost = "0.9.0"
 serde = { version = "1.0.131", features = ["derive"] }
 serde_json = "1.0.72"

common/flight-rpc/src/dns_resolver.rs

@@ -25,6 +25,7 @@ use common_base::tokio;
 use common_base::tokio::task::JoinHandle;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_tracing::tracing;
 use hyper::client::connect::dns::Name;
 use hyper::client::HttpConnector;
 use hyper::service::Service;
@@ -157,7 +158,7 @@ impl ConnectionFactory {
                 let builder = Channel::builder(uri.clone());
 
                 let mut endpoint = if let Some(conf) = rpc_client_config {
-                    log::info!("tls rpc enabled");
+                    tracing::info!("tls rpc enabled");
                     let client_tls_config = Self::client_tls_config(&conf).map_err(|e| {
                         ErrorCode::TLSConfigurationFailure(format!(
                             "loading client tls config failure: {} ",

common/io/Cargo.toml

@@ -15,6 +15,7 @@ test = false
 [dependencies]
 common-exception= {path = "../exception"}
 bytes = "1.1.0"
+serde = { version = "1.0.130", features = ["derive"] }
 
 [dev-dependencies]
 rand = "0.8.4"

common/io/src/lib.rs

@@ -20,6 +20,7 @@ mod binary_ser;
 mod binary_write;
 mod buf_read;
 mod marshal;
+mod options_deserializer;
 mod stat_buffer;
 mod unmarshal;
 mod utils;