Merge remote-tracking branch 'origin/main' into bump-deps

Author: Xuanwo

Cargo.lock

@@ -1,6 +1,6 @@
 ---
-title: Tutorial - Load from an Internal Stage
-sidebar_label: Tutorial - Load from an Internal Stage
+title: Load from an Internal Stage
+sidebar_label: Load from an Internal Stage
 description:
   Load data from Databend stages.
 ---

docs/doc/12-load-data/00-stage.md

@@ -1,6 +1,6 @@
 ---
-title: Tutorial - Load from an Amazon S3 Bucket
-sidebar_label: Tutorial - Load from an Amazon S3 Bucket
+title: Load from an Amazon S3 Bucket
+sidebar_label: Load from an Amazon S3 Bucket
 description:
   Load data from Amazon S3.
 ---

docs/doc/12-load-data/01-s3.md

@@ -1,6 +1,6 @@
 ---
-title: Tutorial - Load from a Local File
-sidebar_label: Tutorial - Load from a Local File
+title: Load from a Local File
+sidebar_label: Load from a Local File
 description:
   Load data from local file system.
 ---

docs/doc/12-load-data/02-local.md

@@ -1,6 +1,6 @@
 ---
-title: Tutorial - Load from Remote File
-sidebar_label: Tutorial - Load from a Remote File
+title: Load from Remote File
+sidebar_label: Load from a Remote File
 description:
   Load data from remote files.
 ---

docs/doc/12-load-data/04-http.md

@@ -0,0 +1,86 @@
+---
+title: Transforming Data During a Load
+sidebar_label: Transforming Data During a Load
+description: Learn how to use Databend to transform data while loading it into a table using the COPY INTO <table> command. 
+---
+
+Databend supports transforming data while loading it into a table using the `COPY INTO <table>` command, which simplifies your ETL pipeline for basic transformations. 
+
+This feature helps you avoid the use of temporary tables to store pre-transformed data when reordering columns during a data load.
+
+The `COPY` command supports:
+- Column reordering, column omission, and casts using a SELECT statement. There is no requirement for your data files to have the same number and ordering of columns as your target table.
+
+:::note
+Transforming is only supported for Parquet format in the stage.
+:::
+
+## Load a Subset of Table Data
+
+Load a subset of data into a table. The following example loads data from columns `id`, `name` of a staged Parquet file:
+
+**Sample Data**
+```text
+id | name       | age
+---|------------|----
+ 1 | John Doe   |  35
+ 2 | Jane Smith |  28
+```
+
+**Example**
+```sql
+-- create a table
+CREATE TABLE my_table(id int, name string);
+
+COPY INTO my_table
+FROM (SELECT t.id, t.name FROM @mystage t)
+FILE_FORMAT = (type = parquet) PATTERN='.*parquet';
+````
+
+## Reorder Columns During a Load
+
+To reorder the columns from a staged Parquet file before loading it into a table, you can use the `COPY INTO` command with a `SELECT` statement. The following example reorders the columns `name` and `id`:
+
+**Sample Data**
+```text
+id | name       | age
+---|------------|----
+ 1 | John Doe   |  35
+ 2 | Jane Smith |  28
+```
+
+**Example**
+````sql
+CREATE TABLE my_table(name string, id int);
+
+COPY INTO my_table
+FROM (SELECT t.name, t.id FROM @mystage t)
+FILE_FORMAT = (type = parquet) PATTERN='.*parquet';
+````
+
+## Convert Data Types During a Load
+
+To convert staged data into other data types during a data load, you can use the appropriate conversion function in your `SELECT` statement.
+
+The following example converts a timestamp into a date:
+
+**Sample Data**
+```text
+id | name    | timestamp
+---|---------|--------------------
+ 1 | John Doe| 2022-03-15 10:30:00
+ 2 | Jane Doe| 2022-03-14 09:00:00
+```
+
+**Example**
+```sql
+CREATE TABLE my_table(id int, name string, time date);
+
+COPY INTO my_table
+FROM (SELECT t.id, t.name, to_date(t.timestamp) FROM @mystage t)
+FILE_FORMAT = (type = parquet) PATTERN='.*parquet';
+```
+
+## Conclusion
+
+Transforming data during a load is a powerful feature of Databend that allows you to simplify your ETL pipeline and avoid the use of temporary tables. With the ability to transform data during a load, you can streamline your ETL pipeline and focus on the analysis of your data rather than the mechanics of moving it around.

docs/doc/12-load-data/05-data-load-transform.md

@@ -135,6 +135,8 @@ build_exceptions! {
     ReadTableDataError(1107),
     AddColumnExistError(1108),
     DropColumnEmptyError(1109),
+    // create table or alter table add column with internal column name
+    TableWithInternalColumnName(1110),
 
     // Data Related Errors
 
@@ -207,6 +209,7 @@ build_exceptions! {
     /// - and without `IF EXISTS`
     CatalogNotFound(2320),
 
+
     // Cluster error codes.
     ClusterUnknownNode(2401),
     ClusterNodeAlreadyExists(2402),

src/common/exception/src/exception_code.rs

@@ -35,6 +35,7 @@ pub struct DataSourcePlan {
 
     pub tbl_args: Option<TableArgs>,
     pub push_downs: Option<PushDownInfo>,
+    pub query_internal_columns: bool,
 }
 
 impl DataSourcePlan {

src/query/catalog/src/plan/datasource/datasource_plan.rs

@@ -0,0 +1,146 @@
+// Copyright 2023 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_expression::types::string::StringColumnBuilder;
+use common_expression::types::DataType;
+use common_expression::types::NumberDataType;
+use common_expression::types::UInt64Type;
+use common_expression::BlockEntry;
+use common_expression::ColumnId;
+use common_expression::FromData;
+use common_expression::Scalar;
+use common_expression::TableDataType;
+use common_expression::Value;
+
+// Segment and Block id Bits when generate internal column `_row_id`
+// Since `DEFAULT_BLOCK_PER_SEGMENT` is 1000, so `block_id` 10 bits is enough.
+const NUM_BLOCK_ID_BITS: usize = 10;
+const NUM_SEGMENT_ID_BITS: usize = 22;
+
+pub const ROW_ID: &str = "_row_id";
+pub const SNAPSHOT_NAME: &str = "_snapshot_name";
+pub const SEGMENT_NAME: &str = "_segment_name";
+pub const BLOCK_NAME: &str = "_block_name";
+
+// meta data for generate internal columns
+#[derive(Debug)]
+pub struct InternalColumnMeta {
+    pub segment_id: usize,
+    pub block_id: usize,
+    pub block_location: String,
+    pub segment_location: String,
+    pub snapshot_location: String,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)]
+pub enum InternalColumnType {
+    RowId,
+    BlockName,
+    SegmentName,
+    SnapshotName,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)]
+pub struct InternalColumn {
+    pub column_name: String,
+    pub column_type: InternalColumnType,
+}
+
+impl InternalColumn {
+    pub fn new(name: &str, column_type: InternalColumnType) -> Self {
+        InternalColumn {
+            column_name: name.to_string(),
+            column_type,
+        }
+    }
+
+    pub fn column_type(&self) -> &InternalColumnType {
+        &self.column_type
+    }
+
+    pub fn table_data_type(&self) -> TableDataType {
+        match &self.column_type {
+            InternalColumnType::RowId => TableDataType::Number(NumberDataType::UInt64),
+            InternalColumnType::BlockName => TableDataType::String,
+            InternalColumnType::SegmentName => TableDataType::String,
+            InternalColumnType::SnapshotName => TableDataType::String,
+        }
+    }
+
+    pub fn data_type(&self) -> DataType {
+        let t = &self.table_data_type();
+        t.into()
+    }
+
+    pub fn column_name(&self) -> &String {
+        &self.column_name
+    }
+
+    pub fn column_id(&self) -> ColumnId {
+        match &self.column_type {
+            InternalColumnType::RowId => u32::MAX,
+            InternalColumnType::BlockName => u32::MAX - 1,
+            InternalColumnType::SegmentName => u32::MAX - 2,
+            InternalColumnType::SnapshotName => u32::MAX - 3,
+        }
+    }
+
+    pub fn generate_column_values(&self, meta: &InternalColumnMeta, num_rows: usize) -> BlockEntry {
+        match &self.column_type {
+            InternalColumnType::RowId => {
+                let block_id = meta.block_id as u64;
+                let seg_id = meta.segment_id as u64;
+                let high_32bit = (seg_id << NUM_SEGMENT_ID_BITS) + (block_id << NUM_BLOCK_ID_BITS);
+                let mut row_ids = Vec::with_capacity(num_rows);
+                for i in 0..num_rows {
+                    let row_id = high_32bit + i as u64;
+                    row_ids.push(row_id);
+                }
+                BlockEntry {
+                    data_type: DataType::Number(NumberDataType::UInt64),
+                    value: Value::Column(UInt64Type::from_data(row_ids)),
+                }
+            }
+            InternalColumnType::BlockName => {
+                let mut builder = StringColumnBuilder::with_capacity(1, meta.block_location.len());
+                builder.put_str(&meta.block_location);
+                builder.commit_row();
+                BlockEntry {
+                    data_type: DataType::String,
+                    value: Value::Scalar(Scalar::String(builder.build_scalar())),
+                }
+            }
+            InternalColumnType::SegmentName => {
+                let mut builder =
+                    StringColumnBuilder::with_capacity(1, meta.segment_location.len());
+                builder.put_str(&meta.segment_location);
+                builder.commit_row();
+                BlockEntry {
+                    data_type: DataType::String,
+                    value: Value::Scalar(Scalar::String(builder.build_scalar())),
+                }
+            }
+            InternalColumnType::SnapshotName => {
+                let mut builder =
+                    StringColumnBuilder::with_capacity(1, meta.snapshot_location.len());
+                builder.put_str(&meta.snapshot_location);
+                builder.commit_row();
+                BlockEntry {
+                    data_type: DataType::String,
+                    value: Value::Scalar(Scalar::String(builder.build_scalar())),
+                }
+            }
+        }
+    }
+}

src/query/catalog/src/plan/internal_column.rs

@@ -13,13 +13,15 @@
 // limitations under the License.
 
 mod datasource;
+mod internal_column;
 mod partition;
 mod partition_statistics;
 mod projection;
 mod pruning_statistics;
 mod pushdown;
 
 pub use datasource::*;
+pub use internal_column::*;
 pub use partition::*;
 pub use partition_statistics::PartStatistics;
 pub use projection::Projection;