refactor: streaming load use header X-Databend-SQL. (#18410)

Author: youngsofun

src/common/base/src/headers.rs

@@ -17,6 +17,7 @@ pub const HEADER_SESSION_ID: &str = "X-DATABEND-SESSION-ID";
 pub const HEADER_TENANT: &str = "X-DATABEND-TENANT";
 pub const HEADER_QUERY_ID: &str = "X-DATABEND-QUERY-ID";
 pub const HEADER_USER: &str = "X-DATABEND-USER";
+pub const HEADER_SQL: &str = "X-DATABEND-SQL";
 
 pub const HEADER_FUNCTION: &str = "X-DATABEND-FUNCTION";
 pub const HEADER_FUNCTION_HANDLER: &str = "X-DATABEND-FUNCTION-HANDLER";

src/query/service/src/servers/http/v1/streaming_load.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 use databend_common_base::base::tokio;
 use databend_common_base::base::tokio::io::AsyncReadExt;
 use databend_common_base::base::ProgressValues;
+use databend_common_base::headers::HEADER_SQL;
 use databend_common_base::runtime::MemStat;
 use databend_common_base::runtime::ThreadTracker;
 use databend_common_base::runtime::TrySpawn;
@@ -131,13 +132,19 @@ async fn streaming_load_handler_inner(
         .await
         .map_err(InternalServerError)?;
 
-    let sql = req.headers().get("sql").ok_or(poem::Error::from_string(
-        "[HTTP-STREAMING-LOAD] Missing required 'sql' header in request",
-        StatusCode::BAD_REQUEST,
-    ))?;
+    let sql = req
+        .headers()
+        .get(HEADER_SQL)
+        .ok_or(poem::Error::from_string(
+            "[HTTP-STREAMING-LOAD] Missing required 'sql' header in request",
+            StatusCode::BAD_REQUEST,
+        ))?;
     let sql = sql.to_str().map_err(|e| {
         poem::Error::from_string(
-            format!("[HTTP-STREAMING-LOAD] Invalid UTF-8 in 'sql' header: {}", e),
+            format!(
+                "[HTTP-STREAMING-LOAD] Invalid UTF-8 in value of header {HEADER_SQL}: {}",
+                e
+            ),
             StatusCode::BAD_REQUEST,
         )
     })?;

tests/suites/1_stateful/01_streaming_load/01_0001_streaming_load_ontime.sh

@@ -10,7 +10,7 @@ cat $TESTS_DATA_DIR/ddl/ontime.sql | sed 's/ontime/ontime_streaming_load/g' | $B
 
 function run() {
   echo "--$2"
-  curl -sS -H "x-databend-query-id:$2" -H "sql:insert into ontime_streaming_load  from @_databend_load  file_format = ($1)" -F "upload=@/${TESTS_DATA_DIR}/$2" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"
+  curl -sS -H "x-databend-query-id:$2" -H "X-Databend-SQL:insert into ontime_streaming_load  from @_databend_load  file_format = ($1)" -F "upload=@/${TESTS_DATA_DIR}/$2" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"
   echo
   echo "select count(1), avg(Year), sum(DayOfWeek)  from ontime_streaming_load;" | $BENDSQL_CLIENT_CONNECT
   echo "truncate table ontime_streaming_load" | $BENDSQL_CLIENT_CONNECT

tests/suites/1_stateful/01_streaming_load/01_0002_streaming_load_variant.sh

@@ -10,7 +10,7 @@ cat ${TESTS_DATA_DIR}/ddl/variant_test.sql | $BENDSQL_CLIENT_CONNECT
 
 # run format path table
 function run() {
-  curl -sS -H "x-databend-query-id:$2" -H "sql:insert into $3 values  from @_databend_load file_format = ($1)" -F "upload=@/${TESTS_DATA_DIR}/$2" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"
+  curl -sS -H "x-databend-query-id:$2" -H "X-Databend-SQL:insert into $3 values  from @_databend_load file_format = ($1)" -F "upload=@/${TESTS_DATA_DIR}/$2" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"
   echo
 }
 

tests/suites/1_stateful/01_streaming_load/01_0003_streaming_load_dedup.sh

@@ -10,7 +10,7 @@ function run() {
 	echo ">>>> load $2 with format ($1), with label $3"
   curl -sS \
   -H "x-databend-query-id:$2" \
-  -H "sql:insert /*+ set_var(deduplicate_label='$3') */ into t1  from @_databend_load file_format = ($1)" \
+  -H "X-Databend-SQL:insert /*+ set_var(deduplicate_label='$3') */ into t1  from @_databend_load file_format = ($1)" \
   -F "upload=@/${TESTS_DATA_DIR}/$2" \
   -u root: -XPUT \
   "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"

tests/suites/1_stateful/01_streaming_load/01_0004_streaming_load_error.sh

@@ -10,7 +10,7 @@ function run() {
 	echo ">>>> load $2 with format ($1)"
   curl -sS \
   -H "x-databend-query-id:$2" \
-  -H "sql:insert into t1 values from @_databend_load  file_format = ($1)" \
+  -H "X-Databend-SQL:insert into t1 values from @_databend_load  file_format = ($1)" \
   -F "upload=@/${TESTS_DATA_DIR}/$2" \
   -u root: -XPUT \
   "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load" | jq .error.code

tests/suites/1_stateful/01_streaming_load/01_0005_streaming_load_compact.sh

@@ -14,7 +14,7 @@ stmt "drop table if exists t1 all"
 stmt "CREATE TABLE t1 ( c0 string );"
 
 curl -sS \
--H "sql:insert /*+ set_var(input_read_buffer_size=100)  */into t1 values  from @_databend_load file_format = (type = CSV)" \
+-H "X-Databend-SQL:insert /*+ set_var(input_read_buffer_size=100)  */into t1 values  from @_databend_load file_format = (type = CSV)" \
 -H "X-databend-query-id: test" \
 -F "upload=@${DATA}" \
 -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load"

tests/suites/1_stateful/01_streaming_load/01_0006_streaming_load_parquet.result

@@ -4,7 +4,7 @@
 >>>> copy into @streaming_load_parquet/q1.parquet from (select '2021-01-01' as c3, '1' as c2)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 q1.parquet	624	1
 >>>> streaming load: q1.parquet error :
-+ curl -sS -H x-databend-query-id:load-q1 -H 'sql:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q1.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-q1 -H 'X-Databend-SQL:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q1.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-q1","stats":{"rows":1,"bytes":25}}
 <<<<
 >>>> select * from streaming_load_parquet;
@@ -15,7 +15,7 @@ ok	1	2021-01-01
 >>>> copy into @streaming_load_parquet/q2.parquet from (select '2021-01-01' as c3)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 q2.parquet	426	1
 >>>> streaming load: q2.parquet error :
-+ curl -sS -H x-databend-query-id:load-q2 -H 'sql:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q2.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-q2 -H 'X-Databend-SQL:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q2.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"error":{"code":400,"message":"[HTTP-STREAMING-LOAD] Query execution failed: file q2.parquet missing column `c2`"}}
 <<<<
 >>>> select * from streaming_load_parquet;
@@ -25,7 +25,7 @@ q2.parquet	426	1
 >>>> copy into @streaming_load_parquet/q3.parquet from (select '2021-01-01' as c3)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 q3.parquet	426	1
 >>>> streaming load: q3.parquet field_default :
-+ curl -sS -H x-databend-query-id:load-q3 -H 'sql:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=field_default, null_if=())' -F upload=@/tmp/streaming_load_parquet/q3.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-q3 -H 'X-Databend-SQL:insert into streaming_load_parquet(c2,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=field_default, null_if=())' -F upload=@/tmp/streaming_load_parquet/q3.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-q3","stats":{"rows":1,"bytes":21}}
 <<<<
 >>>> select * from streaming_load_parquet;
@@ -36,7 +36,7 @@ ok	NULL	2021-01-01
 >>>> copy into @streaming_load_parquet/q4.parquet from (select '2021-01-01' as c3, 'my_null' as c1)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 q4.parquet	643	1
 >>>> streaming load: q4.parquet error :
-+ curl -sS -H x-databend-query-id:load-q4 -H 'sql:insert into streaming_load_parquet(c1,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q4.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-q4 -H 'X-Databend-SQL:insert into streaming_load_parquet(c1,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=())' -F upload=@/tmp/streaming_load_parquet/q4.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-q4","stats":{"rows":1,"bytes":26}}
 <<<<
 >>>> select * from streaming_load_parquet;
@@ -47,7 +47,7 @@ my_null	NULL	2021-01-01
 >>>> copy into @streaming_load_parquet/q5.parquet from (select '2021-01-01' as c3, 'my_null' as c1)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 q5.parquet	643	1
 >>>> streaming load: q5.parquet error 'my_null':
-+ curl -sS -H x-databend-query-id:load-q5 -H 'sql:insert into streaming_load_parquet(c1,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=('\''my_null'\''))' -F upload=@/tmp/streaming_load_parquet/q5.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-q5 -H 'X-Databend-SQL:insert into streaming_load_parquet(c1,c3) from @_databend_load file_format = (type='\''parquet'\'', missing_field_as=error, null_if=('\''my_null'\''))' -F upload=@/tmp/streaming_load_parquet/q5.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-q5","stats":{"rows":1,"bytes":7}}
 <<<<
 >>>> select * from streaming_load_parquet;

tests/suites/1_stateful/01_streaming_load/01_0006_streaming_load_parquet.sh

@@ -14,7 +14,7 @@ function run() {
   echo "--$2"
   stmt "copy into @streaming_load_parquet/$1.parquet from (select $2)  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;"
   echo ">>>> streaming load: $1.parquet $4 $5:"
-  (set -x; curl -sS -H "x-databend-query-id:load-$1" -H "sql:insert into streaming_load_parquet($3) from @_databend_load file_format = (type='parquet', missing_field_as=$4, null_if=($5))" -F "upload=@$DATA/$1.parquet" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load")
+  (set -x; curl -sS -H "x-databend-query-id:load-$1" -H "X-Databend-SQL:insert into streaming_load_parquet($3) from @_databend_load file_format = (type='parquet', missing_field_as=$4, null_if=($5))" -F "upload=@$DATA/$1.parquet" -u root: -XPUT "http://localhost:${QUERY_HTTP_HANDLER_PORT}/v1/streaming_load")
   echo
   echo "<<<<"
   query "select * from streaming_load_parquet;"

tests/suites/1_stateful/01_streaming_load/01_0007_streaming_load_placeholder.result

@@ -3,7 +3,7 @@
 --csv
 >>>> copy into @streaming_load_07/data.csv from (select '2020-01-02' as c4, 110 as c2) file_format=(type='csv')  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 data.csv	17	1
-+ curl -sS -H x-databend-query-id:load-csv -H 'sql:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=csv)' -F upload=@/tmp/streaming_load_07/data.csv -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-csv -H 'X-Databend-SQL:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=csv)' -F upload=@/tmp/streaming_load_07/data.csv -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-csv","stats":{"rows":1,"bytes":39}}
 <<<<
 >>>> select * from streaming_load_07;
@@ -13,7 +13,7 @@ ok	110	a	2020-01-02
 --tsv
 >>>> copy into @streaming_load_07/data.tsv from (select '2020-01-02' as c4, 110 as c2) file_format=(type='tsv')  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 data.tsv	15	1
-+ curl -sS -H x-databend-query-id:load-tsv -H 'sql:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=tsv)' -F upload=@/tmp/streaming_load_07/data.tsv -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-tsv -H 'X-Databend-SQL:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=tsv)' -F upload=@/tmp/streaming_load_07/data.tsv -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-tsv","stats":{"rows":1,"bytes":39}}
 <<<<
 >>>> select * from streaming_load_07;
@@ -23,7 +23,7 @@ ok	110	a	2020-01-02
 --ndjson
 >>>> copy into @streaming_load_07/data.ndjson from (select '2020-01-02' as c4, 110 as c2) file_format=(type='ndjson')  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 data.ndjson	29	1
-+ curl -sS -H x-databend-query-id:load-ndjson -H 'sql:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=ndjson)' -F upload=@/tmp/streaming_load_07/data.ndjson -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-ndjson -H 'X-Databend-SQL:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=ndjson)' -F upload=@/tmp/streaming_load_07/data.ndjson -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-ndjson","stats":{"rows":1,"bytes":39}}
 <<<<
 >>>> select * from streaming_load_07;
@@ -33,7 +33,7 @@ ok	110	a	2020-01-02
 --parquet
 >>>> copy into @streaming_load_07/data.parquet from (select '2020-01-02' as c4, 110 as c2) file_format=(type='parquet')  single=true include_query_id=false use_raw_path=true detailed_output=true overwrite=true;
 data.parquet	665	1
-+ curl -sS -H x-databend-query-id:load-parquet -H 'sql:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=parquet)' -F upload=@/tmp/streaming_load_07/data.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
++ curl -sS -H x-databend-query-id:load-parquet -H 'X-Databend-SQL:insert into streaming_load_07(c3, c4, c2) values ('\''a'\'', ?, ?) from @_databend_load file_format = (type=parquet)' -F upload=@/tmp/streaming_load_07/data.parquet -u root: -XPUT http://localhost:8000/v1/streaming_load
 {"id":"load-parquet","stats":{"rows":1,"bytes":39}}
 <<<<
 >>>> select * from streaming_load_07;