Merge branch 'main' into ISSUE-7528/auto_extend_ts

Author: mergify[bot]

Cargo.lock

@@ -12,7 +12,7 @@ CREATE STAGE [ IF NOT EXISTS ] <internal_stage_name>
   [ FILE_FORMAT = ( { TYPE = { CSV | PARQUET } [ formatTypeOptions ] ) } ]
   [ COPY_OPTIONS = ( copyOptions ) ]
   [ COMMENT = '<string_literal>' ]
-  
+
 -- External stage
 CREATE STAGE [ IF NOT EXISTS ] <external_stage_name>
     externalStageParams
@@ -70,8 +70,8 @@ externalStageParams ::=
 ### formatTypeOptions
 ```
 formatTypeOptions ::=
-  RECORD_DELIMITER = '<character>' 
-  FIELD_DELIMITER = '<character>' 
+  RECORD_DELIMITER = '<character>'
+  FIELD_DELIMITER = '<character>'
   SKIP_HEADER = <integer>
 ```
 
@@ -85,11 +85,13 @@ formatTypeOptions ::=
 ```
 copyOptions ::=
   [ SIZE_LIMIT = <num> ]
+  [ PURGE = <bool> ]
 ```
 
 | Parameters  | Description | Required |
 | ----------- | ----------- | --- |
 | `SIZE_LIMIT = <num>` | Number (> 0) that specifies the maximum rows of data to be loaded for a given COPY statement. Default `0` | Optional |
+| `PURGE = <bool>` | True specifies that the command will purge the files in the stage if they are loaded successfully into table. Default `false` | Optional |
 
 
 ## Examples

docs/doc/30-reference/30-sql/00-ddl/40-stage/01-ddl-create-stage.md

@@ -56,8 +56,8 @@ externalLocation (for Amazon S3) ::=
 ### formatTypeOptions
 ```
 formatTypeOptions ::=
-  RECORD_DELIMITER = '<character>' 
-  FIELD_DELIMITER = '<character>' 
+  RECORD_DELIMITER = '<character>'
+  FIELD_DELIMITER = '<character>'
   SKIP_HEADER = <integer>
 ```
 

docs/doc/30-reference/30-sql/10-dml/dml-copy-into-location.md

@@ -97,8 +97,8 @@ A regular expression pattern string, enclosed in single quotes, specifying the f
 
 ```
 formatTypeOptions ::=
-  RECORD_DELIMITER = '<character>' 
-  FIELD_DELIMITER = '<character>' 
+  RECORD_DELIMITER = '<character>'
+  FIELD_DELIMITER = '<character>'
   SKIP_HEADER = <integer>
   COMPRESSION = AUTO | GZIP | BZ2 | BROTLI | ZSTD | DEFLATE | RAW_DEFLATE | NONE
 ```
@@ -129,7 +129,7 @@ Default: `NONE`
 
 Values:
 
-| Values        | Notes                                                           | 
+| Values        | Notes                                                           |
 |---------------|-----------------------------------------------------------------|
 | `AUTO`        | Auto detect compression via file extensions                     |
 | `GZIP`        |                                                                 |
@@ -145,11 +145,13 @@ Values:
 ```
 copyOptions ::=
   [ SIZE_LIMIT = <num> ]
+  [ PURGE = <bool> ]
 ```
 
 | Parameters  | Description | Required |
 | ----------- | ----------- | --- |
 | `SIZE_LIMIT = <num>` | Number (> 0) that specifies the maximum rows of data to be loaded for a given COPY statement. Default `0` | Optional |
+| `PURGE = <bool>` | True that specifies the command will purge the files in the stage if they are loaded successfully into table. Default `false` | Optional |
 
 ## Examples
 

docs/doc/30-reference/30-sql/10-dml/dml-copy-into-table.md

@@ -2,23 +2,71 @@
 title: system.query_log
 ---
 
-A read-only in-memory table stores all the query logs;
+A read-only in-memory table stores all the query logs.
 
-```sql
-SELECT * FROM system.query_log ORDER BY event_time DESC LIMIT 1\G
-*************************** 1. row ***************************
+
+## Columns
+
+```
+CREATE TABLE `query_log` (
+  `log_type` TINYINT,
+  `handler_type` VARCHAR,
+  `tenant_id` VARCHAR,
+  `cluster_id` VARCHAR,
+  `sql_user` VARCHAR,
+  `sql_user_quota` VARCHAR,
+  `sql_user_privileges` VARCHAR,
+  `query_id` VARCHAR,
+  `query_kind` VARCHAR,
+  `query_text` VARCHAR,
+  `event_date` DATE,
+  `event_time` TIMESTAMP(3),
+  `current_database` VARCHAR,
+  `databases` VARCHAR,
+  `tables` VARCHAR,
+  `columns` VARCHAR,
+  `projections` VARCHAR,
+  `written_rows` BIGINT UNSIGNED,
+  `written_bytes` BIGINT UNSIGNED,
+  `written_io_bytes` BIGINT UNSIGNED,
+  `written_io_bytes_cost_ms` BIGINT UNSIGNED,
+  `scan_rows` BIGINT UNSIGNED,
+  `scan_bytes` BIGINT UNSIGNED,
+  `scan_io_bytes` BIGINT UNSIGNED,
+  `scan_io_bytes_cost_ms` BIGINT UNSIGNED,
+  `scan_partitions` BIGINT UNSIGNED,
+  `total_partitions` BIGINT UNSIGNED,
+  `result_rows` BIGINT UNSIGNED,
+  `result_bytes` BIGINT UNSIGNED,
+  `cpu_usage` INT UNSIGNED,
+  `memory_usage` BIGINT UNSIGNED,
+  `client_info` VARCHAR,
+  `client_address` VARCHAR,
+  `exception_code` INT,
+  `exception_text` VARCHAR,
+  `stack_trace` VARCHAR,
+  `server_version` VARCHAR,
+  `session_settings` VARCHAR,
+  `extra` VARCHAR
+)
+```
+
+## Examples
+
+```
+*************************** 4. row ***************************
                 log_type: 1
             handler_type: MySQL
-               tenant_id: test_tenant
-              cluster_id: test_cluster
+               tenant_id: admin
+              cluster_id:
                 sql_user: root
           sql_user_quota: UserQuota<cpu:0,mem:0,store:0>
-     sql_user_privileges: GRANT ALL ON *.* TO 'root'@'127.0.0.1', ROLES: []
-                query_id: da879c17-94bb-4163-b2ac-ff4786bbe69e
-              query_kind: SelectPlan
-              query_text: SELECT * from system.query_log order by event_time desc limit 1
-              event_date: 2022-03-24
-              event_time: 2022-03-24 11:13:27.414
+     sql_user_privileges: GRANT ALL ON *.*, ROLES: []
+                query_id: eda2a82b-3667-4ffb-b436-953785178c39
+              query_kind: Query
+              query_text: select avg(number) from numbers(1000000)
+              event_date: 2022-09-08
+              event_time: 2022-09-08 03:32:39.517
         current_database: default
                databases:
                   tables:
@@ -34,16 +82,50 @@ written_io_bytes_cost_ms: 0
    scan_io_bytes_cost_ms: 0
          scan_partitions: 0
         total_partitions: 0
-             result_rows: 0
-            result_bytes: 0
-               cpu_usage: 10
-            memory_usage: 1603
+
+, skip_header=0, sql_dialect=PostgreSQL, storage_read_buffer_size=1048576, timezone=UTC, unquoted_ident_case_sensitive=0, wait_for_async_insert=1, wait_for_async_insert_timeout=100, scope: SESSION
+                   extra:
+                   
+                   
+*************************** 5. row ***************************
+                log_type: 2
+            handler_type: MySQL
+               tenant_id: admin
+              cluster_id:
+                sql_user: root
+          sql_user_quota: UserQuota<cpu:0,mem:0,store:0>
+     sql_user_privileges: GRANT ALL ON *.*, ROLES: []
+                query_id: eda2a82b-3667-4ffb-b436-953785178c39
+              query_kind: Query
+              query_text: select avg(number) from numbers(1000000)
+              event_date: 2022-09-08
+              event_time: 2022-09-08 03:32:39.519
+        current_database: default
+               databases:
+                  tables:
+                 columns:
+             projections:
+            written_rows: 0
+           written_bytes: 0
+        written_io_bytes: 0
+written_io_bytes_cost_ms: 0
+               scan_rows: 1000000
+              scan_bytes: 8000000
+           scan_io_bytes: 0
+   scan_io_bytes_cost_ms: 0
+         scan_partitions: 0
+        total_partitions: 0
+             result_rows: 1
+            result_bytes: 9
+               cpu_usage: 24
+            memory_usage: 0
              client_info:
-          client_address: 127.0.0.1:56744
+          client_address: 127.0.0.1:53304
           exception_code: 0
           exception_text:
              stack_trace:
           server_version:
-        session_settings: enable_new_processor_framework=1, flight_client_timeout=60, max_block_size=10000, max_threads=8, storage_occ_backoff_init_delay_ms=5, storage_occ_backoff_max_delay_ms=20000, storage_occ_backoff_max_elapsed_ms=120000, storage_read_buffer_size=1048576, scope: SESSION
+        session_settings: compression=None, empty_as_default=1, enable_async_insert=0, enable_new_processor_framework=1, enable_planner_v2=1, field_delimiter=,, flight_client_timeout=60, group_by_two_level_threshold=10000, max_block_size=10000, max_execute_time=0, max_threads=24, quoted_ident_case_sensitive=1, record_delimiter=
+, skip_header=0, sql_dialect=PostgreSQL, storage_read_buffer_size=1048576, timezone=UTC, unquoted_ident_case_sensitive=0, wait_for_async_insert=1, wait_for_async_insert_timeout=100, scope: SESSION
                    extra:
 ```

docs/doc/30-reference/30-sql/70-system-tables/system-query-log.md

@@ -22,7 +22,6 @@ common-proto-conv = { path = "../proto-conv" }
 common-protos = { path = "../protos" }
 common-tracing = { path = "../../common/tracing" }
 
-anyerror = "=0.1.6"
 anyhow = "1.0.58"
 async-trait = "0.1.56"
 enumflags2 = { version = "0.7.5", features = ["serde"] }

src/meta/api/Cargo.toml

@@ -14,13 +14,13 @@
 
 use std::fmt::Display;
 
-use anyerror::AnyError;
 use common_meta_app::schema::DatabaseId;
 use common_meta_app::schema::DatabaseIdToName;
 use common_meta_app::schema::DatabaseMeta;
 use common_meta_app::schema::DatabaseNameIdent;
 use common_meta_app::schema::TableNameIdent;
 use common_meta_app::share::*;
+use common_meta_types::anyerror::AnyError;
 use common_meta_types::app_error::AppError;
 use common_meta_types::app_error::ShareHasNoGrantedDatabase;
 use common_meta_types::app_error::UnknownDatabase;
@@ -32,6 +32,7 @@ use common_meta_types::txn_condition::Target;
 use common_meta_types::txn_op::Request;
 use common_meta_types::ConditionResult;
 use common_meta_types::MatchSeq;
+use common_meta_types::MetaBytesError;
 use common_meta_types::MetaError;
 use common_meta_types::Operation;
 use common_meta_types::TxnCondition;
@@ -107,7 +108,7 @@ pub async fn list_keys<K: KVApiKey>(
     let mut structured_keys = Vec::with_capacity(n);
 
     for (str_key, _seq_id) in res.iter() {
-        let struct_key = K::from_key(str_key).map_err(meta_encode_err)?;
+        let struct_key = K::from_key(str_key).map_err(to_bytes_err)?;
         structured_keys.push(struct_key);
     }
 
@@ -133,24 +134,24 @@ pub async fn list_u64_value<K: KVApiKey>(
     let mut values = Vec::with_capacity(n);
 
     for (str_key, seqv) in res.iter() {
-        let id = *deserialize_u64(&seqv.data).map_err(meta_encode_err)?;
+        let id = *deserialize_u64(&seqv.data)?;
         values.push(id);
 
         // Parse key
-        let struct_key = K::from_key(str_key).map_err(meta_encode_err)?;
+        let struct_key = K::from_key(str_key).map_err(to_bytes_err)?;
         structured_keys.push(struct_key);
     }
 
     Ok((structured_keys, values))
 }
 
-pub fn serialize_u64(value: impl Into<Id>) -> Result<Vec<u8>, MetaError> {
-    let v = serde_json::to_vec(&*value.into()).map_err(meta_encode_err)?;
+pub fn serialize_u64(value: impl Into<Id>) -> Result<Vec<u8>, MetaBytesError> {
+    let v = serde_json::to_vec(&*value.into())?;
     Ok(v)
 }
 
-pub fn deserialize_u64(v: &[u8]) -> Result<Id, MetaError> {
-    let id = serde_json::from_slice(v).map_err(meta_encode_err)?;
+pub fn deserialize_u64(v: &[u8]) -> Result<Id, MetaBytesError> {
+    let id = serde_json::from_slice(v)?;
     Ok(Id::new(id))
 }
 
@@ -173,30 +174,30 @@ pub async fn fetch_id<T: KVApiKey>(kv_api: &impl KVApi, generator: T) -> Result<
     Ok(seq_v.seq)
 }
 
-pub fn serialize_struct<T>(value: &T) -> Result<Vec<u8>, MetaError>
+pub fn serialize_struct<T>(value: &T) -> Result<Vec<u8>, MetaBytesError>
 where
     T: FromToProto + 'static,
     T::PB: common_protos::prost::Message,
 {
-    let p = value.to_pb().map_err(meta_encode_err)?;
+    let p = value.to_pb().map_err(to_bytes_err)?;
     let mut buf = vec![];
-    common_protos::prost::Message::encode(&p, &mut buf).map_err(meta_encode_err)?;
+    common_protos::prost::Message::encode(&p, &mut buf)?;
     Ok(buf)
 }
 
-pub fn deserialize_struct<T>(buf: &[u8]) -> Result<T, MetaError>
+pub fn deserialize_struct<T>(buf: &[u8]) -> Result<T, MetaBytesError>
 where
     T: FromToProto,
     T::PB: common_protos::prost::Message + Default,
 {
-    let p: T::PB = common_protos::prost::Message::decode(buf).map_err(meta_encode_err)?;
-    let v: T = FromToProto::from_pb(p).map_err(meta_encode_err)?;
+    let p: T::PB = common_protos::prost::Message::decode(buf)?;
+    let v: T = FromToProto::from_pb(p).map_err(to_bytes_err)?;
 
     Ok(v)
 }
 
-pub fn meta_encode_err<E: std::error::Error + 'static>(e: E) -> MetaError {
-    MetaError::EncodeError(AnyError::new(&e))
+pub fn to_bytes_err<E: std::error::Error + 'static>(e: E) -> MetaBytesError {
+    MetaBytesError::new(&e)
 }
 
 pub async fn send_txn(
@@ -485,6 +486,8 @@ where
     Ok((false, None))
 }
 
+/// Get existing value by key. Panic if key is absent.
+/// This function is only used for testing.
 pub async fn get_kv_data<T>(
     kv_api: &(impl KVApi + ?Sized),
     key: &impl KVApiKey,
@@ -495,11 +498,12 @@ where
 {
     let res = kv_api.get_kv(&key.to_key()).await?;
     if let Some(res) = res {
-        return deserialize_struct(&res.data);
+        let s = deserialize_struct(&res.data)?;
+        return Ok(s);
     };
 
-    Err(MetaError::SerdeError(AnyError::error(format!(
-        "get_kv {:?} fail",
+    Err(MetaError::Fatal(AnyError::error(format!(
+        "failed to get {}",
         key.to_key()
     ))))
 }

src/meta/api/src/kv_api_utils.rs

@@ -57,7 +57,6 @@ pub use kv_api_utils::is_all_db_data_removed;
 pub use kv_api_utils::is_db_need_to_be_remove;
 pub use kv_api_utils::list_keys;
 pub use kv_api_utils::list_u64_value;
-pub use kv_api_utils::meta_encode_err;
 pub use kv_api_utils::send_txn;
 pub use kv_api_utils::serialize_struct;
 pub use kv_api_utils::serialize_u64;