Skip to content

[FLINK-39293][table-planner] Fix MATCH_RECOGNIZE uses through view #27907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ldadima wants to merge 1 commit into from
+61 −20
Closed

[FLINK-39293][table-planner] Fix MATCH_RECOGNIZE uses through view #27907

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...c/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ static CatalogView toCatalogView(
context.getSqlValidator().getNamespace(validateQuery);
validateDuplicatedColumnNames(validateQuery, viewFields, validatedNamespace);

String expandedQuery = context.toQuotedSqlString(validateQuery);
String expandedQuery = context.expandSqlIdentifiers(originalQuery);
Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ideally we should keep it as is as it gives fully expanded sql and expandSqlIdentifiers doesn't do that (for instance it doesn't expand star).

I tend to think that rather than fixing here it should be fixed in Calcite
for more details look at https://issues.apache.org/jira/browse/CALCITE-7465, https://issues.apache.org/jira/browse/CALCITE-7466, https://issues.apache.org/jira/browse/CALCITE-7467, probably something else

Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

also https://issues.apache.org/jira/browse/CALCITE-7470

Copy link
Copy Markdown
Contributor Author

@ldadima ldadima Apr 13, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem, as I see it, is somewhat similar to the LATERAL issue: expandSqlIdentifiers was being used until the Calcite version (problem resolved) was updated.

Is correct usage of MATCH_RECOGNIZE in CREATE VIEW syntax important to us? If so, we can keep using expandSqlIdentifiers until upgrading Calcite version (1.42.0).
What do you think?

Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the LATERAL issue: expandSqlIdentifiers was being used until the Calcite version (problem resolved) was updated

this is not entirely true
LATERAL was fixed in https://issues.apache.org/jira/browse/CALCITE-7217 and https://issues.apache.org/jira/browse/CALCITE-7312 which are 1.41.0 and 1.42.0 (not even released yet) and Flink master depends on 1.36.0

Calcite upgrade task is not an easy one because of lots of customizations
So we just port fixes from Calcite and remove such fixes when upgrade to certain version
for that reason you can find calcite classes in Flink repo e.g. here https://github.com/apache/flink/tree/master/flink-table/flink-table-planner/src/main/java/org/apache/calcite

Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 13, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would follow similar approach for MATCH_RECOGNIZE however first need to be sure that all cases are fixed in Calcite, for now I'm aware of at least 3-5 more cases which have to be fixed
for instance https://issues.apache.org/jira/browse/CALCITE-7471

Copy link
Copy Markdown
Contributor Author

@ldadima ldadima Apr 13, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, indeed, you are right.
But then I don't understand why solution for FLINK-38493 is correct. Can you help me to understand?

Code before FLINK-38493
image

Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it was a workaround because of a bug with LATERAL in Calcite and there was a huge comment trying to explain why we need it.
The problem became more serious with MATERIALIZED TABLEs where we might want to use expanded query.
Also if you try to look at other vendors: they also expand star which Flink didn't do.

With this issue we see that we are in a lack of tests for MATCH_RECOGNIZE (looks like Calcite as well) and that's why it was not detected earlier.

for short term we could consider rolling it back, however so far nobody was heavily complaining that hot fix is urgently required. In this case I think we still have time to fix it in a proper way

Copy link
Copy Markdown
Contributor Author

@ldadima ldadima Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for explaining
Now it's clear

Copy link
Copy Markdown

@nictownsend nictownsend Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi both - thanks for the discussion and clarification.

for short term we could consider rolling it back, however so far nobody was heavily complaining that hot fix is urgently required. In this case I think we still have time to fix it in a proper way

I tend to build my SQL as a series of views, making it more composable and easier to read. E.g (pseudo):

CREATE TABLE source_1:
CREATE TABLE source_2;
CREATE TEMPORARY VIEW joined as (source_1 JOIN source_2);
CREATE TEMPORARY_VIEW matches as (joined MATCH_RECOGNIZE);
CREATE TABLE sink_1;
CREATE TABLE sink_1;
INSERT into sink_1 select * from matches;
INSERT into sink_2 select * from matches WHERE x=y;

Are there other ways to achieve the same effect without this bug being fixed? My concern is that I now have to write (pseduo):

CREATE TABLE source_1:
CREATE TABLE source_2;
CREATE TABLE sink_1;
CREATE TABLE sink_2;
INSERT into sink_1 select * from (source_1 JOIN source_1) MATCH_RECOGNIZE;
INSERT into sink_2 select * from ((source_1 JOIN source_1) MATCH_RECOGNIZE) WHERE x=y;

Copy link
Copy Markdown
Contributor

@snuyanzin snuyanzin Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

for. now it is impossible if you use 2.2.0
the fixes in Calcite are still in progress however not ported to Flink repo yet (since not all cases are fixed)

I hope to have it done this/next week, however it means that the fixes might land to Flink branches only. Releases should be done separately


PlannerQueryOperation operation = toQueryOperation(validateQuery, context);
ResolvedSchema schema = operation.getResolvedSchema();
Expand Down
42 changes: 42 additions & 0 deletions .../src/test/java/org/apache/flink/table/planner/runtime/batch/sql/MatchRecognizeITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,48 @@ void testSimplePatternInProcTime() {
.containsExactly(Row.of(6, 7, 8));
}

@Test
void testSimplePatternInProcTimeWithTemporaryView() {
tEnv.createTemporaryView(
"MyTable",
tEnv.fromDataStream(
env.fromData(
Row.of(1, "a"),
Row.of(2, "z"),
Row.of(3, "b"),
Row.of(4, "c"),
Row.of(5, "d"),
Row.of(6, "a"),
Row.of(7, "b"),
Row.of(8, "c"),
Row.of(9, "h"))
.returns(ROW_NAMED(new String[] {"id", "name"}, INT, STRING)),
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("name", DataTypes.STRING())
.columnByExpression("proctime", "PROCTIME()")
.build()));
tEnv.executeSql(
"CREATE TEMPORARY VIEW test_t AS \n"
+ "SELECT T.aid, T.bid, T.cid\n"
+ "FROM MyTable\n"
+ "MATCH_RECOGNIZE (\n"
+ " ORDER BY proctime\n"
+ " MEASURES\n"
+ " `A\"`.id AS aid,\n"
+ " \u006C.id AS bid,\n"
+ " C.id AS cid\n"
+ " PATTERN (`A\"` \u006C C)\n"
+ " DEFINE\n"
+ " `A\"` AS name = 'a',\n"
+ " \u006C AS name = 'b',\n"
+ " C AS name = 'c'\n"
+ ") AS T");
TableResult tableResult = tEnv.executeSql("SELECT * FROM test_t");
assertThat(CollectionUtil.iteratorToList(tableResult.collect()))
.containsExactly(Row.of(6, 7, 8));
}

@Test
void testSimplePatternInEventTime() {
Instant now = Instant.parse("2023-12-01T12:00:00.000Z");
Expand Down
27 changes: 13 additions & 14 deletions ...ble-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala
View file
Open in desktop
Copy link
Copy Markdown
Contributor

@raminqaf raminqaf Apr 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are these changes related to the issue yo are trying to solve?

Copy link
Copy Markdown
Contributor Author

@ldadima ldadima Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Essentially, this pull request is a partial revert of FLINK-38493, so I also had to revert the test changes.
But, as you can see above in PR's discussion, was decided to close PR

Original file line number Diff line number Diff line change
Expand Up @@ -356,7 +356,7 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `c`
|)
|AS SELECT `T1`.`a`, `T1`.`b`, `T1`.`c`
|FROM `default_catalog`.`default_database`.`T1` AS `T1`
|FROM `default_catalog`.`default_database`.`T1`
|""".stripMargin
)
)
Expand All @@ -376,7 +376,7 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `f`
|)
|AS SELECT `T1`.`a`, `T1`.`b`, `T1`.`c`
|FROM `default_catalog`.`default_database`.`T1` AS `T1`
|FROM `default_catalog`.`default_database`.`T1`
|""".stripMargin
)
)
Expand All @@ -403,7 +403,7 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `c`
|)
|AS SELECT `T1`.`a`, `T1`.`b`, `T1`.`c`
|FROM `default_catalog`.`default_database`.`T1` AS `T1`
|FROM `default_catalog`.`default_database`.`T1`
|""".stripMargin
)
)
Expand All @@ -423,7 +423,7 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `z`
|)
|AS SELECT `T1`.`a`, `T1`.`b`, `T1`.`c`
|FROM `default_catalog`.`default_database`.`T1` AS `T1`
|FROM `default_catalog`.`default_database`.`T1`
|""".stripMargin
)
)
Expand All @@ -450,8 +450,8 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `max_value`
|)
|AS SELECT MAX(`t1`.`a`) AS `max_value`
|FROM `default_catalog`.`default_database`.`t1` AS `t1`
|LEFT JOIN `default_catalog`.`default_database`.`t2` AS `t2` ON `t1`.`c` = `t2`.`c`
|FROM `default_catalog`.`default_database`.`t1`
|LEFT JOIN `default_catalog`.`default_database`.`t2` ON `t1`.`c` = `t2`.`c`
|""".stripMargin
)
)
Expand Down Expand Up @@ -485,8 +485,8 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `b2`
|)
|AS SELECT `udfEqualsOne`() AS `a`, `t1`.`a` AS `a1`, `t2`.`b` AS `b2`
|FROM `default_catalog`.`default_database`.`t1` AS `t1`
|CROSS JOIN `default_catalog`.`default_database`.`t2` AS `t2`
|FROM `default_catalog`.`default_database`.`t1`
|CROSS JOIN `default_catalog`.`default_database`.`t2`
|""".stripMargin
)
)
Expand Down Expand Up @@ -515,8 +515,8 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
| `b2`
|)
|AS SELECT `t1`.`a` AS `a1`, `t2`.`b` AS `b2`
|FROM `default_catalog`.`default_database`.`t1` AS `t1`
|INNER JOIN `default_catalog`.`default_database`.`t2` AS `t2` ON `t1`.`c` = `t2`.`c`
|FROM `default_catalog`.`default_database`.`t1`
|INNER JOIN `default_catalog`.`default_database`.`t2` ON `t1`.`c` = `t2`.`c`
|""".stripMargin
)
)
Expand All @@ -543,10 +543,9 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase {
s"""CREATE VIEW `default_catalog`.`default_database`.`tumbleWindowViewWithOrderBy` (
| `a`
|)
|AS SELECT `EXPR$$0`.`a`
|FROM TABLE(TUMBLE((SELECT `t1`.`a`, `t1`.`b`, `t1`.`c`, `t1`.`ts`
|FROM `default_catalog`.`default_database`.`t1` AS `t1`), DESCRIPTOR(`ts`), INTERVAL '1' MINUTE)) AS `EXPR$$0`
|ORDER BY `EXPR$$0`.`ts`
|AS SELECT `a`
|FROM TABLE(TUMBLE(TABLE `default_catalog`.`default_database`.`t1`, DESCRIPTOR(`ts`), INTERVAL '1' MINUTE))
|ORDER BY `ts`
|""".stripMargin
)
)
Expand Down
10 changes: 5 additions & 5 deletions ...lanner/src/test/scala/org/apache/flink/table/planner/plan/common/ViewsExpandingTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ class ViewsExpandingTest(tableTestUtil: TableTestBase => TableTestUtil) extends
.getTable(objectID.toObjectPath)
assertThat(
view.asInstanceOf[CatalogView].getExpandedQuery
).isEqualTo("SELECT CONCAT('a', 'bc', 'def')")
).isEqualTo("SELECT `CONCAT`('a', 'bc', 'def')")
}

@TestTemplate
Expand All @@ -172,7 +172,7 @@ class ViewsExpandingTest(tableTestUtil: TableTestBase => TableTestUtil) extends
.getTable(objectID.toObjectPath)
assertThat(
view.asInstanceOf[CatalogView].getExpandedQuery
).isEqualTo("SELECT `default_catalog`.`default_database`.`func`(1, CAST(2 AS BIGINT), 'abc')")
).isEqualTo("SELECT `default_catalog`.`default_database`.`func`(1, 2, 'abc')")
}

@TestTemplate
Expand Down Expand Up @@ -201,12 +201,12 @@ class ViewsExpandingTest(tableTestUtil: TableTestBase => TableTestUtil) extends
assertThat(
view.asInstanceOf[CatalogView].getExpandedQuery
).isEqualTo(
"SELECT `EXPR$0`.`f0`, `EXPR$0`.`rowNum`\n"
"SELECT *\n"
Copy link
Copy Markdown
Contributor

@raminqaf raminqaf Apr 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we consider adding a second test variant with explicit columns alongside this one?

+ "FROM (SELECT `source`.`f0`, "
+ "ROW_NUMBER() "
+ "OVER (PARTITION BY `source`.`f0` ORDER BY `source`.`f0` DESC) AS `rowNum`\n"
+ "FROM `default_catalog`.`default_database`.`source` AS `source`) AS `EXPR$0`\n"
+ "WHERE `EXPR$0`.`rowNum` = 1")
+ "FROM `default_catalog`.`default_database`.`source`)\n"
+ "WHERE `rowNum` = 1")
}

private def createSqlView(originTable: String): CatalogView = {
Expand Down