snuyanzin commented on code in PR #26780:
URL: https://github.com/apache/flink/pull/26780#discussion_r2200036118
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java:
##########
@@ -37,43 +38,155 @@
class PushCalcPastChangelogNormalizeRuleTest extends TableTestBase {
private StreamTableTestUtil util;
+ private static TableDescriptor sourceDescriptor;
+ private static TableDescriptor sourceDescriptorWithTwoPrimaryKeys;
- @BeforeEach
- void before() {
- util = streamTestUtil(TableConfig.getDefault());
- }
-
- @Test
- void testWithSinglePrimaryKeyFilter() {
- final TableDescriptor sourceDescriptor =
+ @BeforeAll
+ static void setup() {
+ sourceDescriptor =
TableFactoryHarness.newBuilder()
.schema(
Schema.newBuilder()
- .column("f0", STRING())
+ .column("f0", INT())
.column("f1", INT().notNull())
+ .column("f2", STRING())
+ .column("f3", BIGINT().notNull())
.primaryKey("f1")
.build())
.unboundedScanSource(ChangelogMode.upsert())
.build();
- util.tableEnv().createTable("T", sourceDescriptor);
- util.verifyRelPlan("SELECT * FROM T WHERE f1 < 1");
- }
-
- @Test
- void testWithMultipleFilters() {
- final TableDescriptor sourceDescriptor =
+ sourceDescriptorWithTwoPrimaryKeys =
TableFactoryHarness.newBuilder()
.schema(
Schema.newBuilder()
.column("f0", STRING())
.column("f1", INT().notNull())
- .column("f2", STRING())
- .primaryKey("f1")
+ .column("f2", BIGINT().notNull())
+ .column("f3", STRING())
+ .column("f4", BIGINT().notNull())
+ .column("f5", BIGINT().notNull())
+ .column("f6", BIGINT().notNull())
+ .column("f7", BIGINT().notNull())
+ .primaryKey("f1", "f2")
.build())
.unboundedScanSource(ChangelogMode.upsert())
.build();
+ }
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+ }
+
+ @Test
+ void testWithSinglePrimaryKeyFilter() {
+ util.tableEnv().createTable("T", sourceDescriptor);
+ util.verifyRelPlan("SELECT * FROM T WHERE f1 < 1");
+ }
+
+ @Test
+ void testPrimaryKeySeveralSameSources() {
+ util.tableEnv().createTable("T", sourceDescriptor);
+ // Shouldn't be pushed down since there is no common filter for
ChangelogNormalize with the
+ // same source
+ // verifyExecPlan is intended here as it will show whether the node is
reused or not
+ util.verifyExecPlan(
Review Comment:
do you see `+- Reused(reference_id=[1])` in plans?
This indicates that reuse happens
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
