You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/08/27 07:17:47 UTC
[GitHub] [flink] Airblader opened a new pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Airblader opened a new pull request #17011:
URL: https://github.com/apache/flink/pull/17011
## What is the purpose of the change
`ChangelogNormalize` keeps state for each unique key, which is quite expensive. In general it isn't safe to push filters through this operator, but we _can_ allow filters which only filter on the key space to go past it. This allows for a significant reduction in the footprint of the operator if the key space is reduced.
This PR implements a new optimization rule which pushes such primary key filters past the `ChangelogNormalize` operator to allow its state to be reduced. The tests have first been run without the rule and the changes made by the rule are pushed as a second fixup commit for a more pleasant review experience. :-)
Note that sources which implement `SupportsFilterPushDown` currently always have all filters pushed past `ChangelogNormalize` into the source, see FLINK-24014.
This PR is currently in draft mode as we want to implement FLINK-24004 here to improve the test infrastructure.
## Verifying this change
This change added tests and can be verified as follows:
* `PushFilterPastChangelogNormalizeRuleTest`
## 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? JavaDocs
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
}, {
"hash" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"triggerType" : "PUSH"
}, {
"hash" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23113",
"triggerID" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"triggerType" : "PUSH"
}, {
"hash" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23117",
"triggerID" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 211af17558902443958144ba9f202b75c719982e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
* 73c5f356223138fc8c0b0cb74b54772501f8a02e UNKNOWN
* c34d1b07442c3ab208a1585cc888e36daf93b9ff Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23113)
* 361552cf00e91baf13ecd40ad968f7f95490feea Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23117)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] twalthr commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r698472194
##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
##########
@@ -51,14 +51,14 @@
* TableEnvironment#createTemporaryTable(String, TableDescriptor)}.
*/
@PublicEvolving
-public final class TableDescriptor {
+public class TableDescriptor {
private final @Nullable Schema schema;
private final Map<String, String> options;
private final List<String> partitionKeys;
private final @Nullable String comment;
- private TableDescriptor(
+ public TableDescriptor(
Review comment:
`protected`
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -108,39 +101,24 @@
/** Factory identifier for {@link Factory}. */
public static final String IDENTIFIER = "harness";
- public static final ConfigOption<String> SOURCE =
- ConfigOptions.key("source")
- .stringType()
- .noDefaultValue()
- .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
-
- public static final ConfigOption<String> SINK =
- ConfigOptions.key("sink")
- .stringType()
- .noDefaultValue()
- .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
-
// ---------------------------------------------------------------------------------------------
- /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
- public static TableDescriptor forSource(Schema schema, SourceBase source) {
- return TableDescriptor.forConnector(IDENTIFIER)
- .schema(schema)
- .option(SOURCE, source.serialize())
- .build();
- }
-
- /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
- public static TableDescriptor forSink(Schema schema, SinkBase sink) {
- return TableDescriptor.forConnector(IDENTIFIER)
- .schema(schema)
- .option(SINK, sink.serialize())
- .build();
+ /**
+ * Creates a builder for a new {@link TableDescriptor} specialized for this harness.
+ *
+ * <p>Use this method to create a {@link TableDescriptor} and passing in the source / sink
+ * implementation you want to use. The descriptor can for example be used with {@link
+ * TableEnvironment#createTable(String, TableDescriptor)} to register a table.
+ */
+ public static HarnessTableDescriptor.Builder forHarness() {
Review comment:
nit: `newBuilder()`? `TableFactoryHarness.forHarness` reads cumbersome
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
##########
@@ -33,14 +33,14 @@
/** Default implementation of a {@link CatalogTable}. */
@Internal
-class DefaultCatalogTable implements CatalogTable {
+public class DefaultCatalogTable implements CatalogTable {
private final Schema schema;
private final @Nullable String comment;
private final List<String> partitionKeys;
private final Map<String, String> options;
- DefaultCatalogTable(
+ public DefaultCatalogTable(
Review comment:
`protected`
--
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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r698432369
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
Review comment:
I'll settle for "… can lead to …" which adds a bit less noise in the sentence (a system that is only broken in some scenarios is broken overall, IMO).
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
Review comment:
Yeah, I mean I don't see any issues with that, anyway. `a = 1 && b = 42` is equivalent to `a = 1 && b = 42 && c = c`, right?
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
+ primaryKeyPredicates.add(predicate);
+ } else {
+ remainingPredicates.add(predicate);
+ }
+ }
+ }
+
+ /** Pushes {@param primaryKeyPredicates} into the {@link StreamPhysicalChangelogNormalize}. */
+ private StreamPhysicalChangelogNormalize pushFiltersThroughChangeNormalize(
+ RelOptRuleCall call, List<RexNode> primaryKeyPredicates) {
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ if (primaryKeyPredicates.isEmpty()) {
+ // There are no filters which can be pushed, so just return the existing node.
+ return changelogNormalize;
+ }
+
+ final StreamPhysicalCalc pushedFiltersCalc =
+ projectIdentityWithConditions(
+ call.builder(), changelogNormalize.getInput(), primaryKeyPredicates);
+
+ return (StreamPhysicalChangelogNormalize)
+ changelogNormalize.copy(
+ changelogNormalize.getTraitSet(),
+ Collections.singletonList(pushedFiltersCalc));
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} with the given {@param conditions} and an identity
+ * projection.
+ */
+ private StreamPhysicalCalc projectIdentityWithConditions(
+ RelBuilder relBuilder, RelNode newInput, List<RexNode> conditions) {
+
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(newInput.getRowType(), relBuilder.getRexBuilder());
+ programBuilder.addIdentity();
+
+ final RexNode condition = relBuilder.and(conditions);
+ if (!condition.isAlwaysTrue()) {
+ programBuilder.addCondition(condition);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
+ return new StreamPhysicalCalc(
+ newInput.getCluster(),
+ newInput.getTraitSet(),
+ newInput,
+ newProgram,
+ newProgram.getOutputRowType());
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} which is a copy of {@param calc}, but with the
+ * projections applied from {@param projectFromCalc}.
+ */
+ private StreamPhysicalCalc projectWith(
+ RelBuilder relBuilder, StreamPhysicalCalc projectFromCalc, StreamPhysicalCalc calc) {
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(calc.getRowType(), relBuilder.getRexBuilder());
+ if (calc.getProgram().getCondition() != null) {
+ programBuilder.addCondition(
+ calc.getProgram().expandLocalRef(calc.getProgram().getCondition()));
+ }
+
+ for (Pair<RexLocalRef, String> projectRef :
+ projectFromCalc.getProgram().getNamedProjects()) {
+ final RexNode project = projectFromCalc.getProgram().expandLocalRef(projectRef.left);
+ programBuilder.addProject(project, projectRef.right);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
Review comment:
Yeah, it is using local refs again.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
}, {
"hash" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"triggerType" : "PUSH"
}, {
"hash" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 211af17558902443958144ba9f202b75c719982e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
* 73c5f356223138fc8c0b0cb74b54772501f8a02e UNKNOWN
* c34d1b07442c3ab208a1585cc888e36daf93b9ff UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r697805528
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link SourceBase} or {@link SinkBase} which provides default implementations for most
+ * methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
Review comment:
Actually using SharedObjects here is entirely unnecessary. We instantiate the source specifically for this test, so we can just keep state on the source and assert on that. Will update and also the docs for the harness.
--
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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r697805528
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link SourceBase} or {@link SinkBase} which provides default implementations for most
+ * methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
Review comment:
Actually using SharedObjects here is entirely unnecessary. We instantiate the source specifically for this test, so we can just keep state on the source and assert on that. Will update and also the docs for the harness.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e13ee18ff9f927cb23c670236f83956873cfcc1e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r697805629
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link SourceBase} or {@link SinkBase} which provides default implementations for most
+ * methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends SourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SOURCE, source.serialize())
+ .build();
+ }
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
+ public static TableDescriptor forSink(Schema schema, SinkBase sink) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SINK, sink.serialize())
+ .build();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Harness factory for creating sources / sinks from base64-encoded serialized strings. */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.singleton(SOURCE);
Review comment:
Missing SINK
--
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
[GitHub] [flink] twalthr closed pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
twalthr closed pull request #17011:
URL: https://github.com/apache/flink/pull/17011
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e13ee18ff9f927cb23c670236f83956873cfcc1e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958)
* 211af17558902443958144ba9f202b75c719982e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
}, {
"hash" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"triggerType" : "PUSH"
}, {
"hash" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23113",
"triggerID" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"triggerType" : "PUSH"
}, {
"hash" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 211af17558902443958144ba9f202b75c719982e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
* 73c5f356223138fc8c0b0cb74b54772501f8a02e UNKNOWN
* c34d1b07442c3ab208a1585cc888e36daf93b9ff Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23113)
* 361552cf00e91baf13ecd40ad968f7f95490feea UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941)
* e13ee18ff9f927cb23c670236f83956873cfcc1e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot commented on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r697805773
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link SourceBase} or {@link SinkBase} which provides default implementations for most
+ * methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends SourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SOURCE, source.serialize())
+ .build();
+ }
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
+ public static TableDescriptor forSink(Schema schema, SinkBase sink) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SINK, sink.serialize())
+ .build();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Harness factory for creating sources / sinks from base64-encoded serialized strings. */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.singleton(SOURCE);
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSource source =
+ deserializeSourceSink(factoryHelper.getOptions(), SOURCE);
+ if (source instanceof SourceBase) {
+ ((SourceBase) source).factoryContext = context;
+ }
+
+ return source;
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSink sink = deserializeSourceSink(factoryHelper.getOptions(), SINK);
+ if (sink instanceof SinkBase) {
+ ((SinkBase) sink).factoryContext = context;
+ }
+
+ return sink;
+ }
+
+ private <T> T deserializeSourceSink(
+ ReadableConfig options, ConfigOption<String> configOption) {
+ final String serializedValue =
+ options.getOptional(configOption)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Missing option '%s'.",
+ configOption.key())));
+
+ try {
+ return InstantiationUtil.deserializeObject(
+ Base64.getDecoder().decode(serializedValue),
+ Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException | IOException e) {
+ throw new ValidationException(
+ "Serialized source/sink could not be deserialized.", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes a source / sink into a base64-encoded string which can be used by {@link Factory}.
+ *
+ * <p>If your source / sink extends from {@link SourceBase} / {@link SinkBase}, you can use
+ * {@link SourceBase#serialize()} / {@link SinkBase#serialize()} instead, or use {@link
+ * #forSource(Schema, SourceBase)} / {@link #forSink(Schema, SinkBase)}.
+ */
+ public static String serializeImplementation(Object obj) {
+ try {
+ return Base64.getEncoder().encodeToString(InstantiationUtil.serializeObject(obj));
+ } catch (IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
+
+ /** Creates a {@link ScanRuntimeProvider} which produces nothing. */
+ public static ScanRuntimeProvider createEmptyScanProvider() {
+ return SourceFunctionProvider.of(
+ new SourceFunction<RowData>() {
+ @Override
+ public void run(SourceContext<RowData> ctx) {}
+
+ @Override
+ public void cancel() {}
+ },
+ true);
+ }
+
+ /** Creates a {@link SinkRuntimeProvider} which discards all records. */
+ public static SinkRuntimeProvider createEmptySinkProvider() {
+ return SinkFunctionProvider.of(
+ new SinkFunction<RowData>() {
+ @Override
+ public void invoke(RowData value, Context context) {}
+ });
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Serializable base class for custom sources which implements {@link ScanTableSource}.
+ *
+ * <p>Most interface methods are default-implemented for convenience, but can be overridden when
+ * necessary. By default, a {@link ScanRuntimeProvider} is used which doesn't produce anything.
+ *
+ * <p>Sources derived from this base class will also be provided the {@link
+ * DynamicTableFactory.Context} of the factory which gives access to e.g. the {@link
+ * CatalogTable}.
+ */
+ public abstract static class SourceBase implements Serializable, ScanTableSource {
Review comment:
Maybe rename to ScanSourceBase and also have a LookupSourceBase?
--
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
[GitHub] [flink] flinkbot commented on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-906985329
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1 (Fri Aug 27 07:20:04 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941)
* e13ee18ff9f927cb23c670236f83956873cfcc1e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 211af17558902443958144ba9f202b75c719982e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] twalthr commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r698306556
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link ScanSourceBase}, {@link LookupSourceBase}, or {@link SinkBase} which provide default
+ * implementations for most methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
Review comment:
nit: use the new `Schema.derived()`
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link ScanSourceBase}, {@link LookupSourceBase}, or {@link SinkBase} which provide default
+ * implementations for most methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends ScanSourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SOURCE, source.serialize())
+ .build();
+ }
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
+ public static TableDescriptor forSink(Schema schema, SinkBase sink) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SINK, sink.serialize())
+ .build();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Harness factory for creating sources / sinks from base64-encoded serialized strings. */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(SOURCE);
+ options.add(SINK);
+ return options;
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSource source =
+ deserializeSourceSink(factoryHelper.getOptions(), SOURCE);
+ if (source instanceof SourceBase) {
+ ((SourceBase) source).factoryContext = context;
+ }
+
+ return source;
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSink sink = deserializeSourceSink(factoryHelper.getOptions(), SINK);
+ if (sink instanceof SinkBase) {
+ ((SinkBase) sink).factoryContext = context;
+ }
+
+ return sink;
+ }
+
+ private <T> T deserializeSourceSink(
+ ReadableConfig options, ConfigOption<String> configOption) {
+ final String serializedValue =
+ options.getOptional(configOption)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Missing option '%s'.",
+ configOption.key())));
+
+ try {
+ return InstantiationUtil.deserializeObject(
+ Base64.getDecoder().decode(serializedValue),
+ Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException | IOException e) {
+ throw new ValidationException(
+ "Serialized source/sink could not be deserialized.", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes a source / sink into a base64-encoded string which can be used by {@link Factory}.
+ *
+ * <p>If your source / sink extends from {@link ScanSourceBase}, {@link LookupSourceBase}, or
+ * {@link SinkBase}, you can use {@link SourceBase#serialize()} / {@link SinkBase#serialize()}
+ * instead, or use {@link #forSource(Schema, SourceBase)} / {@link #forSink(Schema, SinkBase)}.
+ */
+ public static String serializeImplementation(Object obj) {
Review comment:
the serialize/deserialize part doesn't look very nice. it could also slow down the test, wouldn't it be possible to somehow leverage the `SharedObjects` infrastructure and only pass around `sharedObjectsId`?
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link ScanSourceBase}, {@link LookupSourceBase}, or {@link SinkBase} which provide default
+ * implementations for most methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends ScanSourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
Review comment:
can we add a variant that does not require `SourceBase` instance? This could be useful for basic plan tests with no abilities implemented.
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
Review comment:
nit: `potentially incorrect results` it depends on the query
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
+ primaryKeyPredicates.add(predicate);
+ } else {
+ remainingPredicates.add(predicate);
+ }
+ }
+ }
+
+ /** Pushes {@param primaryKeyPredicates} into the {@link StreamPhysicalChangelogNormalize}. */
+ private StreamPhysicalChangelogNormalize pushFiltersThroughChangeNormalize(
Review comment:
nit: `pushFiltersThroughChangelogNormalize`
##########
File path: flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRuleTest.xml
##########
@@ -30,10 +30,11 @@ LogicalAggregate(group=[{0}], sum=[SUM($0)])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[f1, f1 AS sum], where=[AND(<(f1, 10), OR(>(f1, 3), IS NULL(f2)))])
+Calc(select=[f1, f1 AS sum], where=[OR(>(f1, 3), IS NULL(f2))])
+- ChangelogNormalize(key=[f1])
- +- Exchange(distribution=[hash[f1]])
- +- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1, f2])
+ +- Calc(select=[f0, f1, f2], where=[<(f1, 10)])
+ +- Exchange(distribution=[hash[f1]])
Review comment:
sorry, if I was not clear enough. but of course we can also push it through the exchange that will (always?) be in front of the `ChangelogNormalize`. Shuffling is also expensive due to network IO, the more filtering here, the better.
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link ScanSourceBase}, {@link LookupSourceBase}, or {@link SinkBase} which provide default
+ * implementations for most methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends ScanSourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SOURCE, source.serialize())
+ .build();
+ }
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
+ public static TableDescriptor forSink(Schema schema, SinkBase sink) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SINK, sink.serialize())
+ .build();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Harness factory for creating sources / sinks from base64-encoded serialized strings. */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(SOURCE);
+ options.add(SINK);
+ return options;
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSource source =
+ deserializeSourceSink(factoryHelper.getOptions(), SOURCE);
+ if (source instanceof SourceBase) {
+ ((SourceBase) source).factoryContext = context;
+ }
+
+ return source;
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSink sink = deserializeSourceSink(factoryHelper.getOptions(), SINK);
+ if (sink instanceof SinkBase) {
+ ((SinkBase) sink).factoryContext = context;
+ }
+
+ return sink;
+ }
+
+ private <T> T deserializeSourceSink(
+ ReadableConfig options, ConfigOption<String> configOption) {
+ final String serializedValue =
+ options.getOptional(configOption)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Missing option '%s'.",
+ configOption.key())));
+
+ try {
+ return InstantiationUtil.deserializeObject(
+ Base64.getDecoder().decode(serializedValue),
+ Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException | IOException e) {
+ throw new ValidationException(
+ "Serialized source/sink could not be deserialized.", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes a source / sink into a base64-encoded string which can be used by {@link Factory}.
+ *
+ * <p>If your source / sink extends from {@link ScanSourceBase}, {@link LookupSourceBase}, or
+ * {@link SinkBase}, you can use {@link SourceBase#serialize()} / {@link SinkBase#serialize()}
+ * instead, or use {@link #forSource(Schema, SourceBase)} / {@link #forSink(Schema, SinkBase)}.
+ */
+ public static String serializeImplementation(Object obj) {
+ try {
+ return Base64.getEncoder().encodeToString(InstantiationUtil.serializeObject(obj));
+ } catch (IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
+
+ /** Creates a {@link ScanRuntimeProvider} which produces nothing. */
+ public static ScanRuntimeProvider createEmptyScanProvider() {
+ return SourceFunctionProvider.of(
+ new SourceFunction<RowData>() {
+ @Override
+ public void run(SourceContext<RowData> ctx) {}
+
+ @Override
+ public void cancel() {}
+ },
+ true);
Review comment:
it would be useful to make this configurable, maybe even via the `TableFactoryHarness.forBoundedSource`, ``TableFactoryHarness.forUnboundedSource``
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
Review comment:
I'm just mentioning here if it was a conscious decision to support use cases such as:
primary key with `(a,b,c)` with a predicate `a=1 && b=42` would be pushed through?
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
+ primaryKeyPredicates.add(predicate);
+ } else {
+ remainingPredicates.add(predicate);
+ }
+ }
+ }
+
+ /** Pushes {@param primaryKeyPredicates} into the {@link StreamPhysicalChangelogNormalize}. */
+ private StreamPhysicalChangelogNormalize pushFiltersThroughChangeNormalize(
+ RelOptRuleCall call, List<RexNode> primaryKeyPredicates) {
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ if (primaryKeyPredicates.isEmpty()) {
+ // There are no filters which can be pushed, so just return the existing node.
+ return changelogNormalize;
+ }
+
+ final StreamPhysicalCalc pushedFiltersCalc =
+ projectIdentityWithConditions(
+ call.builder(), changelogNormalize.getInput(), primaryKeyPredicates);
+
+ return (StreamPhysicalChangelogNormalize)
+ changelogNormalize.copy(
+ changelogNormalize.getTraitSet(),
+ Collections.singletonList(pushedFiltersCalc));
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} with the given {@param conditions} and an identity
+ * projection.
+ */
+ private StreamPhysicalCalc projectIdentityWithConditions(
+ RelBuilder relBuilder, RelNode newInput, List<RexNode> conditions) {
+
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(newInput.getRowType(), relBuilder.getRexBuilder());
+ programBuilder.addIdentity();
+
+ final RexNode condition = relBuilder.and(conditions);
+ if (!condition.isAlwaysTrue()) {
+ programBuilder.addCondition(condition);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
+ return new StreamPhysicalCalc(
+ newInput.getCluster(),
+ newInput.getTraitSet(),
+ newInput,
+ newProgram,
+ newProgram.getOutputRowType());
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} which is a copy of {@param calc}, but with the
+ * projections applied from {@param projectFromCalc}.
+ */
+ private StreamPhysicalCalc projectWith(
+ RelBuilder relBuilder, StreamPhysicalCalc projectFromCalc, StreamPhysicalCalc calc) {
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(calc.getRowType(), relBuilder.getRexBuilder());
+ if (calc.getProgram().getCondition() != null) {
+ programBuilder.addCondition(
+ calc.getProgram().expandLocalRef(calc.getProgram().getCondition()));
+ }
+
+ for (Pair<RexLocalRef, String> projectRef :
+ projectFromCalc.getProgram().getNamedProjects()) {
+ final RexNode project = projectFromCalc.getProgram().expandLocalRef(projectRef.left);
+ programBuilder.addProject(project, projectRef.right);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
Review comment:
will the resulting program have local refs again? we are not using them yet but just for the future, we should restore a compact representation after applying this rule
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.DynamicTableSink.SinkRuntimeProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>Note that the underlying source/sink must be {@link Serializable}. It is recommended to extend
+ * from {@link ScanSourceBase}, {@link LookupSourceBase}, or {@link SinkBase} which provide default
+ * implementations for most methods as well as some convenience methods.
+ *
+ * <p>The harness provides a {@link Factory}. You can register a source / sink through configuration
+ * by passing a base64-encoded serialization. The harness provides convenience methods to make this
+ * process as simple as possible.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ *
+ * Schema schema = Schema.newBuilder().build();
+ * TableDescriptor sourceDescriptor = TableFactoryHarness.forSource(schema,
+ * new CustomSource(appliedLimits));
+ *
+ * tEnv.createTable("T", sourceDescriptor);
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends ScanSourceBase implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * {@literal @}Override
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ public static final ConfigOption<String> SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
+
+ public static final ConfigOption<String> SINK =
+ ConfigOptions.key("sink")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
+ public static TableDescriptor forSource(Schema schema, SourceBase source) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SOURCE, source.serialize())
+ .build();
+ }
+
+ /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
+ public static TableDescriptor forSink(Schema schema, SinkBase sink) {
+ return TableDescriptor.forConnector(IDENTIFIER)
+ .schema(schema)
+ .option(SINK, sink.serialize())
+ .build();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /** Harness factory for creating sources / sinks from base64-encoded serialized strings. */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(SOURCE);
+ options.add(SINK);
+ return options;
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSource source =
+ deserializeSourceSink(factoryHelper.getOptions(), SOURCE);
+ if (source instanceof SourceBase) {
+ ((SourceBase) source).factoryContext = context;
+ }
+
+ return source;
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final DynamicTableSink sink = deserializeSourceSink(factoryHelper.getOptions(), SINK);
+ if (sink instanceof SinkBase) {
+ ((SinkBase) sink).factoryContext = context;
+ }
+
+ return sink;
+ }
+
+ private <T> T deserializeSourceSink(
+ ReadableConfig options, ConfigOption<String> configOption) {
+ final String serializedValue =
+ options.getOptional(configOption)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Missing option '%s'.",
+ configOption.key())));
+
+ try {
+ return InstantiationUtil.deserializeObject(
+ Base64.getDecoder().decode(serializedValue),
+ Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException | IOException e) {
+ throw new ValidationException(
+ "Serialized source/sink could not be deserialized.", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes a source / sink into a base64-encoded string which can be used by {@link Factory}.
+ *
+ * <p>If your source / sink extends from {@link ScanSourceBase}, {@link LookupSourceBase}, or
+ * {@link SinkBase}, you can use {@link SourceBase#serialize()} / {@link SinkBase#serialize()}
+ * instead, or use {@link #forSource(Schema, SourceBase)} / {@link #forSink(Schema, SinkBase)}.
+ */
+ public static String serializeImplementation(Object obj) {
+ try {
+ return Base64.getEncoder().encodeToString(InstantiationUtil.serializeObject(obj));
+ } catch (IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
+
+ /** Creates a {@link ScanRuntimeProvider} which produces nothing. */
+ public static ScanRuntimeProvider createEmptyScanProvider() {
+ return SourceFunctionProvider.of(
+ new SourceFunction<RowData>() {
+ @Override
+ public void run(SourceContext<RowData> ctx) {}
+
+ @Override
+ public void cancel() {}
+ },
+ true);
Review comment:
for `forUnboundedSource ` we can also offer the changelog mode parameter already
--
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
[GitHub] [flink] twalthr closed pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
twalthr closed pull request #17011:
URL: https://github.com/apache/flink/pull/17011
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
--
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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r698432369
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
Review comment:
I'll settle for "… can lead to …" which adds a bit less noise in the sentence (a system that is only broken in some scenarios is broken overall, IMO).
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
Review comment:
Yeah, I mean I don't see any issues with that, anyway. `a = 1 && b = 42` is equivalent to `a = 1 && b = 42 && c = c`, right?
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
+ * to incorrect results.
+ */
+@Internal
+public class PushFilterPastChangelogNormalizeRule
+ extends RelRule<PushFilterPastChangelogNormalizeRule.Config> {
+
+ public static final RelOptRule INSTANCE =
+ Config.EMPTY.as(Config.class).onFilterWithChangelogNormalize().toRule();
+
+ public PushFilterPastChangelogNormalizeRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final StreamPhysicalCalc calc = call.rel(0);
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ final RexProgram program = calc.getProgram();
+ final RexNode condition =
+ RexUtil.toCnf(
+ call.builder().getRexBuilder(),
+ program.expandLocalRef(program.getCondition()));
+
+ final Set<Integer> primaryKeyIndices =
+ IntStream.of(changelogNormalize.uniqueKeys()).boxed().collect(Collectors.toSet());
+
+ // Determine which filters can be pushed (= involve only primary key columns)
+ final List<RexNode> primaryKeyPredicates = new ArrayList<>();
+ final List<RexNode> otherPredicates = new ArrayList<>();
+ partitionPrimaryKeyPredicates(
+ RelOptUtil.conjunctions(condition),
+ primaryKeyIndices,
+ primaryKeyPredicates,
+ otherPredicates);
+
+ // Construct a new ChangelogNormalize which has primary key filters pushed into it
+ final StreamPhysicalChangelogNormalize newChangelogNormalize =
+ pushFiltersThroughChangeNormalize(call, primaryKeyPredicates);
+
+ // Retain only filters which haven't been pushed
+ transformWithRemainingPredicates(call, newChangelogNormalize, otherPredicates);
+ }
+
+ /**
+ * Separates the given {@param predicates} into filters which affect only the primary key and
+ * anything else.
+ */
+ private void partitionPrimaryKeyPredicates(
+ List<RexNode> predicates,
+ Set<Integer> primaryKeyIndices,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> remainingPredicates) {
+ for (RexNode predicate : predicates) {
+ int[] inputRefs = extractRefInputFields(Collections.singletonList(predicate));
+ if (Arrays.stream(inputRefs).allMatch(primaryKeyIndices::contains)) {
+ primaryKeyPredicates.add(predicate);
+ } else {
+ remainingPredicates.add(predicate);
+ }
+ }
+ }
+
+ /** Pushes {@param primaryKeyPredicates} into the {@link StreamPhysicalChangelogNormalize}. */
+ private StreamPhysicalChangelogNormalize pushFiltersThroughChangeNormalize(
+ RelOptRuleCall call, List<RexNode> primaryKeyPredicates) {
+ final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
+
+ if (primaryKeyPredicates.isEmpty()) {
+ // There are no filters which can be pushed, so just return the existing node.
+ return changelogNormalize;
+ }
+
+ final StreamPhysicalCalc pushedFiltersCalc =
+ projectIdentityWithConditions(
+ call.builder(), changelogNormalize.getInput(), primaryKeyPredicates);
+
+ return (StreamPhysicalChangelogNormalize)
+ changelogNormalize.copy(
+ changelogNormalize.getTraitSet(),
+ Collections.singletonList(pushedFiltersCalc));
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} with the given {@param conditions} and an identity
+ * projection.
+ */
+ private StreamPhysicalCalc projectIdentityWithConditions(
+ RelBuilder relBuilder, RelNode newInput, List<RexNode> conditions) {
+
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(newInput.getRowType(), relBuilder.getRexBuilder());
+ programBuilder.addIdentity();
+
+ final RexNode condition = relBuilder.and(conditions);
+ if (!condition.isAlwaysTrue()) {
+ programBuilder.addCondition(condition);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
+ return new StreamPhysicalCalc(
+ newInput.getCluster(),
+ newInput.getTraitSet(),
+ newInput,
+ newProgram,
+ newProgram.getOutputRowType());
+ }
+
+ /**
+ * Returns a {@link StreamPhysicalCalc} which is a copy of {@param calc}, but with the
+ * projections applied from {@param projectFromCalc}.
+ */
+ private StreamPhysicalCalc projectWith(
+ RelBuilder relBuilder, StreamPhysicalCalc projectFromCalc, StreamPhysicalCalc calc) {
+ final RexProgramBuilder programBuilder =
+ new RexProgramBuilder(calc.getRowType(), relBuilder.getRexBuilder());
+ if (calc.getProgram().getCondition() != null) {
+ programBuilder.addCondition(
+ calc.getProgram().expandLocalRef(calc.getProgram().getCondition()));
+ }
+
+ for (Pair<RexLocalRef, String> projectRef :
+ projectFromCalc.getProgram().getNamedProjects()) {
+ final RexNode project = projectFromCalc.getProgram().expandLocalRef(projectRef.left);
+ programBuilder.addProject(project, projectRef.right);
+ }
+
+ final RexProgram newProgram = programBuilder.getProgram();
Review comment:
Yeah, it is using local refs again.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e13ee18ff9f927cb23c670236f83956873cfcc1e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958)
* 211af17558902443958144ba9f202b75c719982e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
}, {
"hash" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 211af17558902443958144ba9f202b75c719982e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075)
* 73c5f356223138fc8c0b0cb74b54772501f8a02e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] twalthr commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r698472194
##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
##########
@@ -51,14 +51,14 @@
* TableEnvironment#createTemporaryTable(String, TableDescriptor)}.
*/
@PublicEvolving
-public final class TableDescriptor {
+public class TableDescriptor {
private final @Nullable Schema schema;
private final Map<String, String> options;
private final List<String> partitionKeys;
private final @Nullable String comment;
- private TableDescriptor(
+ public TableDescriptor(
Review comment:
`protected`
##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
##########
@@ -108,39 +101,24 @@
/** Factory identifier for {@link Factory}. */
public static final String IDENTIFIER = "harness";
- public static final ConfigOption<String> SOURCE =
- ConfigOptions.key("source")
- .stringType()
- .noDefaultValue()
- .withDescription("Serialized instance of DynamicTableSource (Base64-encoded)");
-
- public static final ConfigOption<String> SINK =
- ConfigOptions.key("sink")
- .stringType()
- .noDefaultValue()
- .withDescription("Serialized instance of DynamicTableSink (Base64-encoded)");
-
// ---------------------------------------------------------------------------------------------
- /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param source}. */
- public static TableDescriptor forSource(Schema schema, SourceBase source) {
- return TableDescriptor.forConnector(IDENTIFIER)
- .schema(schema)
- .option(SOURCE, source.serialize())
- .build();
- }
-
- /** Creates a {@link TableDescriptor} for the given {@param schema} and {@param sink}. */
- public static TableDescriptor forSink(Schema schema, SinkBase sink) {
- return TableDescriptor.forConnector(IDENTIFIER)
- .schema(schema)
- .option(SINK, sink.serialize())
- .build();
+ /**
+ * Creates a builder for a new {@link TableDescriptor} specialized for this harness.
+ *
+ * <p>Use this method to create a {@link TableDescriptor} and passing in the source / sink
+ * implementation you want to use. The descriptor can for example be used with {@link
+ * TableEnvironment#createTable(String, TableDescriptor)} to register a table.
+ */
+ public static HarnessTableDescriptor.Builder forHarness() {
Review comment:
nit: `newBuilder()`? `TableFactoryHarness.forHarness` reads cumbersome
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
##########
@@ -33,14 +33,14 @@
/** Default implementation of a {@link CatalogTable}. */
@Internal
-class DefaultCatalogTable implements CatalogTable {
+public class DefaultCatalogTable implements CatalogTable {
private final Schema schema;
private final @Nullable String comment;
private final List<String> partitionKeys;
private final Map<String, String> options;
- DefaultCatalogTable(
+ public DefaultCatalogTable(
Review comment:
`protected`
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17011:
URL: https://github.com/apache/flink/pull/17011#issuecomment-907015222
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22941",
"triggerID" : "7d4c86b39ba4eee643fdbd6dcb55dc5ad03779b1",
"triggerType" : "PUSH"
}, {
"hash" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22958",
"triggerID" : "e13ee18ff9f927cb23c670236f83956873cfcc1e",
"triggerType" : "PUSH"
}, {
"hash" : "211af17558902443958144ba9f202b75c719982e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23075",
"triggerID" : "211af17558902443958144ba9f202b75c719982e",
"triggerType" : "PUSH"
}, {
"hash" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "73c5f356223138fc8c0b0cb74b54772501f8a02e",
"triggerType" : "PUSH"
}, {
"hash" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23113",
"triggerID" : "c34d1b07442c3ab208a1585cc888e36daf93b9ff",
"triggerType" : "PUSH"
}, {
"hash" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23117",
"triggerID" : "361552cf00e91baf13ecd40ad968f7f95490feea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 73c5f356223138fc8c0b0cb74b54772501f8a02e UNKNOWN
* 361552cf00e91baf13ecd40ad968f7f95490feea Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23117)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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
[GitHub] [flink] Airblader commented on a change in pull request #17011: [FLINK-23663][table-planner] Reduce state size of ChangelogNormalize
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #17011:
URL: https://github.com/apache/flink/pull/17011#discussion_r697805051
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushFilterPastChangelogNormalizeRule.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extractRefInputFields;
+
+/**
+ * Pushes primary key filters through a {@link StreamPhysicalChangelogNormalize ChangelogNormalize}
+ * operator to reduce its state size.
+ *
+ * <p>This rule looks for Calc → ChangelogNormalize where the {@link StreamPhysicalCalc Calc}
+ * contains a filter condition. The condition is transformed into CNF and then each conjunction is
+ * tested for whether it affects only primary key columns. If such conditions exist, they are moved
+ * into a new, separate Calc and pushed through the ChangelogNormalize operator. ChangelogNormalize
+ * keeps state for every unique key it encounters, thus pushing filters on the primary key in front
+ * of it helps reduce the size of its state.
+ *
+ * <p>Note that pushing primary key filters is safe to do, but pushing any other filters would lead
Review comment:
Wonky sentence here.
--
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