You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2020/06/14 16:50:20 UTC
[GitHub] [samza] atoomula opened a new pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
atoomula opened a new pull request #1384:
URL: https://github.com/apache/samza/pull/1384
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula merged pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula merged pull request #1384:
URL: https://github.com/apache/samza/pull/1384
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456933183
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +160,48 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ if (planner != null) {
+ planner.close();
+ }
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(Planner planner, RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(planner.transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ planner.close();
Review comment:
oh yeah.. good point!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456932347
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +166,45 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
Review comment:
1. \n is not for EOL. It is for separating the log.
2. You mean all log messages in Samza or just this ? And what is the point of capitalizing ? Is it to catch the eye while going thru logs ? If yes, we don't need to as their format makes them stand out in the logs,
2020-07-19 10:11:16.814 [main] [] QueryPlanner [INFO] query plan without optimization:
LogicalProject(__key__=[$9], pageKey=[$9], companyName=['N/A'], profileName=[$2], profileAddress=[$4])
LogicalFilter(condition=[AND(=($2, 'Mike'), =($10, 1))])
LogicalProject(__key__=[$0], id=[$1], name=[$2], companyId=[$3], address=[$4], selfEmployed=[$5], phoneNumbers=[$6], mapValues=[$7], __key__0=[$8], pageKey=[$9], profileId=[$10])
LogicalJoin(condition=[=($0, $11)], joinType=[inner])
LogicalTableScan(table=[[testRemoteStore, Profile, $table]])
LogicalProject(__key__=[$0], pageKey=[$1], profileId=[$2], $f3=[BuildOutputRecord('id', $2)])
LogicalTableScan(table=[[testavro, PAGEVIEW]])
2020-07-19 10:11:16.816 [main] [] QueryPlanner [INFO] query plan with optimization:
LogicalProject(__key__=[$9], pageKey=[$9], companyName=['N/A'], profileName=[$2], profileAddress=[$4])
LogicalFilter(condition=[=($2, 'Mike')])
LogicalJoin(condition=[=($0, $11)], joinType=[inner])
LogicalTableScan(table=[[testRemoteStore, Profile, $table]])
LogicalFilter(condition=[=($2, 1)])
LogicalProject(__key__=[$0], pageKey=[$1], profileId=[$2], $f3=[BuildOutputRecord('id', $2)])
LogicalTableScan(table=[[testavro, PAGEVIEW]])
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r444317206
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -109,15 +124,19 @@ private void registerSourceSchemas(SchemaPlus rootSchema) {
}
}
- public RelRoot plan(String query) {
+ private Planner getPlanner() {
Review comment:
The planner is a closable resource I think it would be better to use it within a try block or make sure to close it when done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r444319271
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try {
+ Planner planner = getPlanner();
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info("query plan without optimization:\n"
+ + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
If the main concern is that planning will fail with the new rules my suggestion is to have it on by default and catch the exception and re-plan without optimization. In this way we can learn the logs. It is up to you if you think this can be too much work
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r440604386
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a FilterJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
Review comment:
I plan to add JoinConditionPushRule as well in next iteration (with no filter) which will reuse most of the code here and onMatch will be different.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
Review comment:
This is vestige from my earlier attempt to enable volcano planner. I removed it now.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try {
+ Planner planner = getPlanner();
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info("query plan without optimization:\n"
+ + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
I'm considering this as experimental right now considering this is the first version. Once we vet it thru' the real use-cases and gain confidence, we can enable it by default.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r440325678
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try {
+ Planner planner = getPlanner();
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info("query plan without optimization:\n"
+ + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
When would you turn off the optimization ? if it is just optimization it should be always turned on ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] shanthoosh commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
shanthoosh commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r441004901
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/dsl/SamzaSqlDslConverter.java
##########
@@ -52,9 +52,9 @@
public Collection<RelRoot> convertDsl(String dsl) {
// TODO: Introduce an API to parse a dsl string and return one or more sql statements
List<String> sqlStmts = fetchSqlFromConfig(config);
- QueryPlanner planner = getQueryPlanner(getSqlConfig(sqlStmts, config));
List<RelRoot> relRoots = new LinkedList<>();
for (String sql: sqlStmts) {
+ QueryPlanner planner = getQueryPlanner(getSqlConfig(Collections.singletonList(sql), config));
Review comment:
What is the rationale for recreating the planner for every sql-statement in the app?
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +166,45 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
Review comment:
Nit:
1. By default, Logger adds `\n` to EOL. Unnecessary to explicitly add it.
2. Also, it's better to capitalize all the log-messages.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -129,6 +142,13 @@ public RelRoot plan(String query) {
sqlOperatorTables.add(new SamzaSqlOperatorTable());
sqlOperatorTables.add(new SamzaSqlUdfOperatorTable(samzaSqlFunctions));
+ // TODO: Introduce a pluggable rule factory.
Review comment:
Would be better to create a follow-up ticket for this action-item.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a SamzaSqlFilterRemoteJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
Review comment:
nit: %s/optimnization/optimization
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a SamzaSqlFilterRemoteJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
+ if (inputTypeOnLeft == InputType.LOCAL_TABLE || inputTypeOnRight == InputType.LOCAL_TABLE) {
+ donotOptimizeLeft = true;
+ donotOptimizeRight = true;
+ }
+
+ // There is nothing to optimize on the remote table side as the lookup needs to happen first before filtering.
+ if (inputTypeOnLeft == InputType.REMOTE_TABLE) {
+ donotOptimizeLeft = true;
+ }
+ if (inputTypeOnRight == InputType.REMOTE_TABLE) {
+ donotOptimizeRight = true;
+ }
+
+ // If there is only the joinRel,
+ // make sure it does not match a cartesian product joinRel
+ // (with "true" condition), otherwise this rule will be applied
+ // again on the new cartesian product joinRel.
+ if (filter == null && joinFilters.isEmpty()) {
+ return;
+ }
+
+ final List<RexNode> aboveFilters =
+ filter != null
+ ? RelOptUtil.conjunctions(filter.getCondition())
+ : new ArrayList<>();
+ final com.google.common.collect.ImmutableList<RexNode> origAboveFilters =
+ com.google.common.collect.ImmutableList.copyOf(aboveFilters);
+
+ // Simplify Outer Joins
+ JoinRelType joinType = join.getJoinType();
+ if (smart
+ && !origAboveFilters.isEmpty()
+ && join.getJoinType() != JoinRelType.INNER) {
+ joinType = RelOptUtil.simplifyJoin(join, origAboveFilters, joinType);
+ }
+
+ final List<RexNode> leftFilters = new ArrayList<>();
+ final List<RexNode> rightFilters = new ArrayList<>();
+
+ // TODO - add logic to derive additional filters. E.g., from
+ // (t1.a = 1 AND t2.a = 2) OR (t1.b = 3 AND t2.b = 4), you can
+ // derive table filters:
+ // (t1.a = 1 OR t1.b = 3)
+ // (t2.a = 2 OR t2.b = 4)
+
+ // Try to push down above filters. These are typically where clause
+ // filters. They can be pushed down if they are not on the NULL
+ // generating side.
+ // We do not push into join condition as we do not benefit much. There is also correctness issue
+ // with remote table as we will not have values for the remote table before the join/lookup.
+ boolean filterPushed = false;
+ if (RelOptUtil.classifyFilters(
+ join,
+ aboveFilters,
+ joinType,
+ false, // Let's not push into join filter
+ !joinType.generatesNullsOnLeft() && !donotOptimizeLeft,
+ !joinType.generatesNullsOnRight() && !donotOptimizeRight,
+ joinFilters,
+ leftFilters,
+ rightFilters)) {
+ filterPushed = true;
+ }
+
+ // If no filter got pushed after validate, reset filterPushed flag
+ if (leftFilters.isEmpty()
+ && rightFilters.isEmpty()) {
+ filterPushed = false;
+ }
+
+ boolean isAntiJoin = joinType == JoinRelType.ANTI;
+
+ // Try to push down filters in ON clause. A ON clause filter can only be
+ // pushed down if it does not affect the non-matching set, i.e. it is
+ // not on the side which is preserved.
+ // A ON clause filter of anti-join can not be pushed down.
+ if (!isAntiJoin && RelOptUtil.classifyFilters(
+ join,
+ joinFilters,
Review comment:
Just curious. Can you please clarify if we should use `aboveFilters` here for third argument? From the semantics of classifyFilters method, it seems like appropriate choice. Might be better to rename the variable to something better(other than aboveFilters).
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -129,6 +142,13 @@ public RelRoot plan(String query) {
sqlOperatorTables.add(new SamzaSqlOperatorTable());
sqlOperatorTables.add(new SamzaSqlUdfOperatorTable(samzaSqlFunctions));
+ // TODO: Introduce a pluggable rule factory.
+ List<RelOptRule> rules = ImmutableList.of(
Review comment:
Is there a way to determine if these rel-rules are applied on a rel-plan and emit a metric(or log it before/after the optimization) for debugging purposes.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +160,48 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ if (planner != null) {
+ planner.close();
+ }
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(Planner planner, RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(planner.transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ planner.close();
Review comment:
Calling close() here seems unnecessary, It's already closed at the caller already with try-with-resources-closeable.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +160,48 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ if (planner != null) {
+ planner.close();
+ }
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(Planner planner, RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(planner.transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ planner.close();
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try (Planner planner = getPlanner()) {
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info(
+ "query plan without optimization:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
Please capitalize log messages and remove \n at the end(which gets added by default).
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java
##########
@@ -165,6 +165,7 @@ private static Planner createPlanner() {
.traitDefs(traitDefs)
.context(Contexts.EMPTY_CONTEXT)
.costFactory(null)
+ //.programs(Programs.CALC_PROGRAM)
Review comment:
Just curious. Why did we comment this out? If this is not necessary, then can we please remove it?
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a SamzaSqlFilterRemoteJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
+ if (inputTypeOnLeft == InputType.LOCAL_TABLE || inputTypeOnRight == InputType.LOCAL_TABLE) {
+ donotOptimizeLeft = true;
+ donotOptimizeRight = true;
+ }
+
+ // There is nothing to optimize on the remote table side as the lookup needs to happen first before filtering.
+ if (inputTypeOnLeft == InputType.REMOTE_TABLE) {
+ donotOptimizeLeft = true;
+ }
+ if (inputTypeOnRight == InputType.REMOTE_TABLE) {
+ donotOptimizeRight = true;
+ }
+
+ // If there is only the joinRel,
+ // make sure it does not match a cartesian product joinRel
+ // (with "true" condition), otherwise this rule will be applied
+ // again on the new cartesian product joinRel.
+ if (filter == null && joinFilters.isEmpty()) {
+ return;
+ }
+
+ final List<RexNode> aboveFilters =
+ filter != null
+ ? RelOptUtil.conjunctions(filter.getCondition())
+ : new ArrayList<>();
+ final com.google.common.collect.ImmutableList<RexNode> origAboveFilters =
+ com.google.common.collect.ImmutableList.copyOf(aboveFilters);
+
+ // Simplify Outer Joins
+ JoinRelType joinType = join.getJoinType();
+ if (smart
+ && !origAboveFilters.isEmpty()
+ && join.getJoinType() != JoinRelType.INNER) {
+ joinType = RelOptUtil.simplifyJoin(join, origAboveFilters, joinType);
+ }
+
+ final List<RexNode> leftFilters = new ArrayList<>();
+ final List<RexNode> rightFilters = new ArrayList<>();
+
+ // TODO - add logic to derive additional filters. E.g., from
+ // (t1.a = 1 AND t2.a = 2) OR (t1.b = 3 AND t2.b = 4), you can
+ // derive table filters:
+ // (t1.a = 1 OR t1.b = 3)
+ // (t2.a = 2 OR t2.b = 4)
+
+ // Try to push down above filters. These are typically where clause
+ // filters. They can be pushed down if they are not on the NULL
+ // generating side.
+ // We do not push into join condition as we do not benefit much. There is also correctness issue
+ // with remote table as we will not have values for the remote table before the join/lookup.
+ boolean filterPushed = false;
+ if (RelOptUtil.classifyFilters(
+ join,
+ aboveFilters,
+ joinType,
+ false, // Let's not push into join filter
+ !joinType.generatesNullsOnLeft() && !donotOptimizeLeft,
+ !joinType.generatesNullsOnRight() && !donotOptimizeRight,
+ joinFilters,
+ leftFilters,
+ rightFilters)) {
+ filterPushed = true;
+ }
+
+ // If no filter got pushed after validate, reset filterPushed flag
+ if (leftFilters.isEmpty()
+ && rightFilters.isEmpty()) {
+ filterPushed = false;
+ }
+
+ boolean isAntiJoin = joinType == JoinRelType.ANTI;
+
+ // Try to push down filters in ON clause. A ON clause filter can only be
+ // pushed down if it does not affect the non-matching set, i.e. it is
+ // not on the side which is preserved.
+ // A ON clause filter of anti-join can not be pushed down.
+ if (!isAntiJoin && RelOptUtil.classifyFilters(
+ join,
+ joinFilters,
+ joinType,
+ false,
+ !joinType.generatesNullsOnLeft() && !donotOptimizeLeft,
+ !joinType.generatesNullsOnRight() && !donotOptimizeRight,
+ joinFilters,
+ leftFilters,
Review comment:
`Leftfilters` and `rightFilters` are initialized and are not modified. It's very hard to find where they're populated. Please add a comment here that leftFilters and rightFilters will be populated by this classifyFilters method.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a SamzaSqlFilterRemoteJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
+ if (inputTypeOnLeft == InputType.LOCAL_TABLE || inputTypeOnRight == InputType.LOCAL_TABLE) {
+ donotOptimizeLeft = true;
+ donotOptimizeRight = true;
+ }
+
+ // There is nothing to optimize on the remote table side as the lookup needs to happen first before filtering.
+ if (inputTypeOnLeft == InputType.REMOTE_TABLE) {
+ donotOptimizeLeft = true;
+ }
+ if (inputTypeOnRight == InputType.REMOTE_TABLE) {
+ donotOptimizeRight = true;
+ }
+
+ // If there is only the joinRel,
+ // make sure it does not match a cartesian product joinRel
+ // (with "true" condition), otherwise this rule will be applied
+ // again on the new cartesian product joinRel.
+ if (filter == null && joinFilters.isEmpty()) {
+ return;
+ }
+
+ final List<RexNode> aboveFilters =
+ filter != null
+ ? RelOptUtil.conjunctions(filter.getCondition())
+ : new ArrayList<>();
+ final com.google.common.collect.ImmutableList<RexNode> origAboveFilters =
Review comment:
Please import the ImmutableList class and don't hardcode the package paths. There're multiple occurrences in this file and else-where.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
Review comment:
Just curious. There seems to be considerable duplication with FilterJoinRule calcite native-class. Post [CALCITE-3170](https://issues.apache.org/jira/browse/CALCITE-3170), calcite supports anti-join on conditions push-down natively. If we upgrade to 1.21.0 rel-planner, then wouldn't overriding match suffice 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r444321120
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a FilterJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
Review comment:
In my opinion if we can minimize the copy and past that will be ideal and have the rest in a follow up, but it is up to you this is not a blocking point.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456932978
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +160,48 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ if (planner != null) {
+ planner.close();
+ }
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(Planner planner, RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(planner.transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ planner.close();
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try (Planner planner = getPlanner()) {
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info(
+ "query plan without optimization:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
Responded in earlier comment.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r444321891
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a FilterJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
+ if (inputTypeOnLeft == InputType.LOCAL_TABLE || inputTypeOnRight == InputType.LOCAL_TABLE) {
Review comment:
Why are we turning off for local tables things work fine, not sure I am getting this ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456933380
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -129,6 +142,13 @@ public RelRoot plan(String query) {
sqlOperatorTables.add(new SamzaSqlOperatorTable());
sqlOperatorTables.add(new SamzaSqlUdfOperatorTable(samzaSqlFunctions));
+ // TODO: Introduce a pluggable rule factory.
+ List<RelOptRule> rules = ImmutableList.of(
Review comment:
AFAIK, only Hex/VolcanoPlanners make such decisions. Not sure if we can determine that in the rule itself.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r440327510
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a FilterJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
Review comment:
As far I can tell it is only testing if filter is on the good side of the join, seems to me most of the work can be done at the onMatch, we can just extend and override onMatch ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r446711851
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -109,15 +124,19 @@ private void registerSourceSchemas(SchemaPlus rootSchema) {
}
}
- public RelRoot plan(String query) {
+ private Planner getPlanner() {
Review comment:
Sure. Added code to close it now.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
+ /** Whether to try to strengthen join-type. */
+ private final boolean smart;
+
+ Map<String, SqlIOConfig> systemStreamConfigBySource;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates a FilterJoinRule with an explicit root operand and
+ * factories.
+ */
+ protected SamzaSqlFilterRemoteJoinRule(RelOptRuleOperand operand, String id,
+ boolean smart, RelBuilderFactory relBuilderFactory, Map<String, SqlIOConfig> systemStreamConfigBySource) {
+ super(operand, relBuilderFactory, "SamzaSqlFilterRemoteJoinRule:" + id);
+ this.smart = smart;
+ this.systemStreamConfigBySource = systemStreamConfigBySource;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ protected void perform(RelOptRuleCall call, Filter filter,
+ Join join) {
+ final List<RexNode> joinFilters =
+ RelOptUtil.conjunctions(join.getCondition());
+
+ boolean donotOptimizeLeft = false;
+ boolean donotOptimizeRight = false;
+
+ JoinInputNode.InputType inputTypeOnLeft =
+ JoinInputNode.getInputType(join.getLeft(), systemStreamConfigBySource);
+ JoinInputNode.InputType inputTypeOnRight =
+ JoinInputNode.getInputType(join.getRight(), systemStreamConfigBySource);
+
+ // Disable this optimnization for queries using local table.
+ if (inputTypeOnLeft == InputType.LOCAL_TABLE || inputTypeOnRight == InputType.LOCAL_TABLE) {
Review comment:
There could be more optimizations done for local table as local tables do not have the limitations that remote tables have. We could directly enable Calcite's FilterJoinRule for local tables.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
mynameborat commented on pull request #1384:
URL: https://github.com/apache/samza/pull/1384#issuecomment-644229103
Can you please update the PR description according to - https://cwiki.apache.org/confluence/display/SAMZA/SEP-25%3A+PR+Title+And+Description+Guidelines
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] b-slim commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
b-slim commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r440323413
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
Review comment:
This is basically tricking the planner to use Enum Convention plus Volcano planner to generate a Plan with Convention of `EnumerableConvention.INSTANCE`. This will lead to complex branches down the line when doing the conversion as a follow up stage. You don't think the best way to do this is by using Calcite Conventions where Rules and Translation is happening at the same time to avoid complex code and very cryptic comments on what to expect when doing the translation ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456932653
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/dsl/SamzaSqlDslConverter.java
##########
@@ -52,9 +52,9 @@
public Collection<RelRoot> convertDsl(String dsl) {
// TODO: Introduce an API to parse a dsl string and return one or more sql statements
List<String> sqlStmts = fetchSqlFromConfig(config);
- QueryPlanner planner = getQueryPlanner(getSqlConfig(sqlStmts, config));
List<RelRoot> relRoots = new LinkedList<>();
for (String sql: sqlStmts) {
+ QueryPlanner planner = getQueryPlanner(getSqlConfig(Collections.singletonList(sql), config));
Review comment:
Good question. Calcite Planner as it stands today does not seem to be supported for reuse. Although the intent is there as they have exposed reset API. But it does not work. But it is such low cost to create new planner for each sql.
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -129,6 +142,13 @@ public RelRoot plan(String query) {
sqlOperatorTables.add(new SamzaSqlOperatorTable());
sqlOperatorTables.add(new SamzaSqlUdfOperatorTable(samzaSqlFunctions));
+ // TODO: Introduce a pluggable rule factory.
Review comment:
Sure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r456933648
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlFilterRemoteJoinRule.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.samza.sql.planner;
+
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.tools.RelBuilderFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.samza.sql.interfaces.SqlIOConfig;
+import org.apache.samza.sql.translator.JoinInputNode;
+import org.apache.samza.sql.translator.JoinInputNode.InputType;
+
+/**
+ * Planner rule for remote table joins that pushes filters above and
+ * within a join node into its children nodes.
+ * This class is customized form of Calcite's {@link org.apache.calcite.rel.rules.FilterJoinRule} for
+ * remote table joins.
+ */
+public abstract class SamzaSqlFilterRemoteJoinRule extends RelOptRule {
Review comment:
Are you talking about just anti-joins or in general about this rule ? If latter, Slim has a comment on this as well and I responded to him. I will have add another rule for join condition which reuses the same logic. Let me see at that time if I can inherit from Calcite rule and just override match.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] atoomula commented on a change in pull request #1384: SAMZA-2549 - Samza-sql: Add query optimizations for remote table joins
Posted by GitBox <gi...@apache.org>.
atoomula commented on a change in pull request #1384:
URL: https://github.com/apache/samza/pull/1384#discussion_r446711831
##########
File path: samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java
##########
@@ -140,16 +171,46 @@ public RelRoot plan(String query) {
.operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
.sqlToRelConverterConfig(SqlToRelConverter.Config.DEFAULT)
.traitDefs(traitDefs)
- .context(Contexts.EMPTY_CONTEXT)
- .costFactory(null)
+ .programs(Programs.hep(rules, true, DefaultRelMetadataProvider.INSTANCE))
.build();
- Planner planner = Frameworks.getPlanner(frameworkConfig);
+ planner = Frameworks.getPlanner(frameworkConfig);
+ return planner;
+ } catch (Exception e) {
+ String errorMsg = "Failed to create planner.";
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+ private RelRoot optimize(RelRoot relRoot) {
+ RelTraitSet relTraitSet = RelTraitSet.createEmpty();
+ relTraitSet = relTraitSet.plus(EnumerableConvention.INSTANCE);
+ try {
+ RelRoot optimizedRelRoot =
+ RelRoot.of(getPlanner().transform(0, relTraitSet, relRoot.project()), SqlKind.SELECT);
+ LOG.info("query plan with optimization:\n"
+ + RelOptUtil.toString(optimizedRelRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
+ return optimizedRelRoot;
+ } catch (Exception e) {
+ String errorMsg =
+ "Error while optimizing query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
+ LOG.error(errorMsg, e);
+ throw new SamzaException(errorMsg, e);
+ }
+ }
+
+ public RelRoot plan(String query) {
+ try {
+ Planner planner = getPlanner();
SqlNode sql = planner.parse(query);
SqlNode validatedSql = planner.validate(sql);
RelRoot relRoot = planner.rel(validatedSql);
- LOG.info("query plan:\n" + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES));
- return relRoot;
+ LOG.info("query plan without optimization:\n"
+ + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.EXPPLAN_ATTRIBUTES));
Review comment:
I'm not strongly against turning it on by default. I turned it on by default now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org