fix(http): improve missing-@ upload errors (#19752)

fix(http): improve missing-@ upload errors (#10967)

Author: KKould

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

@@ -27,7 +27,6 @@ mod verify;
 #[cfg(not(feature = "storage-stage"))]
 use databend_common_base::base::ProgressValues;
 pub use discovery::discovery_nodes;
-#[cfg(not(feature = "storage-stage"))]
 use http::StatusCode;
 pub use http_query_handlers::QueryResponse;
 pub use http_query_handlers::QueryResponseField;
@@ -76,6 +75,24 @@ pub use verify::verify_handler;
 
 pub use crate::servers::http::error::QueryError;
 
+fn require_upload_filename(
+    field_name: Option<&str>,
+    filename: Option<&str>,
+) -> poem::Result<String> {
+    match filename.filter(|name| !name.is_empty()) {
+        Some(name) => Ok(name.to_string()),
+        None => {
+            let field_name = field_name.unwrap_or("upload");
+            Err(poem::Error::from_string(
+                format!(
+                    "Invalid form-data field '{field_name}': expected a file upload with a filename. If you are using curl, did you forget the '@' in -F '{field_name}=@/path/to/file'?"
+                ),
+                StatusCode::BAD_REQUEST,
+            ))
+        }
+    }
+}
+
 #[cfg(not(feature = "storage-stage"))]
 #[derive(Serialize, Deserialize, Debug)]
 pub struct LoadResponse {

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

@@ -29,6 +29,7 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::HttpQueryContext;
+use super::require_upload_filename;
 use crate::sessions::TableContext;
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -115,10 +116,7 @@ pub async fn upload_to_stage(
 
     let mut files = vec![];
     while let Ok(Some(field)) = multipart.next_field().await {
-        let name = match field.file_name() {
-            Some(name) => name.to_string(),
-            None => uuid::Uuid::new_v4().to_string(),
-        };
+        let name = require_upload_filename(field.name(), field.file_name())?;
         let file_path = format!("{}/{}", args.relative_path, name)
             .trim_start_matches('/')
             .to_string();
@@ -141,7 +139,7 @@ pub async fn upload_to_stage(
         files.push(name.clone());
     }
 
-    let mut id = uuid::Uuid::new_v4().to_string();
+    let id = uuid::Uuid::new_v4().to_string();
     Ok(Json(UploadToStageResponse {
         id,
         stage_name: args.stage_name,

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

@@ -56,6 +56,7 @@ use tokio::sync::mpsc::Sender;
 use super::HttpQueryContext;
 use super::HttpSessionConf;
 use super::HttpSessionStateInternal;
+use super::require_upload_filename;
 use crate::interpreters::InterpreterFactory;
 use crate::servers::http::error::HttpErrorCode;
 use crate::servers::http::error::JsonErrorOnly;
@@ -333,7 +334,7 @@ async fn read_multi_part(
                         StatusCode::BAD_REQUEST,
                     ));
                 }
-                let filename = field.file_name().unwrap_or("file_with_no_name").to_string();
+                let filename = require_upload_filename(name, field.file_name())?;
                 debug!("Started reading file: {}", &filename);
                 let mut reader = field.into_async_read();
                 match file_format {

tests/nox/suites/http_handler/test_upload_missing_at.py

@@ -0,0 +1,64 @@
+import requests
+
+
+query_url = "http://localhost:8000/v1/query"
+streaming_url = "http://localhost:8000/v1/streaming_load"
+upload_url = "http://localhost:8000/v1/upload_to_stage"
+auth = ("root", "")
+
+
+def execute_sql(sql):
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 6}}
+    response = requests.post(
+        query_url, auth=auth, headers={"Content-Type": "application/json"}, json=payload
+    )
+    return response.json()
+
+
+def missing_at_upload():
+    return {"upload": (None, "./abc.json")}
+
+
+def test_upload_to_stage_requires_file_upload():
+    stage_name = "missing_at_stage"
+    assert execute_sql(f"drop stage if exists {stage_name}")["error"] == None
+    assert execute_sql(f"create stage {stage_name}")["error"] == None
+
+    try:
+        response = requests.put(
+            upload_url,
+            auth=auth,
+            headers={"x-databend-stage-name": stage_name},
+            files=missing_at_upload(),
+        )
+
+        assert response.status_code == 400
+        assert "expected a file upload with a filename" in response.text
+        assert "did you forget the '@'" in response.text.lower()
+        assert "upload=@/path/to/file" in response.text
+    finally:
+        assert execute_sql(f"drop stage if exists {stage_name}")["error"] == None
+
+
+def test_streaming_load_requires_file_upload():
+    table_name = "missing_at_streaming"
+    assert execute_sql(f"drop table if exists {table_name}")["error"] == None
+    assert execute_sql(f"create table {table_name} (a string)")["error"] == None
+
+    try:
+        response = requests.put(
+            streaming_url,
+            auth=auth,
+            headers={
+                "X-Databend-SQL": f"insert into {table_name} from @_databend_load file_format = (type = csv)",
+                "x-databend-query-id": "missing-at-streaming-load",
+            },
+            files=missing_at_upload(),
+        )
+
+        assert response.status_code == 400
+        assert "expected a file upload with a filename" in response.text
+        assert "did you forget the '@'" in response.text.lower()
+        assert "upload=@/path/to/file" in response.text
+    finally:
+        assert execute_sql(f"drop table if exists {table_name}")["error"] == None