feat(query): enable runtime cast transform in loading parquet files

Author: sundy-li

scripts/benchmark/query/load/tpch100.sh

@@ -7,7 +7,7 @@ select version();
 SQL
 
 for t in customer lineitem nation orders partsupp part region supplier; do
-  echo "DROP TABLE IF EXISTS $t;" | bendsql query
+	echo "DROP TABLE IF EXISTS $t;" | bendsql query
 done
 
 cat <<SQL | bendsql query
@@ -113,8 +113,8 @@ cat <<SQL | bendsql query
 SQL
 
 for t in nation region; do
-  echo "loading into $t ..."
-  cat <<SQL | bendsql query
+	echo "loading into $t ..."
+	cat <<SQL | bendsql query
 COPY INTO $t FROM 's3://repo.databend.rs/tpch100/${t}.tbl'
 credentials=(aws_key_id='$REPO_ACCESS_KEY_ID' aws_secret_key='$REPO_SECRET_ACCESS_KEY')
 file_format=(type='CSV' field_delimiter='|' record_delimiter='\\n' skip_header=1);
@@ -124,8 +124,8 @@ SQL
 done
 
 for t in customer lineitem orders partsupp part supplier; do
-  echo "loading into $t ..."
-  cat <<SQL | bendsql query
+	echo "loading into $t ..."
+	cat <<SQL | bendsql query
 COPY INTO $t FROM 's3://repo.databend.rs/tpch100/${t}/'
 credentials=(aws_key_id='$REPO_ACCESS_KEY_ID' aws_secret_key='$REPO_SECRET_ACCESS_KEY') pattern ='${t}.tbl.*'
 file_format=(type='CSV' field_delimiter='|' record_delimiter='\\n' skip_header=1);

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

@@ -367,7 +367,7 @@ impl BlockBuilderTrait for ParquetBlockBuilder {
             let fields: Vec<DataField> = rg
                 .fields_to_read
                 .iter()
-                .map(|f| DataField::from(f))
+                .map(DataField::from)
                 .collect::<Vec<_>>();
 
             let input_schema = DataSchema::new(fields);

tests/suites/1_stateful/01_load/01_0000_streaming_load.result

@@ -12,5 +12,5 @@
 198	2020.0	767
 --parquet less
 199	2020.0	769
---parquet mismatch schema
+--parquet runtime cast schema
 199	2020.0	769

tests/suites/1_stateful/01_load/01_0000_streaming_load.sh

@@ -88,7 +88,7 @@ curl -s -H "insert_sql:insert into ontime_less file_format = (type = Parquet)" -
 echo "select count(1), avg(Year), sum(DayOfWeek)  from ontime_less;" | $MYSQL_CLIENT_CONNECT
 
 # load parquet with mismatch schema, will auto cast
-echo "--parquet mismatch schema"
+echo "--parquet runtime cast schema"
 cat $CURDIR/../ddl/ontime.sql | sed 's/ontime/ontime_test_schmea_mismatch/g' | sed 's/DATE/TIMESTAMP/g' | $MYSQL_CLIENT_CONNECT
 curl -s -H "insert_sql:insert into ontime_test_schmea_mismatch file_format = (type = Parquet)" -F "upload=@/tmp/ontime_200.parquet" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"  > /dev/null 2>&1
 echo "select count(1), avg(Year), sum(DayOfWeek)  from ontime_test_schmea_mismatch;" | $MYSQL_CLIENT_CONNECT