Merge remote-tracking branch 'sundyli/tpcds-1g' into iceberg-hdfs

Xuanwo · Xuanwo · commit d2873e161f2b · 2025-02-12T09:46:37.000+08:00
diff --git a/benchmark/tpcds/README.md b/benchmark/tpcds/README.md
@@ -2,27 +2,12 @@
 
 ## Preparing the Table and Data
 
-We use [DuckDB](https://github.com/duckdb/duckdb) to generate TPC-DS data.
+We use [DuckDB](https://duckdb.org/docs/installation/) to generate TPC-DS data.
 
 After installing DuckDB, you can use these commands to generate the data ([more information](https://github.com/duckdb/duckdb/tree/master/extension/tpcds)):
 
-```sql
-INSTALL tpcds;
-LOAD tpcds;
-SELECT * FROM dsdgen(sf=0.01) -- sf can be other values, such as 0.1, 1, 10, ...
-EXPORT DATABASE '/tmp/tpcds_0_01/' (FORMAT CSV, DELIMITER '|');
-```
-
-Then, move the data to current directory:
-
-```shell
-mv /tmp/tpcds_0_01/ "$(pwd)/data/"
-```
-
-After that, you can load data to Databend:
-
 ```shell
-./load_data.sh
+./load_data.sh 0.1
 ```
 
 ## Benchmark
@@ -32,5 +17,5 @@ To run the TPC-DS Benchmark, first build `databend-sqllogictests` binary.
 Then, execute the following command in your shell:
 
 ```shell
-databend-sqllogictests --handlers mysql --database tpcds --run_dir tpcds --bench 
+databend-sqllogictests --handlers mysql --database tpcds --run_dir tpcds --bench
 ```
diff --git a/benchmark/tpcds/load_data.sh b/benchmark/tpcds/load_data.sh
@@ -3,33 +3,44 @@
 CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 . "$CURDIR"/shell_env.sh
 
+factor=$1
+
+echo """
+INSTALL tpcds;
+LOAD tpcds;
+SELECT * FROM dsdgen(sf=$factor); -- sf can be other values, such as 0.1, 1, 10, ...
+EXPORT DATABASE '/tmp/tpcds_$factor/' (FORMAT CSV, DELIMITER '|');
+""" | duckdb
+
+mv /tmp/tpcds_$factor/ "$(pwd)/data/"
+
 # Create Database
 echo "CREATE DATABASE IF NOT EXISTS ${MYSQL_DATABASE}" | $BENDSQL_CLIENT_CONNECT_DEFAULT
 
 tables=(
-    call_center   
-    catalog_returns  
-    customer_address  
-    customer_demographics  
-    household_demographics  
-    inventory  
-    promotion  
-    ship_mode  
-    store_returns  
-    time_dim   
-    web_page     
+    call_center
+    catalog_returns
+    customer_address
+    customer_demographics
+    household_demographics
+    inventory
+    promotion
+    ship_mode
+    store_returns
+    time_dim
+    web_page
     web_sales
-    catalog_page  
-    catalog_sales    
-    customer          
-    date_dim               
-    income_band             
-    item       
-    reason     
-    store      
-    store_sales    
-    warehouse  
-    web_returns  
+    catalog_page
+    catalog_sales
+    customer
+    date_dim
+    income_band
+    item
+    reason
+    store
+    store_sales
+    warehouse
+    web_returns
     web_site
 )
 
@@ -43,11 +54,13 @@ done
 cat "$CURDIR"/tpcds.sql | $BENDSQL_CLIENT_CONNECT
 
 # Load Data
+# note: export STORAGE_ALLOW_INSECURE=true to start databend-query
 for t in ${tables[@]}
 do
     echo "$t"
-    insert_sql="insert into $MYSQL_DATABASE.$t file_format = (type = CSV skip_header = 0 field_delimiter = '|' record_delimiter = '\n')"
-    curl -s -u root: -XPUT "http://localhost:8000/v1/streaming_load" -H "database: tpcds" -H "insert_sql: ${insert_sql}" -F 'upload=@"'${CURDIR}'/data/'$t'.csv"' > /dev/null 2>&1
+    fp="`pwd`/data/$t.csv"
+    echo "copy into ${MYSQL_DATABASE}.$t from 'fs://${fp}' file_format = (type = CSV skip_header = 1 field_delimiter = '|' record_delimiter = '\n')" | $BENDSQL_CLIENT_CONNECT
+    echo "analyze table ${MYSQL_DATABASE}.$t" | $BENDSQL_CLIENT_CONNECT
 done
 
 
diff --git a/benchmark/tpch/README.md b/benchmark/tpch/README.md
@@ -3,6 +3,7 @@
 
 ## Preparing the Table and Data
 
+We use [DuckDB](https://duckdb.org/docs/installation/) to generate TPCH data.
 To prepare the table and data for the TPC-H Benchmark, run the following command in your shell:
 
 ```shell
diff --git a/benchmark/tpch/gen_data.sh b/benchmark/tpch/gen_data.sh
diff --git a/benchmark/tpch/load_data.sh b/benchmark/tpch/load_data.sh
@@ -111,9 +111,11 @@ echo "CREATE TABLE IF NOT EXISTS lineitem
 ) CLUSTER BY(l_shipdate, l_orderkey) ${options}" | $BENDSQL_CLIENT_CONNECT
 
 # insert data to tables
+# note: export STORAGE_ALLOW_INSECURE=true to start databend-query
 for t in customer lineitem nation orders partsupp part region supplier
 do
     echo "$t"
-    insert_sql="insert into ${MYSQL_DATABASE}.$t file_format = (type = CSV skip_header = 0 field_delimiter = '|' record_delimiter = '\n')"
-    curl -s -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" -H "database: tpch" -H "insert_sql: ${insert_sql}" -F 'upload=@"./data/'$t'.tbl"'
+    fp="`pwd`/data/$t.tbl"
+    echo "copy into ${MYSQL_DATABASE}.$t from 'fs://${fp}' file_format = (type = CSV skip_header = 1 field_delimiter = '|' record_delimiter = '\n')" | $BENDSQL_CLIENT_CONNECT
+    echo "analyze table ${MYSQL_DATABASE}.$t" | $BENDSQL_CLIENT_CONNECT
 done
diff --git a/benchmark/tpch/tpch.sh b/benchmark/tpch/tpch.sh
@@ -1,13 +1,20 @@
 #!/usr/bin/env bash
 
-# generate tpch data
-sh ./gen_data.sh $1
+
+echo """
+INSTALL tpch;
+LOAD tpch;
+SELECT * FROM dsdgen(sf=1); -- sf can be other values, such as 0.1, 1, 10, ...
+EXPORT DATABASE '/tmp/tpch_1/' (FORMAT CSV, DELIMITER '|');
+""" | duckdb
+
+mv /tmp/tpch_1/ "$(pwd)/data/"
 
 if [[ $2 == native ]]; then
   echo "native"
-  sh ./prepare_table.sh "storage_format = 'native' compression = 'lz4'"
+  sh ./load_data.sh "storage_format = 'native' compression = 'lz4'"
 else
   echo "fuse"
-  sh ./prepare_table.sh ""
+  sh ./load_data.sh ""
 fi
 
diff --git a/src/common/metrics/src/metrics/storage.rs b/src/common/metrics/src/metrics/storage.rs
@@ -35,6 +35,7 @@ static OMIT_FILTER_ROWS: LazyLock<Counter> = LazyLock::new(|| register_counter("
 struct StorageHttpLabels {
     host: String,
     method: String,
+    bucket: String,
 }
 
 static STORAGE_HTTP_REQUESTS_COUNT: LazyLock<FamilyCounter<StorageHttpLabels>> =
@@ -320,9 +321,13 @@ pub fn metrics_inc_omit_filter_rows(c: u64) {
 }
 
 /// Storage Http metrics.
-pub fn metrics_inc_storage_http_requests_count(host: String, method: String) {
+pub fn metrics_inc_storage_http_requests_count(host: String, method: String, bucket: String) {
     STORAGE_HTTP_REQUESTS_COUNT
-        .get_or_create(&StorageHttpLabels { host, method })
+        .get_or_create(&StorageHttpLabels {
+            host,
+            method,
+            bucket,
+        })
         .inc();
 }
 
diff --git a/src/common/storage/src/http_client.rs b/src/common/storage/src/http_client.rs
@@ -60,7 +60,16 @@ impl HttpFetch for StorageHttpClient {
             }
             m => m.as_str(),
         };
-        metrics_inc_storage_http_requests_count(host.to_string(), method.to_string());
+        // get first component in path as bucket name
+        let bucket = match url.path_segments() {
+            Some(mut segments) => segments.next().unwrap_or("-"),
+            None => "-",
+        };
+        metrics_inc_storage_http_requests_count(
+            host.to_string(),
+            method.to_string(),
+            bucket.to_string(),
+        );
 
         let (parts, body) = req.into_parts();
 
diff --git a/src/common/storage/src/statistics.rs b/src/common/storage/src/statistics.rs
@@ -182,4 +182,21 @@ impl Datum {
             ))),
         }
     }
+
+    pub fn can_compare(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Datum::Bool(_), Datum::Bool(_))
+            | (Datum::Int(_), Datum::Int(_))
+            | (Datum::Int(_), Datum::UInt(_))
+            | (Datum::Int(_), Datum::Float(_))
+            | (Datum::UInt(_), Datum::UInt(_))
+            | (Datum::UInt(_), Datum::Int(_))
+            | (Datum::UInt(_), Datum::Float(_))
+            | (Datum::Float(_), Datum::Float(_))
+            | (Datum::Float(_), Datum::Int(_))
+            | (Datum::Float(_), Datum::UInt(_))
+            | (Datum::Bytes(_), Datum::Bytes(_)) => true,
+            _ => false,
+        }
+    }
 }
diff --git a/src/query/expression/src/evaluator.rs b/src/query/expression/src/evaluator.rs
@@ -377,6 +377,40 @@ impl<'a> Evaluator<'a> {
                     .set_span(span))
                 }
             }
+            (
+                DataType::Nullable(box DataType::Variant) | DataType::Variant,
+                DataType::Boolean
+                | DataType::Number(_)
+                | DataType::String
+                | DataType::Date
+                | DataType::Timestamp,
+            ) => {
+                // allow cast variant to not null types.
+                let cast_fn = format!("to_{}", dest_type.to_string().to_lowercase());
+                if let Some(new_value) = self.run_simple_cast(
+                    span,
+                    src_type,
+                    &dest_type.wrap_nullable(),
+                    value.clone(),
+                    &cast_fn,
+                    validity.clone(),
+                    options,
+                )? {
+                    let (new_value, has_null) = new_value.remove_nullable();
+                    if has_null {
+                        return Err(ErrorCode::BadArguments(format!(
+                            "unable to cast type `{src_type}` to type `{dest_type}`, result has null values"
+                        ))
+                        .set_span(span));
+                    }
+                    Ok(new_value)
+                } else {
+                    Err(ErrorCode::BadArguments(format!(
+                        "unable to cast type `{src_type}` to type `{dest_type}`"
+                    ))
+                    .set_span(span))
+                }
+            }
             (DataType::Nullable(inner_src_ty), DataType::Nullable(inner_dest_ty)) => match value {
                 Value::Scalar(Scalar::Null) => Ok(Value::Scalar(Scalar::Null)),
                 Value::Scalar(_) => {
diff --git a/src/query/expression/src/values.rs b/src/query/expression/src/values.rs
@@ -345,6 +345,18 @@ impl Value<AnyType> {
         }
     }
 
+    // returns result without nullable and has_null flag
+    pub fn remove_nullable(self) -> (Self, bool) {
+        match self {
+            Value::Scalar(Scalar::Null) => (Value::Scalar(Scalar::Null), true),
+            Value::Column(Column::Nullable(box nullable_column)) => (
+                Value::Column(nullable_column.column),
+                nullable_column.validity.null_count() > 0,
+            ),
+            other => (other, false),
+        }
+    }
+
     pub fn domain(&self, data_type: &DataType) -> Domain {
         match self {
             Value::Scalar(scalar) => scalar.as_ref().domain(data_type),
diff --git a/src/query/sql/src/planner/optimizer/property/selectivity.rs b/src/query/sql/src/planner/optimizer/property/selectivity.rs
@@ -442,6 +442,10 @@ impl<'a> SelectivityEstimator<'a> {
             let lower_bound = bucket.lower_bound();
             let upper_bound = bucket.upper_bound();
 
+            if !const_datum.can_compare(&lower_bound) {
+                return Ok(DEFAULT_SELECTIVITY);
+            }
+
             let const_gte_upper_bound = matches!(
                 const_datum.compare(upper_bound)?,
                 Ordering::Greater | Ordering::Equal
diff --git a/src/query/sql/src/planner/semantic/type_check.rs b/src/query/sql/src/planner/semantic/type_check.rs
@@ -3169,10 +3169,10 @@ impl<'a> TypeChecker<'a> {
         if let SetExpr::Select(select_stmt) = &subquery.body {
             if typ == SubqueryType::Scalar {
                 let select = &select_stmt.select_list[0];
-                if let SelectTarget::AliasedExpr { expr, .. } = select {
+                if matches!(select, SelectTarget::AliasedExpr { .. }) {
                     // Check if contain aggregation function
                     #[derive(Visitor)]
-                    #[visitor(ASTFunctionCall(enter))]
+                    #[visitor(Expr(enter), ASTFunctionCall(enter))]
                     struct AggFuncVisitor {
                         contain_agg: bool,
                     }
@@ -3182,9 +3182,12 @@ impl<'a> TypeChecker<'a> {
                                 || AggregateFunctionFactory::instance()
                                     .contains(func.name.to_string());
                         }
+                        fn enter_expr(&mut self, expr: &Expr) {
+                            self.contain_agg = matches!(expr, Expr::CountAll { window: None, .. });
+                        }
                     }
                     let mut visitor = AggFuncVisitor { contain_agg: false };
-                    expr.drive(&mut visitor);
+                    select.drive(&mut visitor);
                     contain_agg = Some(visitor.contain_agg);
                 }
             }
diff --git a/tests/sqllogictests/suites/base/03_common/03_0014_insert_into_select.test b/tests/sqllogictests/suites/base/03_common/03_0014_insert_into_select.test
@@ -124,5 +124,29 @@ DROP TABLE aggregate_table
 statement ok
 DROP TABLE test_insert
 
+statement ok
+CREATE OR REPLACE TABLE t4(data variant);
+
+statement ok
+INSERT INTO t4 values('{"user_id":1}'),('{"user_id":2}')
+
+statement ok
+CREATE OR REPLACE TABLE t5(data variant, user_id uint8 not null);
+
+statement ok
+INSERT INTO t5 SELECT data, data:user_id AS user_id FROM t4;
+
+query TI
+SELECT * FROM t5;
+----
+{"user_id":1} 1
+{"user_id":2} 2
+
+statement ok
+INSERT INTO t4 values('{}');
+
+statement error
+INSERT INTO t5 SELECT data, data:user_id AS user_id FROM t4;
+
 statement ok
 DROP DATABASE db1
diff --git a/tests/sqllogictests/suites/tpcds/opt.test b/tests/sqllogictests/suites/tpcds/opt.test
