fix(pgwire): preserve procedural BEGIN...END blocks during SQL splitting

The SQL statement splitter treated semicolons inside procedural function
bodies as statement separators, breaking CREATE FUNCTION with IF/ELSIF
control flow. Detect DDL containing a BEGIN...END block and return it
as a single statement without splitting on internal semicolons.

Add tests covering:
- Procedural body with IF/ELSIF/END IF not split on semicolons
- Trailing semicolon after END stripped correctly
- Procedural body parse roundtrip via parse_create_function
- DML rejection validation inside function bodies

Author: farhan-syah

nodedb/src/control/server/pgwire/ddl/function/create.rs

@@ -395,4 +395,49 @@ mod tests {
         let sql = "CREATE FUNCTION f(x INT) RETURNS INT AS";
         assert!(parse_create_function(sql).is_err());
     }
+
+    #[test]
+    fn parse_procedural_body() {
+        let sql = "CREATE FUNCTION classify(score INT) RETURNS TEXT AS \
+                    BEGIN \
+                      IF score > 90 THEN RETURN 'excellent'; \
+                      ELSIF score > 70 THEN RETURN 'good'; \
+                      ELSE RETURN 'needs improvement'; \
+                      END IF; \
+                    END";
+        let parsed = parse_create_function(sql).unwrap();
+        assert_eq!(parsed.name, "classify");
+        assert!(parsed.body_sql.starts_with("BEGIN"));
+
+        // Verify the procedural parser can handle the body.
+        use crate::control::planner::procedural::ast::BodyKind;
+        assert!(matches!(
+            BodyKind::detect(&parsed.body_sql),
+            BodyKind::Procedural
+        ));
+        let block = crate::control::planner::procedural::parse_block(&parsed.body_sql);
+        assert!(block.is_ok(), "procedural parse failed: {:?}", block.err());
+    }
+
+    #[test]
+    fn parse_dml_in_procedural_body() {
+        let sql = "CREATE FUNCTION bad_func(x INT) RETURNS INT AS \
+                    BEGIN INSERT INTO t (id) VALUES (x); RETURN x; END";
+        let parsed = parse_create_function(sql).unwrap();
+
+        use crate::control::planner::procedural::ast::BodyKind;
+        assert!(matches!(
+            BodyKind::detect(&parsed.body_sql),
+            BodyKind::Procedural
+        ));
+        let block = crate::control::planner::procedural::parse_block(&parsed.body_sql).unwrap();
+
+        let result = crate::control::planner::procedural::validate_function_block(&block);
+        assert!(result.is_err(), "should reject DML: {:?}", result);
+        let err_msg = format!("{}", result.unwrap_err());
+        assert!(
+            err_msg.contains("DML"),
+            "error should mention DML, got: {err_msg}"
+        );
+    }
 }

nodedb/src/control/server/pgwire/handler/sql_exec.rs

@@ -441,12 +441,27 @@ impl NodeDbPgHandler {
 /// query message (e.g. heredoc input), ensuring `parts[2]` in DDL handlers
 /// never contains a trailing semicolon.
 pub(super) fn split_sql_statements(sql: &str) -> Vec<String> {
+    // Procedural function bodies (AS BEGIN ... END) contain semicolons that
+    // are NOT statement separators. Detect and return as single statement.
+    let upper = sql.to_uppercase();
+    // Detect procedural blocks: any DDL containing a BEGIN...END body.
+    // Standalone "BEGIN" (transaction) is handled separately before split.
+    let has_procedural_block = (upper.contains(" BEGIN ") || upper.contains(" BEGIN\n"))
+        && upper.trim() != "BEGIN"
+        && !upper.starts_with("BEGIN");
+    if has_procedural_block {
+        let trimmed = sql.trim().trim_end_matches(';').trim().to_string();
+        if trimmed.is_empty() {
+            return Vec::new();
+        }
+        return vec![trimmed];
+    }
+
     let mut stmts = Vec::new();
     let mut current = String::new();
     let mut chars = sql.chars().peekable();
     let mut in_single_quote = false;
     let mut in_double_quote = false;
-    // Simple line-comment skip flag.
     let mut in_line_comment = false;
     let mut in_block_comment = false;
     let mut prev = '\0';
@@ -551,4 +566,24 @@ mod tests {
         let stmts = split_sql_statements(";;  ;SELECT 1;;");
         assert_eq!(stmts, vec!["SELECT 1"]);
     }
+
+    #[test]
+    fn procedural_begin_end_not_split() {
+        let sql = "CREATE FUNCTION f(x INT) RETURNS TEXT AS \
+                    BEGIN IF x > 0 THEN RETURN 'pos'; ELSE RETURN 'neg'; END IF; END";
+        let stmts = split_sql_statements(sql);
+        assert_eq!(stmts.len(), 1);
+        assert!(stmts[0].starts_with("CREATE FUNCTION"));
+        assert!(stmts[0].ends_with("END"));
+    }
+
+    #[test]
+    fn procedural_with_trailing_semicolon() {
+        let sql = "CREATE FUNCTION f(x INT) RETURNS INT AS \
+                    BEGIN RETURN x; END;";
+        let stmts = split_sql_statements(sql);
+        assert_eq!(stmts.len(), 1);
+        assert!(stmts[0].contains("BEGIN"));
+        assert!(stmts[0].ends_with("END"));
+    }
 }