feat(datafusion): support RLIKE and REGEXP operators

crates/integrations/datafusion/src/catalog.rs

@@ -1426,8 +1426,9 @@ async fn validate_view_dependencies(
 }
 
 fn view_relation_identifiers(view: &View, catalog_name: &str) -> DFResult<Vec<Identifier>> {
+    let parseable = crate::regex_operators::parseable_sql(view.query_for("datafusion"))?;
     let statements =
-        Parser::parse_sql(&GenericDialect {}, view.query_for("datafusion")).map_err(|error| {
+        Parser::parse_sql(&GenericDialect {}, parseable.as_ref()).map_err(|error| {
             plan_datafusion_err!(
                 "Invalid SQL for REST catalog view '{}': {error}",
                 view.full_name()

crates/integrations/datafusion/src/lib.rs

@@ -62,6 +62,7 @@ mod merge_into;
 mod partition_count_pushdown;
 mod physical_plan;
 mod procedures;
+mod regex_operators;
 mod relation_planner;
 pub mod runtime;
 mod scalar_compat;

crates/integrations/datafusion/src/regex_operators.rs

@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::{DataFusionError, Result};
+use datafusion::sql::sqlparser::ast::{visit_expressions_mut, BinaryOperator, Expr, Statement};
+use datafusion::sql::sqlparser::{
+    dialect::GenericDialect,
+    keywords::Keyword,
+    tokenizer::{Token, Tokenizer},
+};
+use std::borrow::Cow;
+use std::ops::ControlFlow;
+
+/// Work around sqlparser 0.62 consuming a bare NULL regex pattern before parsing it.
+pub(crate) fn parseable_sql(sql: &str) -> Result<Cow<'_, str>> {
+    let tokens = Tokenizer::new(&GenericDialect {}, sql)
+        .tokenize_with_location()
+        .map_err(|error| DataFusionError::Plan(format!("SQL parse error: {error}")))?;
+    let mut regex = false;
+    let mut operand = false;
+    let mut pending_not = false;
+    let mut positions = Vec::new();
+    for token in tokens {
+        if matches!(token.token, Token::Whitespace(_)) {
+            continue;
+        }
+        if regex && matches!(&token.token,Token::Word(w) if w.keyword==Keyword::NULL) {
+            let mut offset = 0;
+            for (index, line) in sql.split_inclusive('\n').enumerate() {
+                if index + 1 == token.span.start.line as usize {
+                    offset += line
+                        .char_indices()
+                        .nth(token.span.start.column as usize - 1)
+                        .map(|(i, _)| i)
+                        .unwrap_or(line.len());
+                    break;
+                }
+                offset += line.len();
+            }
+            positions.push(offset);
+        }
+        if matches!(&token.token,Token::Word(w) if w.quote_style.is_none() && matches!(w.keyword,Keyword::RLIKE|Keyword::REGEXP))
+        {
+            regex = operand || pending_not;
+            operand = !regex;
+            pending_not = false;
+        } else if matches!(&token.token,Token::Word(w) if w.quote_style.is_none() && w.keyword == Keyword::NOT)
+        {
+            pending_not = operand;
+            operand = false;
+            regex = false;
+        } else {
+            operand = expression_end(&token.token);
+            pending_not = false;
+            regex = false;
+        }
+    }
+    if positions.is_empty() {
+        return Ok(Cow::Borrowed(sql));
+    }
+    let mut output = sql.to_string();
+    for offset in positions.into_iter().rev() {
+        output.replace_range(offset..offset + 4, "(NULL)");
+    }
+    Ok(Cow::Owned(output))
+}
+
+fn expression_end(token: &Token) -> bool {
+    match token {
+        Token::Word(word) => {
+            word.keyword == Keyword::END
+                || (!datafusion::sql::sqlparser::keywords::RESERVED_FOR_COLUMN_ALIAS
+                    .contains(&word.keyword)
+                    && !matches!(
+                        word.keyword,
+                        Keyword::AND
+                            | Keyword::OR
+                            | Keyword::NOT
+                            | Keyword::WHEN
+                            | Keyword::THEN
+                            | Keyword::ELSE
+                            | Keyword::AS
+                            | Keyword::ON
+                            | Keyword::BY
+                            | Keyword::CASE
+                            | Keyword::IN
+                            | Keyword::IS
+                            | Keyword::LIKE
+                            | Keyword::ILIKE
+                            | Keyword::RLIKE
+                            | Keyword::REGEXP
+                            | Keyword::BETWEEN
+                            | Keyword::DISTINCT
+                            | Keyword::ALL
+                    ))
+        }
+        Token::Number(..)
+        | Token::SingleQuotedString(_)
+        | Token::DoubleQuotedString(_)
+        | Token::EscapedStringLiteral(_)
+        | Token::NationalStringLiteral(_)
+        | Token::HexStringLiteral(_)
+        | Token::DollarQuotedString(_)
+        | Token::Placeholder(_)
+        | Token::RParen
+        | Token::RBracket => true,
+        _ => false,
+    }
+}
+
+/// Lower regex operators to DataFusion's native regex operators after parsing.
+pub(crate) fn rewrite(statement: &mut Statement) -> bool {
+    let mut changed = false;
+    let _: ControlFlow<()> = visit_expressions_mut(statement, |node| {
+        if let Expr::RLike {
+            negated,
+            expr,
+            pattern,
+            ..
+        } = node
+        {
+            *node = Expr::Nested(Box::new(Expr::BinaryOp {
+                left: expr.clone(),
+                op: if *negated {
+                    BinaryOperator::PGRegexNotMatch
+                } else {
+                    BinaryOperator::PGRegexMatch
+                },
+                right: pattern.clone(),
+            }));
+            changed = true;
+        }
+        ControlFlow::Continue(())
+    });
+    changed
+}

crates/integrations/datafusion/src/sql_context.rs

@@ -498,7 +498,12 @@ impl SQLContext {
         let rewritten;
         let persistent_definition = matches!(&statements[0], Statement::CreateView(view) if !view.temporary && self.is_paimon_catalog_ref(&TableReference::from(view.name.to_string().as_str())))
             || matches!(&statements[0], Statement::CreateFunction(_));
-        let sql = if !persistent_definition && crate::binary_cast::rewrite(&mut statements[0])? {
+        let changed = !persistent_definition && {
+            let binary_changed = crate::binary_cast::rewrite(&mut statements[0])?;
+            let regex_changed = crate::regex_operators::rewrite(&mut statements[0]);
+            binary_changed || regex_changed
+        };
+        let sql = if changed {
             rewritten = statements[0].to_string();
             rewritten.as_str()
         } else {
@@ -2687,6 +2692,8 @@ fn validate_persistent_create_function(create_function: &CreateFunction) -> DFRe
 }
 
 fn parse_sql_statements(sql: &str) -> DFResult<Vec<Statement>> {
+    let parseable = crate::regex_operators::parseable_sql(sql)?;
+    let sql = parseable.as_ref();
     let dialect = GenericDialect {};
     let mut tokens = Tokenizer::new(&dialect, sql)
         .tokenize_with_location()
@@ -6010,6 +6017,45 @@ mod tests {
         assert_eq!(answers.value(0), 42);
     }
 
+    #[tokio::test]
+    async fn regex_operators_preserve_stored_sql_and_existing_null_definitions() {
+        let catalog = Arc::new(MockCatalog::new());
+        add_unary_sql_function(
+            &catalog,
+            "null_pattern",
+            "CAST(x AS VARCHAR) RLIKE NULL",
+            true,
+        );
+        add_bigint_view(
+            &catalog,
+            "default",
+            "null_view",
+            "SELECT CAST('abc' RLIKE NULL AS BIGINT) AS answer",
+        );
+        let ctx = make_sql_context(catalog.clone()).await;
+        ctx.sql("CREATE VIEW regex_view AS SELECT 'abc' RLIKE '^a' AS flag")
+            .await
+            .unwrap();
+        let view = catalog
+            .get_view(&Identifier::new("default", "regex_view"))
+            .await
+            .unwrap();
+        assert!(view.query_for("spark").contains("RLIKE"));
+        assert!(!view.query_for("spark").contains(" ~ "));
+        for sql in [
+            "SELECT flag FROM regex_view",
+            "SELECT null_pattern(1) IS NULL",
+            "SELECT answer IS NULL FROM null_view",
+        ] {
+            let rows = ctx.sql(sql).await.unwrap().collect().await.unwrap();
+            assert_eq!(
+                ScalarValue::try_from_array(rows[0].column(0), 0).unwrap(),
+                ScalarValue::Boolean(Some(true)),
+                "{sql}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn rest_sql_function_result_is_cast_to_declared_type() {
         let catalog = Arc::new(MockCatalog::new());

crates/integrations/datafusion/src/sql_function.rs

@@ -61,7 +61,8 @@ async fn expand_sql_internal(
     current_database: &str,
     candidate: Option<&Function>,
 ) -> DFResult<String> {
-    let mut statements = Parser::parse_sql(&GenericDialect {}, sql)
+    let parseable = crate::regex_operators::parseable_sql(sql)?;
+    let mut statements = Parser::parse_sql(&GenericDialect {}, parseable.as_ref())
         .map_err(|error| DataFusionError::Plan(format!("Invalid REST SQL: {error}")))?;
     if statements.len() != 1 {
         return Err(DataFusionError::Plan(format!(
@@ -244,6 +245,7 @@ async fn expand_statement_internal(
         }
         if expanded_count == 0 {
             crate::binary_cast::rewrite(&mut statement)?;
+            crate::regex_operators::rewrite(&mut statement);
             return Ok(statement);
         }
     }
@@ -265,7 +267,10 @@ fn function_dependencies(
     else {
         return Vec::new();
     };
-    let Ok(mut parser) = Parser::new(&GenericDialect {}).try_with_sql(definition) else {
+    let Ok(parseable) = crate::regex_operators::parseable_sql(definition) else {
+        return Vec::new();
+    };
+    let Ok(mut parser) = Parser::new(&GenericDialect {}).try_with_sql(parseable.as_ref()) else {
         return Vec::new();
     };
     let Ok(body) = parser.parse_expr() else {
@@ -440,8 +445,9 @@ fn expand_call(
         )));
     }
 
+    let parseable = crate::regex_operators::parseable_sql(definition)?;
     let mut body = Parser::new(&GenericDialect {})
-        .try_with_sql(definition)
+        .try_with_sql(parseable.as_ref())
         .map_err(|error| {
             DataFusionError::Plan(format!("Invalid SQL function definition: {error}"))
         })?

crates/integrations/datafusion/tests/regex_operators.rs

@@ -0,0 +1,79 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::ScalarValue;
+use paimon_datafusion::SQLContext;
+#[tokio::test]
+async fn regex_operators_preserve_boolean_and_null_semantics() {
+    let ctx = SQLContext::new();
+    for expression in [
+        "'abc' RLIKE '^a'",
+        "'abc' REGEXP '^a'",
+        "CAST(CAST('abc' AS VARBINARY(2)) AS STRING) RLIKE '^ab$'",
+        "'abc' NOT RLIKE '^z'",
+        "('abc' NOT RLIKE '^a') = FALSE",
+        "('abc' NOT REGEXP '^a') = FALSE",
+        "(NULL RLIKE '^a') IS NULL",
+        "('abc' RLIKE NULL) IS NULL",
+        "'汉字' RLIKE '^汉'",
+        "'Ab' RLIKE '(?i)^a'",
+        "'abc RLIKE x' = 'abc RLIKE x'",
+    ] {
+        let batches = ctx
+            .sql(&format!("SELECT {expression}"))
+            .await
+            .unwrap_or_else(|error| panic!("{expression}: {error}"))
+            .collect()
+            .await
+            .unwrap();
+        assert_eq!(
+            ScalarValue::try_from_array(batches[0].column(0), 0).unwrap(),
+            ScalarValue::Boolean(Some(true)),
+            "{expression}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn regex_operators_cover_native_views_dynamic_inputs_and_parser_boundaries() {
+    let ctx = SQLContext::new();
+    ctx.sql("CREATE VIEW regex_view AS SELECT 'abc' RLIKE '^a' AS matched")
+        .await
+        .unwrap();
+    for sql in [
+        "SELECT matched FROM regex_view",
+        "SELECT actual IS NOT DISTINCT FROM expected FROM (SELECT s RLIKE p AS actual, expected FROM (VALUES ('abc','^a',TRUE), ('abc','^z',FALSE), (NULL,'^a',NULL), ('abc',NULL,NULL)) t(s,p,expected)) q",
+        "SELECT regexp IS NOT DISTINCT FROM 'x' FROM (SELECT regexp NULL FROM (VALUES ('x')) AS t(regexp)) q(regexp)",
+        "SELECT original = 'x' AND matched IS NULL FROM (SELECT regexp NULL, ('abc' RLIKE NULL) AS matched FROM (VALUES ('x')) t(regexp)) q(original,matched)",
+        "SELECT ('汉' RLIKE /* comment */ NULL) IS NULL",
+        "SELECT (regexp RLIKE NULL) IS NULL FROM (VALUES ('x')) t(regexp)",
+        "SELECT (rlike NOT REGEXP NULL) IS NULL FROM (VALUES ('x')) t(rlike)",
+        "SELECT ('汉' NOT REGEXP\nNULL) IS NULL",
+    ] {
+        let batches=ctx.sql(sql).await.unwrap_or_else(|e|panic!("{sql}: {e}")).collect().await.unwrap();
+        for batch in batches {for row in 0..batch.num_rows() {
+            assert_eq!(ScalarValue::try_from_array(batch.column(0),row).unwrap(),ScalarValue::Boolean(Some(true)),"{sql}");
+        }}
+    }
+    let result = ctx
+        .sql("SELECT s RLIKE p FROM (VALUES ('a', '[')) t(s,p)")
+        .await;
+    match result {
+        Err(_) => {}
+        Ok(frame) => assert!(frame.collect().await.is_err()),
+    }
+}

docs/src/sql.md

@@ -2044,6 +2044,25 @@ column masks it returns (see `create_policy`) are applied to the result:
 - Time travel, branches, system tables, incremental and audit-log reads, and
   search are refused on such tables.
 
+### Regular expression operators
+
+SQLContext supports `value RLIKE pattern`, `value REGEXP pattern`, and their
+`NOT` forms in queries, expressions, views and SQL functions. They use
+DataFusion's native case-sensitive regex matching: a match may occur anywhere
+unless the pattern contains anchors. Inline flags such as `(?i)` can change
+case sensitivity.
+
+```sql
+SELECT phone RLIKE '^1[3-9][0-9]{9}$' FROM users;
+SELECT name NOT REGEXP '(?i)^test';
+```
+
+Patterns use Rust regex syntax, including its restrictions on look-around and
+backreferences. A NULL value or pattern produces SQL NULL; an invalid pattern
+reports an error. Operator text inside strings, comments and quoted identifiers
+is not interpreted as SQL syntax. Persistent Paimon view and function definitions
+retain their portable SQL text.
+
 ## Python Multimodal Helper Functions
 
 When you use `pypaimon_rust.datafusion.SQLContext`, the Python binding registers a small set of scalar helper functions for BLOB-backed media and vector workflows. These helpers are Python-binding built-ins; they are not registered by the Rust `paimon_datafusion::SQLContext`.