feat: Make VALIDATE clause optional when creating webhook connection

e2e_test/webhook/create_table.slt.part

@@ -1,3 +1,10 @@
+statement ok
+create table optional_validate (
+  data JSONB
+) WITH (
+  connector = 'webhook'
+);
+
 statement ok
 create table rudderstack (
   data JSONB

src/frontend/src/handler/create_table.rs

@@ -2565,27 +2565,36 @@ fn bind_webhook_info(
         (None, None)
     };
 
-    let secure_compare_context = SecureCompareContext {
-        column_name: columns_defs[0].name.real_value(),
-        secret_name,
-    };
-    let mut binder = Binder::new_for_ddl(session).with_secure_compare(secure_compare_context);
-    let expr = binder.bind_expr(&signature_expr)?;
+    let signature_expr = if let Some(signature_expr) = signature_expr {
+        let secure_compare_context = SecureCompareContext {
+            column_name: columns_defs[0].name.real_value(),
+            secret_name,
+        };
+        let mut binder = Binder::new_for_ddl(session).with_secure_compare(secure_compare_context);
+        let expr = binder.bind_expr(&signature_expr)?;
 
-    // validate expr, ensuring it is SECURE_COMPARE()
-    if expr.as_function_call().is_none()
-        || expr.as_function_call().unwrap().func_type()
-            != crate::optimizer::plan_node::generic::ExprType::SecureCompare
-    {
-        return Err(ErrorCode::InvalidInputSyntax(
-            "The signature verification function must be SECURE_COMPARE()".to_owned(),
-        )
-        .into());
-    }
+        // validate expr, ensuring it is SECURE_COMPARE()
+        if expr.as_function_call().is_none()
+            || expr.as_function_call().unwrap().func_type()
+                != crate::optimizer::plan_node::generic::ExprType::SecureCompare
+        {
+            return Err(ErrorCode::InvalidInputSyntax(
+                "The signature verification function must be SECURE_COMPARE()".to_owned(),
+            )
+            .into());
+        }
+
+        Some(expr.to_expr_proto())
+    } else {
+        session.notice_to_user(
+            "VALIDATE clause is strongly recommended for safety or production usages",
+        );
+        None
+    };
 
     let pb_webhook_info = PbWebhookSourceInfo {
         secret_ref: pb_secret_ref,
-        signature_expr: Some(expr.to_expr_proto()),
+        signature_expr,
         wait_for_persistence,
         is_batched,
     };

src/frontend/src/webhook/mod.rs

@@ -89,26 +89,30 @@ pub(super) mod handlers {
             is_batched,
         } = webhook_source_info;
 
-        let secret_string = if let Some(secret_ref) = secret_ref {
-            LocalSecretManager::global()
-                .fill_secret(secret_ref)
-                .map_err(|e| err(e, StatusCode::NOT_FOUND))?
+        let is_valid = if let Some(signature_expr) = signature_expr {
+            let secret_string = if let Some(secret_ref) = secret_ref {
+                LocalSecretManager::global()
+                    .fill_secret(secret_ref)
+                    .map_err(|e| err(e, StatusCode::NOT_FOUND))?
+            } else {
+                String::new()
+            };
+
+            // Once limitation here is that the key is no longer case-insensitive, users must user the lowercase key when defining the webhook source table.
+            let headers_jsonb = header_map_to_json(&headers);
+
+            // verify the signature
+            verify_signature(
+                headers_jsonb,
+                secret_string.as_str(),
+                body.as_ref(),
+                signature_expr,
+            )
+            .await?
         } else {
-            String::new()
+            true
         };
 
-        // Once limitation here is that the key is no longer case-insensitive, users must user the lowercase key when defining the webhook source table.
-        let headers_jsonb = header_map_to_json(&headers);
-
-        // verify the signature
-        let is_valid = verify_signature(
-            headers_jsonb,
-            secret_string.as_str(),
-            body.as_ref(),
-            signature_expr.unwrap(),
-        )
-        .await?;
-
         if !is_valid {
             return Err(err(
                 anyhow!("Signature verification failed"),

src/sqlparser/src/ast/ddl.rs

@@ -1034,7 +1034,7 @@ impl fmt::Display for ReferentialAction {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct WebhookSourceInfo {
     pub secret_ref: Option<SecretRefValue>,
-    pub signature_expr: Expr,
+    pub signature_expr: Option<Expr>,
     pub wait_for_persistence: bool,
     pub is_batched: bool,
 }

src/sqlparser/src/ast/mod.rs

@@ -2110,13 +2110,15 @@ impl Statement {
                     write!(f, " FROM {}", info.source_name)?;
                     write!(f, " TABLE '{}'", info.external_table_name)?;
                 }
-                if let Some(info) = webhook_info {
+                if let Some(info) = webhook_info
+                    && let Some(signature_expr) = &info.signature_expr
+                {
                     if let Some(secret) = &info.secret_ref {
                         write!(f, " VALIDATE SECRET {}", secret.secret_name)?;
                     } else {
                         write!(f, " VALIDATE")?;
                     }
-                    write!(f, " AS {}", info.signature_expr)?;
+                    write!(f, " AS {}", signature_expr)?;
                 }
                 match engine {
                     Engine::Hummock => {}

src/sqlparser/src/parser.rs

@@ -2600,16 +2600,22 @@ impl Parser<'_> {
             None
         };
 
+        let webhook_wait_for_persistence = with_options
+            .iter()
+            .find(|&opt| opt.name.real_value() == WEBHOOK_WAIT_FOR_PERSISTENCE)
+            .map(|opt| opt.value.to_string().eq_ignore_ascii_case("true"))
+            .unwrap_or(true);
+        let webhook_is_batched = with_options
+            .iter()
+            .find(|&opt| opt.name.real_value() == WEBHOOK_IS_BATCHED)
+            .map(|opt| opt.value.to_string().eq_ignore_ascii_case("true"))
+            .unwrap_or(false);
+
         let webhook_info = if self.parse_keyword(Keyword::VALIDATE) {
             if !contain_webhook {
                 parser_err!("VALIDATE is only supported for tables created with webhook source");
             }
 
-            let wait_for_persistence = with_options
-                .iter()
-                .find(|&opt| opt.name.real_value() == WEBHOOK_WAIT_FOR_PERSISTENCE)
-                .map(|opt| opt.value.to_string().eq_ignore_ascii_case("true"))
-                .unwrap_or(true);
             let secret_ref = if self.parse_keyword(Keyword::SECRET) {
                 let secret_ref = self.parse_secret_ref()?;
                 if secret_ref.ref_as == SecretRefAsType::File {
@@ -2623,17 +2629,18 @@ impl Parser<'_> {
             self.expect_keyword(Keyword::AS)?;
             let signature_expr = self.parse_function()?;
 
-            let is_batched = with_options
-                .iter()
-                .find(|&opt| opt.name.real_value() == WEBHOOK_IS_BATCHED)
-                .map(|opt| opt.value.to_string().eq_ignore_ascii_case("true"))
-                .unwrap_or(false);
-
             Some(WebhookSourceInfo {
                 secret_ref,
-                signature_expr,
-                wait_for_persistence,
-                is_batched,
+                signature_expr: Some(signature_expr),
+                wait_for_persistence: webhook_wait_for_persistence,
+                is_batched: webhook_is_batched,
+            })
+        } else if contain_webhook {
+            Some(WebhookSourceInfo {
+                secret_ref: None,
+                signature_expr: None,
+                wait_for_persistence: webhook_wait_for_persistence,
+                is_batched: webhook_is_batched,
             })
         } else {
             None