You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "RubyChou (via GitHub)" <gi...@apache.org> on 2024/03/26 11:42:53 UTC
[PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
RubyChou opened a new pull request, #24567:
URL: https://github.com/apache/flink/pull/24567
## What is the purpose of the change
*The PR aims to support pushing a Project past a Window in stream mode(It has been supported in batch mode).*
## Brief change log
- *Introduce `FlinkProjectWindowTransposeRule` into `project_rewrite` and `logical` stage .*
- *Update xml and .out of effected test case*
- *Move `CoreRules.PROJECT_WINDOW_TRANSPOSE` from `default_rewrite` stage into `project_rewrite` and `logical` stage. (To keep behaviors consistent in stream and batch mode. )*
## Verifying this change
This change added tests and can be verified as follows:
- *Added `FlinkProjectWindowTransposeRuleTest` to test if the plan is as expected.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2780304744
This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "RubyChou (via GitHub)" <gi...@apache.org>.
RubyChou commented on code in PR #24567:
URL: https://github.com/apache/flink/pull/24567#discussion_r1549314520
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala:
##########
@@ -656,18 +656,14 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
"(true,1,book,b,3,2)",
"(true,2,fruit,a,2,1)",
"(true,3,book,a,1,2)",
- "(true,3,book,a,1,2)",
Review Comment:
It's not expected, as the upsert key is wrong in the last commit, should be partition by key and row number instead of partition by key and order by key. The states are stored by upsert keys, therefore wrong upsert keys will cause wrong result.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "libenchao (via GitHub)" <gi...@apache.org>.
libenchao commented on code in PR #24567:
URL: https://github.com/apache/flink/pull/24567#discussion_r1542265389
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,27 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER rank, then the upsert keys are partition by key and order key.
Review Comment:
the upsert keys should be : partition key + order key + rownumber column?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,27 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
Review Comment:
And since you changed `FlinkRelMdUniqueKeys`, it would be good to also add tests for it in `FlinkRelMdUniqueKeysTest`
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala:
##########
@@ -656,18 +656,14 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
"(true,1,book,b,3,2)",
"(true,2,fruit,a,2,1)",
"(true,3,book,a,1,2)",
- "(true,3,book,a,1,2)",
Review Comment:
Is this change expected?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,27 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER rank, then the upsert keys are partition by key and order key.
+ if (rel.groups.length == 1) {
Review Comment:
`Window` can have multiple groups, if any one of them contains `row_number`, it could be an upsert key condidate
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,27 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
Review Comment:
It seems like a good improvement to both unique key and upsert key metadata, do you think it's possible to put it in `FlinkRelMdUniqueKeys` and reuse it in `FlinkRelMdUpsertKeys`, just like `FlinkRelMdUniqueKeys.getRankUniqueKeys`?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectWindowTransposeRule.java:
##########
@@ -0,0 +1,294 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.Window;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalWindow;
+import org.apache.calcite.rel.rules.ProjectRemoveRule;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.util.BitSets;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.immutables.value.Value;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Copied from {@link org.apache.calcite.rel.rules.ProjectWindowTransposeRule} The only difference
+ * is line 251 to 257, the upsert keys of window are kept not pushed down. Planner rule that pushes
Review Comment:
Could you add the reasoning why you want to do this change?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #24567: [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule.
URL: https://github.com/apache/flink/pull/24567
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "RubyChou (via GitHub)" <gi...@apache.org>.
RubyChou commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2041291550
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2853414867
This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "libenchao (via GitHub)" <gi...@apache.org>.
libenchao commented on code in PR #24567:
URL: https://github.com/apache/flink/pull/24567#discussion_r1554798121
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnUniquenessTest.scala:
##########
@@ -193,26 +193,26 @@ class FlinkRelMdColumnUniquenessTest extends FlinkRelMdHandlerTestBase {
@Test
def testAreColumnsUniqueOnRank(): Unit = {
- Array(
- logicalRank,
- flinkLogicalRank,
- batchLocalRank,
- batchGlobalRank,
- streamRank,
- logicalRankWithVariableRange,
- flinkLogicalRankWithVariableRange,
- streamRankWithVariableRange
- )
- .foreach {
- rank =>
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0)))
- (1 until rank.getRowType.getFieldCount).foreach {
- idx => assertFalse(mq.areColumnsUnique(rank, ImmutableBitSet.of(idx)))
- }
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0, 1)))
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0, 2)))
- assertFalse(mq.areColumnsUnique(rank, ImmutableBitSet.of(1, 2)))
- }
+// Array(
Review Comment:
Why is this commented out?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "RubyChou (via GitHub)" <gi...@apache.org>.
RubyChou commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2033820926
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "RubyChou (via GitHub)" <gi...@apache.org>.
RubyChou commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2034009333
@libenchao hi, comments are all resolved, please help reviewing it when you have time, thx : )
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "libenchao (via GitHub)" <gi...@apache.org>.
libenchao commented on code in PR #24567:
URL: https://github.com/apache/flink/pull/24567#discussion_r1554809730
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala:
##########
@@ -438,9 +438,36 @@ class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.Uniqu
rel: Window,
mq: RelMetadataQuery,
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
+ if (rel.groups.length == 1) {
+ val group = rel.groups.get(0)
+ getUniqueKeysOfWindowGroup(group, rel) match {
+ case Some(uniqueKeys) =>
+ val retSet = new JHashSet[ImmutableBitSet]
+ retSet.add(uniqueKeys)
+ val inputKeys = mq.getUniqueKeys(rel.getInput, ignoreNulls)
+ if (inputKeys != null && inputKeys.nonEmpty) {
+ inputKeys.foreach(uniqueKey => retSet.add(uniqueKey))
+ }
+ return retSet
+ case _ =>
+ }
+ }
getUniqueKeysOfOverAgg(rel, mq, ignoreNulls)
}
+ def getUniqueKeysOfWindowGroup(group: Window.Group, rel: Window): Option[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER window, then the unique keys are partition by key and row number.
+ val aggCalls = group.aggCalls
+ if (
+ aggCalls.length == 1 && aggCalls.get(0).getOperator.equals(SqlStdOperatorTable.ROW_NUMBER)
Review Comment:
`aggCalls` could be more than one, and any of the aggCalls is `ROW_NUMBER` would make it a unique key?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala:
##########
@@ -438,9 +438,36 @@ class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.Uniqu
rel: Window,
mq: RelMetadataQuery,
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
+ if (rel.groups.length == 1) {
Review Comment:
`groups` can be more than one, and any of the `groups` could have `ROW_NUMBER`, and all of them are valid unique key candidates?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,25 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER Rank, then the upsert keys are partition by key and row number.
+ if (rel.groups.length == 1) {
Review Comment:
Again, limiting the number of `groups` to 1 is weird. If in streaming, we can only support one `group` for now (I'm not 100 percent sure about it), you should at least add a comment about it for future maintainability.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeysTest.scala:
##########
@@ -172,6 +172,8 @@ class FlinkRelMdUniqueKeysTest extends FlinkRelMdHandlerTestBase {
@Test
def testGetUniqueKeysOnRank(): Unit = {
+ assertEquals(uniqueKeys(Array(7), Array(0)), mq.getUniqueKeys(logicalWindow).toSet)
Review Comment:
Isn't this being tested below?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala:
##########
@@ -328,6 +329,11 @@ object RankUtil {
}
}
+ def getRankNumberColumnIndex(window: Window): Int = {
Review Comment:
It would be good to check if the window function is `ROW_NUMBER` in case of misusage.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnUniquenessTest.scala:
##########
@@ -193,26 +193,26 @@ class FlinkRelMdColumnUniquenessTest extends FlinkRelMdHandlerTestBase {
@Test
def testAreColumnsUniqueOnRank(): Unit = {
- Array(
- logicalRank,
- flinkLogicalRank,
- batchLocalRank,
- batchGlobalRank,
- streamRank,
- logicalRankWithVariableRange,
- flinkLogicalRankWithVariableRange,
- streamRankWithVariableRange
- )
- .foreach {
- rank =>
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0)))
- (1 until rank.getRowType.getFieldCount).foreach {
- idx => assertFalse(mq.areColumnsUnique(rank, ImmutableBitSet.of(idx)))
- }
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0, 1)))
- assertTrue(mq.areColumnsUnique(rank, ImmutableBitSet.of(0, 2)))
- assertFalse(mq.areColumnsUnique(rank, ImmutableBitSet.of(1, 2)))
- }
+// Array(
Review Comment:
Why this is commented out?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,25 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER Rank, then the upsert keys are partition by key and row number.
+ if (rel.groups.length == 1) {
+ val group = rel.groups.get(0)
+ FlinkRelMdUniqueKeys.INSTANCE.getUniqueKeysOfWindowGroup(group, rel) match {
+ case Some(uniqueKeys) =>
+ val inputKeys = filterKeys(
+ FlinkRelMetadataQuery
+ .reuseOrCreate(mq)
+ .getUpsertKeys(rel.getInput),
+ group.keys)
+ val retSet = new JHashSet[ImmutableBitSet]
+ retSet.add(uniqueKeys)
+ if (inputKeys != null && inputKeys.nonEmpty) {
+ inputKeys.foreach(uniqueKey => retSet.add(uniqueKey))
Review Comment:
`uniqueKey` is better renamed as `upsertKey`
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala:
##########
@@ -186,6 +187,25 @@ class FlinkRelMdUpsertKeys private extends MetadataHandler[UpsertKeys] {
}
def getUpsertKeys(rel: Window, mq: RelMetadataQuery): JSet[ImmutableBitSet] = {
+ // If it's a ROW_NUMBER Rank, then the upsert keys are partition by key and row number.
+ if (rel.groups.length == 1) {
+ val group = rel.groups.get(0)
+ FlinkRelMdUniqueKeys.INSTANCE.getUniqueKeysOfWindowGroup(group, rel) match {
+ case Some(uniqueKeys) =>
+ val inputKeys = filterKeys(
+ FlinkRelMetadataQuery
+ .reuseOrCreate(mq)
+ .getUpsertKeys(rel.getInput),
+ group.keys)
Review Comment:
This is duplicated with `getUpsertKeysOnOver`. Since you always call `getUpsertKeysOnOver` to get upsert keys derived from child, you can put this at the beginning of the method, and use the result in various places.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34529][table-planner] Introduce FlinkProjectWindowTransposeRule. [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24567:
URL: https://github.com/apache/flink/pull/24567#issuecomment-2020210585
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "839b369305d886c7575fe9c179db9965a3534432",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "839b369305d886c7575fe9c179db9965a3534432",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 839b369305d886c7575fe9c179db9965a3534432 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org