Merge branch 'main' into desc

Author: mergify[bot]

docs/doc/13-monitor/jaeger.md

@@ -1,5 +1,5 @@
 ---
-title: Monitoring with Jaeger
+title: Jaeger
 ---
 
 [Jaeger](https://github.com/jaegertracing/jaeger) is an open-source, end-to-end distributed tracing tool that originated from [Uber](https://www.uber.com/). It helps monitor and troubleshoot microservices-based applications.
@@ -16,9 +16,9 @@ This tutorial uses the All In One image to deploy Jaeger in Docker. If you alrea
 docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 jaegertracing/all-in-one:latest
 ```
 
-### Step 2. Deploy Databend
+### Step 2. Set Environment Variables
 
-1. Set the following environment variables according to your actual tracing level requirements and Jaeger endpoint.
+Set the following environment variables according to your actual tracing level requirements and Jaeger endpoint.
     - `RUST_LOG`: Sets the log level.
     - `DATABEND_JAEGER_AGENT_ENDPOINT`: Sets the endpoint the Jaeger agent is listening on.
 
@@ -27,17 +27,19 @@ export RUST_LOG=DEBUG
 export DATABEND_JAEGER_AGENT_ENDPOINT=localhost:6831
 ```
 
-2. Follow the [Deployment Guide](https://databend.rs/doc/deploy) to deploy Databend.
+### Step 3. Deploy Databend
 
-3. Run the following SQL statements:
+1. Follow the [Deployment Guide](https://databend.rs/doc/deploy) to deploy Databend.
+
+2. Run the following SQL statements:
 
 ```sql
 CREATE TABLE t1(a INT);
 INSERT INTO t1 VALUES(1);
 INSERT INTO t1 SELECT * FROM t1;
 ```
 
-### Step 3. Check Tracing Information on Jaegar
+### Step 4. Check Tracing Information on Jaegar
 
 1. Go to <http://127.0.0.1:16686/> and select the **Search** tab.
 

docs/doc/13-monitor/sentry.md

@@ -1,48 +1,51 @@
 ---
-title: Monitoring with Sentry
+title: Sentry
 ---
 
-With automated error reporting, you can not only get alerted when something breaks, but also access the details you need to reproduce the issue.
+[Sentry](https://sentry.io/welcome/) is a developer-first error tracking and performance monitoring platform that helps developers see what actually matters, solve quicker, and learn continuously about their applications.
 
-## Sentry
+Databend provides integration with both Cloud and self-hosted Sentry solutions. The following tutorial walks you through the integration process.
 
-Databend provides integration with [Sentry](https://github.com/getsentry/sentry), a developer-first error tracking and performance monitoring platform.
+## Tutorial: Monitor Databend with Sentry
 
-### Deploy Sentry
+### Step 1. Deploy Sentry
 
-You can use Sentry as a cloud service by signing up at [sentry.io](https://sentry.io), or you can host it yourself by following the instructions at [Self-Hosted Sentry](https://develop.sentry.dev/self-hosted/).
+To deploy an on-premises Sentry, follow the instructions: https://develop.sentry.dev/self-hosted/
 
-<img src="/img/tracing/sentry-hosted.png"/>
+This tutorial uses the Sentry service on the cloud. To sign up an account for Cloud Sentry, go to https://sentry.io
 
-### Create a Project
+### Step 2. Create a Sentry Project
 
-To use Sentry with Databend, you need to create a project in Sentry and get its DSN (Data Source Name). The DSN is a unique identifier for your project that tells Sentry where to send your data. In this example, the DSN is `http://[email protected]:9000/5` 。
+Once you're logged into Sentry, create a Sentry project for the `Rust` platform to start. For how to create a project on Sentry, see https://docs.sentry.io/product/sentry-basics/integrate-frontend/create-new-project/
 
-<img src="/img/tracing/sentry-get-dsn.png"/>
+![Alt text](../../public/img/tracing/sentry-rust.png)
 
-### Start Databend
+### Step 3. Set Environment Variables
 
-You can start Databend with Sentry in two ways:
+1. Get the DSN (Data Source Name) of your project. For what DSN is and where to find it, see https://docs.sentry.io/product/sentry-basics/dsn-explainer/
 
-- **Enable Error Tracking Only**
+2. Set environment variables.
 
-  This option will only use the `sentry-log` feature, which will send error logs to Sentry.
+  - To enable the error-tracking feature, run the following commands:
 
-  ```bash
-  export DATABEND_SENTRY_DSN="<your-sentry-dsn>"
-  ```
+```bash
+export DATABEND_SENTRY_DSN="<your-DSN>"
+```
 
-  <img src="/img/tracing/sentry-error.png"/>
+  - To enable the performance monitoring feature, run the following commands:
 
-- **Also Enable Performance Monitoring**
+```bash
+export DATABEND_SENTRY_DSN="<your-DSN>"
+export SENTRY_TRACES_SAMPLE_RATE=1.0 LOG_LEVEL=DEBUG
+```
+:::tip
+Set `SENTRY_TRACES_SAMPLE_RATE` to a small value in production.
+:::
 
-  Setting `SENTRY_TRACES_SAMPLE_RATE` greater than `0.0` will allow sentry to perform trace sampling, which will help set up performance monitoring.
+### Step 4. Deploy Databend
 
-  ```bash
-  export DATABEND_SENTRY_DSN="<your-sentry-dsn>"
-  export SENTRY_TRACES_SAMPLE_RATE=1.0 LOG_LEVEL=DEBUG
-  ```
+Follow the [Deployment Guide](https://databend.rs/doc/deploy) to deploy Databend.
 
-  **Note:** Set `SENTRY_TRACES_SAMPLE_RATE` a to lower value in production.
+You're all set now. Check the pages on Sentry for alerts and performce information.
 
-  <img src="/img/tracing/sentry-performance.png"/>
+![Alt text](../../public/img/tracing/sentry-done.png)

docs/public/img/tracing/sentry-done.png

@@ -1869,6 +1869,15 @@ impl<KV: kvapi::KVApi<Error = MetaError>> SchemaApi for KV {
         let mut retry = 0;
         let table_id = req.table_id;
 
+        let mut keys = Vec::with_capacity(req.file_info.len());
+        for file in req.file_info.iter() {
+            let key = TableCopiedFileNameIdent {
+                table_id,
+                file: file.0.clone(),
+            };
+            keys.push(key.to_string_key());
+        }
+
         while retry < TXN_MAX_RETRY_TIMES {
             retry += 1;
 
@@ -1900,25 +1909,32 @@ impl<KV: kvapi::KVApi<Error = MetaError>> SchemaApi for KV {
             // So now, in case that `TableCopiedFileInfo` has expire time, remove `TableCopiedFileLockKey`
             // in each function. In this case there is chance that some `TableCopiedFileInfo` may not be
             // removed in `remove_table_copied_files`, but these data can be purged in case of expire time.
-            for (file, file_info) in req.file_info.iter() {
-                let key = TableCopiedFileNameIdent {
-                    table_id,
-                    file: file.to_owned(),
-                };
-                let (file_seq, _): (_, Option<TableCopiedFileInfo>) =
-                    get_pb_value(self, &key).await?;
-
-                condition.push(txn_cond_seq(&key, Eq, file_seq));
-                match &req.expire_at {
-                    Some(expire_at) => {
-                        if_then.push(txn_op_put_with_expire(
-                            &key,
-                            serialize_struct(file_info)?,
-                            *expire_at,
-                        ));
-                    }
-                    None => {
-                        if_then.push(txn_op_put(&key, serialize_struct(file_info)?));
+
+            let mut file_name_infos = req.file_info.clone().into_iter();
+
+            for c in keys.chunks(DEFAULT_MGET_SIZE) {
+                let seq_infos: Vec<(u64, Option<TableCopiedFileInfo>)> =
+                    mget_pb_values(self, c).await?;
+
+                for (file_seq, _file_info_opt) in seq_infos {
+                    let (f_name, file_info) = file_name_infos.next().unwrap();
+
+                    let key = TableCopiedFileNameIdent {
+                        table_id,
+                        file: f_name.to_owned(),
+                    };
+                    condition.push(txn_cond_seq(&key, Eq, file_seq));
+                    match &req.expire_at {
+                        Some(expire_at) => {
+                            if_then.push(txn_op_put_with_expire(
+                                &key,
+                                serialize_struct(&file_info)?,
+                                *expire_at,
+                            ));
+                        }
+                        None => {
+                            if_then.push(txn_op_put(&key, serialize_struct(&file_info)?));
+                        }
                     }
                 }
             }

docs/public/img/tracing/sentry-rust.png

@@ -34,6 +34,7 @@ use common_arrow::parquet::metadata::ColumnChunkMetaData;
 use common_arrow::parquet::metadata::RowGroupMetaData;
 use common_arrow::parquet::read::read_metadata;
 use common_arrow::read_columns_async;
+use common_catalog::plan::StageFileInfo;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_expression::DataBlock;
@@ -70,23 +71,28 @@ fn col_offset(meta: &ColumnChunkMetaData) -> i64 {
 impl InputFormat for InputFormatParquet {
     async fn get_splits(
         &self,
-        files: &[String],
+        file_infos: Vec<StageFileInfo>,
         _stage_info: &StageInfo,
         op: &Operator,
         _settings: &Arc<Settings>,
     ) -> Result<Vec<Arc<SplitInfo>>> {
         let mut infos = vec![];
-        for path in files {
-            let size = op.stat(path).await?.content_length() as usize;
-            let mut reader = op.reader(path).await?;
+        let mut schema = None;
+        for info in file_infos {
+            let size = info.size as usize;
+            let path = info.path.clone();
+
+            let mut reader = op.reader(&path).await?;
             let mut file_meta = read_metadata_async(&mut reader).await?;
             let row_groups = mem::take(&mut file_meta.row_groups);
-            let infer_schema = infer_schema(&file_meta)?;
-            let fields = Arc::new(infer_schema.fields);
+            if schema.is_none() {
+                schema = Some(infer_schema(&file_meta)?);
+            }
+            let fields = Arc::new(schema.clone().unwrap().fields);
             let read_file_meta = Arc::new(FileMeta { fields });
 
             let file_info = Arc::new(FileInfo {
-                path: path.clone(),
+                path,
                 size,
                 num_splits: row_groups.len(),
                 compress_alg: None,

src/meta/api/src/schema_api_impl.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use common_catalog::plan::StageFileInfo;
 use common_exception::Result;
 use common_expression::TableSchemaRef;
 use common_meta_app::principal::StageInfo;
@@ -28,7 +29,7 @@ use crate::input_formats::SplitInfo;
 pub trait InputFormat: Send + Sync {
     async fn get_splits(
         &self,
-        files: &[String],
+        file_infos: Vec<StageFileInfo>,
         stage_info: &StageInfo,
         op: &Operator,
         settings: &Arc<Settings>,

src/query/pipeline/sources/src/input_formats/impls/input_format_parquet.rs

@@ -19,6 +19,7 @@ use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
+use common_catalog::plan::StageFileInfo;
 use common_compress::DecompressDecoder;
 use common_compress::DecompressState;
 use common_exception::ErrorCode;
@@ -286,17 +287,19 @@ impl<T: InputFormatTextBase> InputFormatPipe for InputFormatTextPipe<T> {
 impl<T: InputFormatTextBase> InputFormat for T {
     async fn get_splits(
         &self,
-        files: &[String],
+        file_infos: Vec<StageFileInfo>,
         stage_info: &StageInfo,
-        op: &Operator,
+        _op: &Operator,
         _settings: &Arc<Settings>,
     ) -> Result<Vec<Arc<SplitInfo>>> {
         let mut infos = vec![];
-        for path in files {
-            let size = op.stat(path).await?.content_length() as usize;
+        for info in file_infos {
+            let size = info.size as usize;
+            let path = info.path.clone();
+
             let compress_alg = InputContext::get_compression_alg_copy(
                 stage_info.file_format_options.compression,
-                path,
+                &path,
             )?;
             let split_size = stage_info.copy_options.split_size;
             if compress_alg.is_none() && T::is_splittable() && split_size > 0 {
@@ -310,7 +313,7 @@ impl<T: InputFormatTextBase> InputFormat for T {
                     split_size
                 );
                 let file = Arc::new(FileInfo {
-                    path: path.clone(),
+                    path,
                     size,
                     num_splits: split_offsets.len(),
                     compress_alg,
@@ -327,7 +330,7 @@ impl<T: InputFormatTextBase> InputFormat for T {
                 }
             } else {
                 let file = Arc::new(FileInfo {
-                    path: path.clone(),
+                    path,
                     size, // dummy
                     num_splits: 1,
                     compress_alg,