Remove sql statements

Signed-off-by: Xuanwo <github@xuanwo.io>

Author: Xuanwo

src/query/service/src/sql/mod.rs

@@ -14,7 +14,6 @@
 
 pub mod executor;
 pub mod planner;
-pub mod statements;
 use common_legacy_parser::sql_common;
 use common_storages_util::table_option_keys;
 pub use planner::*;

src/query/service/src/sql/planner/binder/copy.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use common_ast::ast::CopyStmt;
 use common_ast::ast::CopyUnit;
@@ -22,22 +24,26 @@ use common_ast::parser::parse_sql;
 use common_ast::parser::tokenize_sql;
 use common_ast::Backtrace;
 use common_ast::Dialect;
+use common_catalog::table_context::TableContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_io::prelude::parse_escape_string;
 use common_legacy_planners::ReadDataSourcePlan;
 use common_legacy_planners::SourceInfo;
 use common_legacy_planners::StageTableInfo;
+use common_meta_types::FileFormatOptions;
+use common_meta_types::StageFileFormatType;
 use common_meta_types::UserStageInfo;
 use common_storage::parse_uri_location;
 use common_storage::UriLocation;
+use common_users::UserApiProvider;
+use tracing::debug;
 
 use crate::sql::binder::Binder;
 use crate::sql::normalize_identifier;
 use crate::sql::plans::CopyPlanV2;
 use crate::sql::plans::Plan;
 use crate::sql::plans::ValidationMode;
-use crate::sql::statements::parse_copy_file_format_options;
-use crate::sql::statements::parse_stage_location_v2;
 use crate::sql::BindContext;
 
 impl<'a> Binder {
@@ -500,3 +506,123 @@ impl<'a> Binder {
         Ok(())
     }
 }
+
+/// Named stage(start with `@`):
+///
+/// ```sql
+/// copy into mytable from @my_ext_stage
+///     file_format = (type = csv);
+/// ```
+///
+/// Returns user's stage info and relative path towards the stage's root.
+///
+/// If input location is empty we will convert it to `/` means the root of stage
+///
+/// - @mystage => (mystage, "/")
+///
+/// If input location is endswith `/`, it's a folder.
+///
+/// - @mystage/ => (mystage, "/")
+///
+/// Otherwise, it's a file
+///
+/// - @mystage/abc => (mystage, "abc")
+///
+/// For internal stage, we will also add prefix `/stage/<stage>/`
+///
+/// - @internal/abc => (internal, "/stage/internal/abc")
+pub async fn parse_stage_location(
+    ctx: &Arc<dyn TableContext>,
+    location: &str,
+) -> Result<(UserStageInfo, String)> {
+    let s: Vec<&str> = location.split('@').collect();
+    // @my_ext_stage/abc/
+    let names: Vec<&str> = s[1].splitn(2, '/').filter(|v| !v.is_empty()).collect();
+    let stage = UserApiProvider::instance()
+        .get_stage(&ctx.get_tenant(), names[0])
+        .await?;
+
+    let path = names.get(1).unwrap_or(&"").trim_start_matches('/');
+
+    let prefix = stage.get_prefix();
+    let relative_path = format!("{prefix}{path}");
+
+    debug!("parsed stage: {stage:?}, path: {relative_path}");
+    Ok((stage, relative_path))
+}
+
+/// parse_stage_location_v2 work similar to parse_stage_location.
+///
+/// Difference is input location has already been parsed by parser.
+pub async fn parse_stage_location_v2(
+    ctx: &Arc<dyn TableContext>,
+    name: &str,
+    path: &str,
+) -> Result<(UserStageInfo, String)> {
+    debug_assert!(path.starts_with('/'), "path should starts with '/'");
+
+    let stage = UserApiProvider::instance()
+        .get_stage(&ctx.get_tenant(), name)
+        .await?;
+
+    let prefix = stage.get_prefix();
+    debug_assert!(prefix.ends_with('/'), "prefix should ends with '/'");
+
+    // prefix must be endswith `/`, so we should trim path here.
+    let relative_path = format!("{prefix}{}", path.trim_start_matches('/'));
+
+    debug!("parsed stage: {stage:?}, path: {relative_path}");
+    Ok((stage, relative_path))
+}
+
+/// TODO(xuanwo): Move those logic into parser
+pub fn parse_copy_file_format_options(
+    file_format_options: &BTreeMap<String, String>,
+) -> Result<FileFormatOptions> {
+    // File format type.
+    let format = file_format_options
+        .get("type")
+        .ok_or_else(|| ErrorCode::SyntaxException("File format type must be specified"))?;
+    let file_format = StageFileFormatType::from_str(format)
+        .map_err(|e| ErrorCode::SyntaxException(format!("File format type error:{:?}", e)))?;
+
+    // Skip header.
+    let skip_header = file_format_options
+        .get("skip_header")
+        .unwrap_or(&"0".to_string())
+        .parse::<u64>()?;
+
+    // Field delimiter.
+    let field_delimiter = parse_escape_string(
+        file_format_options
+            .get("field_delimiter")
+            .unwrap_or(&"".to_string())
+            .as_bytes(),
+    );
+
+    // Record delimiter.
+    let record_delimiter = parse_escape_string(
+        file_format_options
+            .get("record_delimiter")
+            .unwrap_or(&"".to_string())
+            .as_bytes(),
+    );
+
+    // Compression delimiter.
+    let compression = parse_escape_string(
+        file_format_options
+            .get("compression")
+            .unwrap_or(&"none".to_string())
+            .as_bytes(),
+    )
+    .parse()
+    .map_err(ErrorCode::UnknownCompressionType)?;
+
+    Ok(FileFormatOptions {
+        format: file_format,
+        skip_header,
+        field_delimiter,
+        record_delimiter,
+        compression,
+    })
+}

src/query/service/src/sql/planner/binder/ddl/stage.rs

@@ -26,10 +26,10 @@ use common_planner::plans::RemoveStagePlan;
 use common_storage::parse_uri_location;
 use common_storage::UriLocation;
 
+use super::super::copy::parse_copy_file_format_options;
+use super::super::copy::parse_stage_location;
 use crate::sql::binder::Binder;
 use crate::sql::plans::Plan;
-use crate::sql::statements::parse_copy_file_format_options;
-use crate::sql::statements::parse_stage_location;
 
 impl<'a> Binder {
     pub(in crate::sql::planner::binder) async fn bind_list_stage(

src/query/service/src/sql/planner/binder/presign.rs

@@ -18,11 +18,11 @@ use common_ast::ast::PresignStmt;
 use common_exception::Result;
 use time::Duration;
 
+use super::copy::parse_stage_location_v2;
 use crate::sql::binder::Binder;
 use crate::sql::plans::Plan;
 use crate::sql::plans::PresignAction;
 use crate::sql::plans::PresignPlan;
-use crate::sql::statements::parse_stage_location_v2;
 use crate::sql::BindContext;
 
 impl<'a> Binder {