geoffreyclaude commented on code in PR #17843: URL: https://github.com/apache/datafusion/pull/17843#discussion_r2593070576
########## datafusion-examples/examples/relation_planner/match_recognize.rs: ########## @@ -0,0 +1,345 @@ +// 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. + +//! This example demonstrates using custom relation planners to implement +//! MATCH_RECOGNIZE-style pattern matching on event streams. +//! +//! MATCH_RECOGNIZE is a SQL extension for pattern matching on ordered data, +//! similar to regular expressions but for relational data. This example shows +//! how to use custom planners to implement new SQL syntax. + +use std::{any::Any, cmp::Ordering, hash::Hasher, sync::Arc}; + +use arrow::array::{ArrayRef, Float64Array, Int32Array, StringArray}; +use arrow::record_batch::RecordBatch; +use datafusion::prelude::*; +use datafusion_common::{DFSchemaRef, DataFusionError, Result}; +use datafusion_expr::{ + logical_plan::{Extension, InvariantLevel, LogicalPlan}, + planner::{ + PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning, + }, + Expr, UserDefinedLogicalNode, +}; +use datafusion_sql::sqlparser::ast::TableFactor; + +/// This example demonstrates using custom relation planners to implement +/// MATCH_RECOGNIZE-style pattern matching on event streams. +pub async fn match_recognize() -> Result<()> { + let ctx = SessionContext::new(); + + // Register sample data tables + register_sample_data(&ctx)?; + + // Register custom planner + ctx.register_relation_planner(Arc::new(MatchRecognizePlanner))?; + + println!("Custom Relation Planner: MATCH_RECOGNIZE Pattern Matching"); + println!("==========================================================\n"); + + // Example 1: Basic MATCH_RECOGNIZE with MEASURES and DEFINE clauses + // Shows: How to use MATCH_RECOGNIZE to find patterns in event data with aggregations + // Expected: Logical plan showing MiniMatchRecognize node with SUM and AVG measures + // Note: This demonstrates the logical planning phase - actual execution would require physical implementation + // Actual (Logical Plan): + // Projection: t.price + // SubqueryAlias: t + // MiniMatchRecognize measures=[total_price := sum(price), avg_price := avg(price)] define=[a := price > Int64(10)] + // EmptyRelation: rows=0 + run_example( + &ctx, + "Example 1: MATCH_RECOGNIZE with measures and definitions", + r#"SELECT * FROM events + MATCH_RECOGNIZE ( + PARTITION BY 1 + MEASURES SUM(price) AS total_price, AVG(price) AS avg_price + PATTERN (A) + DEFINE A AS price > 10 + ) AS t"#, + ) + .await?; + + // Example 2: Stock price pattern detection using MATCH_RECOGNIZE + // Shows: How to detect patterns in financial data (e.g., stocks above threshold) + // Expected: Logical plan showing MiniMatchRecognize with MIN, MAX, AVG measures on stock prices + // Note: Uses real stock data (DDOG prices: 150, 155, 152, 158) to find patterns above 151.0 + // Actual (Logical Plan): + // Projection: trends.column1, trends.column2 + // SubqueryAlias: trends + // MiniMatchRecognize measures=[min_price := min(column2), max_price := max(column2), avg_price := avg(column2)] define=[high := column2 > Float64(151)] + // Values: (Utf8("DDOG"), Float64(150)), (Utf8("DDOG"), Float64(155)), (Utf8("DDOG"), Float64(152)), (Utf8("DDOG"), Float64(158)) + run_example( + &ctx, + "Example 2: Detect stocks above threshold using MATCH_RECOGNIZE", + r#"SELECT * FROM stock_prices + MATCH_RECOGNIZE ( + MEASURES MIN(column2) AS min_price, + MAX(column2) AS max_price, + AVG(column2) AS avg_price + PATTERN (HIGH) + DEFINE HIGH AS column2 > 151.0 + ) AS trends"#, + ) + .await?; + + Ok(()) +} + +/// Register sample data tables for the examples +fn register_sample_data(ctx: &SessionContext) -> Result<()> { + // Create events table with price column + let price: ArrayRef = Arc::new(Int32Array::from(vec![5, 12, 8, 15, 20])); + let batch = RecordBatch::try_from_iter(vec![("price", price)])?; + ctx.register_batch("events", batch)?; + + // Create stock_prices table with symbol and price columns + let symbol: ArrayRef = + Arc::new(StringArray::from(vec!["DDOG", "DDOG", "DDOG", "DDOG"])); + let price: ArrayRef = Arc::new(Float64Array::from(vec![150.0, 155.0, 152.0, 158.0])); + let batch = + RecordBatch::try_from_iter(vec![("column1", symbol), ("column2", price)])?; Review Comment: That makes total sense! done ########## datafusion-examples/examples/relation_planner/match_recognize.rs: ########## @@ -0,0 +1,345 @@ +// 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. + +//! This example demonstrates using custom relation planners to implement +//! MATCH_RECOGNIZE-style pattern matching on event streams. +//! +//! MATCH_RECOGNIZE is a SQL extension for pattern matching on ordered data, +//! similar to regular expressions but for relational data. This example shows +//! how to use custom planners to implement new SQL syntax. + +use std::{any::Any, cmp::Ordering, hash::Hasher, sync::Arc}; + +use arrow::array::{ArrayRef, Float64Array, Int32Array, StringArray}; +use arrow::record_batch::RecordBatch; +use datafusion::prelude::*; +use datafusion_common::{DFSchemaRef, DataFusionError, Result}; +use datafusion_expr::{ + logical_plan::{Extension, InvariantLevel, LogicalPlan}, + planner::{ + PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning, + }, + Expr, UserDefinedLogicalNode, +}; +use datafusion_sql::sqlparser::ast::TableFactor; + +/// This example demonstrates using custom relation planners to implement +/// MATCH_RECOGNIZE-style pattern matching on event streams. +pub async fn match_recognize() -> Result<()> { + let ctx = SessionContext::new(); + + // Register sample data tables + register_sample_data(&ctx)?; + + // Register custom planner + ctx.register_relation_planner(Arc::new(MatchRecognizePlanner))?; + + println!("Custom Relation Planner: MATCH_RECOGNIZE Pattern Matching"); + println!("==========================================================\n"); + + // Example 1: Basic MATCH_RECOGNIZE with MEASURES and DEFINE clauses + // Shows: How to use MATCH_RECOGNIZE to find patterns in event data with aggregations + // Expected: Logical plan showing MiniMatchRecognize node with SUM and AVG measures + // Note: This demonstrates the logical planning phase - actual execution would require physical implementation + // Actual (Logical Plan): + // Projection: t.price + // SubqueryAlias: t + // MiniMatchRecognize measures=[total_price := sum(price), avg_price := avg(price)] define=[a := price > Int64(10)] + // EmptyRelation: rows=0 + run_example( + &ctx, + "Example 1: MATCH_RECOGNIZE with measures and definitions", + r#"SELECT * FROM events + MATCH_RECOGNIZE ( + PARTITION BY 1 + MEASURES SUM(price) AS total_price, AVG(price) AS avg_price + PATTERN (A) + DEFINE A AS price > 10 + ) AS t"#, + ) + .await?; + + // Example 2: Stock price pattern detection using MATCH_RECOGNIZE + // Shows: How to detect patterns in financial data (e.g., stocks above threshold) + // Expected: Logical plan showing MiniMatchRecognize with MIN, MAX, AVG measures on stock prices + // Note: Uses real stock data (DDOG prices: 150, 155, 152, 158) to find patterns above 151.0 + // Actual (Logical Plan): + // Projection: trends.column1, trends.column2 + // SubqueryAlias: trends + // MiniMatchRecognize measures=[min_price := min(column2), max_price := max(column2), avg_price := avg(column2)] define=[high := column2 > Float64(151)] + // Values: (Utf8("DDOG"), Float64(150)), (Utf8("DDOG"), Float64(155)), (Utf8("DDOG"), Float64(152)), (Utf8("DDOG"), Float64(158)) + run_example( + &ctx, + "Example 2: Detect stocks above threshold using MATCH_RECOGNIZE", + r#"SELECT * FROM stock_prices + MATCH_RECOGNIZE ( + MEASURES MIN(column2) AS min_price, + MAX(column2) AS max_price, Review Comment: done -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
