KYLIN-2758 Complete input SQL's default database name when the query is entering query pushdown engine
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/bc347b7b Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/bc347b7b Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/bc347b7b Branch: refs/heads/2.1.x Commit: bc347b7b213e881b689a813f220390716e9b04b6 Parents: 625e706 Author: Jiatao Tao <245915...@qq.com> Authored: Fri Jul 28 03:29:54 2017 -0500 Committer: GitHub <nore...@github.com> Committed: Fri Jul 28 03:29:54 2017 -0500 ---------------------------------------------------------------------- .../metadata/model/tool/CalciteParser.java | 13 + .../org/apache/kylin/query/KylinTestBase.java | 4 +- query/pom.xml | 7 + .../apache/kylin/query/util/PushDownUtil.java | 294 +++++++++++++++++++ .../kylin/query/util/PushDownUtilTest.java | 163 ++++++++++ .../apache/kylin/rest/service/QueryService.java | 4 +- .../apache/kylin/rest/util/PushDownUtil.java | 180 ------------ .../kylin/rest/util/PushDownUtilTest.java | 94 ------ 8 files changed, 481 insertions(+), 278 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/core-metadata/src/main/java/org/apache/kylin/metadata/model/tool/CalciteParser.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/tool/CalciteParser.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/tool/CalciteParser.java index 5352708..a48abae 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/tool/CalciteParser.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/tool/CalciteParser.java @@ -68,6 +68,19 @@ public class CalciteParser { } + public static SqlNode getFromNode(String sql) { + //When the sql have limit clause, calcite will parse it as a SqlOrder Object. + sql = sql.split("LIMIT")[0]; + SqlNode fromNode = null; + try { + fromNode = ((SqlSelect) (CalciteParser.parse(sql))).getFrom(); + } catch (SqlParseException e) { + throw new RuntimeException("Failed to parse expression \'" + sql + + "\', please make sure the expression is valid"); + } + return fromNode; + } + public static boolean isNodeEqual(SqlNode node0, SqlNode node1) { if (node0 == null) { return node1 == null; http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/kylin-it/src/test/java/org/apache/kylin/query/KylinTestBase.java ---------------------------------------------------------------------- diff --git a/kylin-it/src/test/java/org/apache/kylin/query/KylinTestBase.java b/kylin-it/src/test/java/org/apache/kylin/query/KylinTestBase.java index e1d4a44..0c6feb3 100644 --- a/kylin-it/src/test/java/org/apache/kylin/query/KylinTestBase.java +++ b/kylin-it/src/test/java/org/apache/kylin/query/KylinTestBase.java @@ -51,7 +51,7 @@ import org.apache.kylin.metadata.project.ProjectInstance; import org.apache.kylin.metadata.querymeta.SelectedColumnMeta; import org.apache.kylin.query.relnode.OLAPContext; import org.apache.kylin.query.routing.rules.RemoveBlackoutRealizationsRule; -import org.apache.kylin.rest.util.PushDownUtil; +import org.apache.kylin.query.util.PushDownUtil; import org.dbunit.DatabaseUnitException; import org.dbunit.database.DatabaseConfig; import org.dbunit.database.DatabaseConnection; @@ -263,7 +263,7 @@ public class KylinTestBase { } catch (SQLException sqlException) { List<List<String>> results = Lists.newArrayList(); List<SelectedColumnMeta> columnMetas = Lists.newArrayList(); - boolean b = PushDownUtil.doPushDownQuery(ProjectInstance.DEFAULT_PROJECT_NAME, sql, results, columnMetas, + boolean b = PushDownUtil.doPushDownQuery(ProjectInstance.DEFAULT_PROJECT_NAME, "DEFAULT", sql, results, columnMetas, sqlException); if (!b) { throw sqlException; http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/query/pom.xml ---------------------------------------------------------------------- diff --git a/query/pom.xml b/query/pom.xml index 941eb8a..24c63f9 100644 --- a/query/pom.xml +++ b/query/pom.xml @@ -70,5 +70,12 @@ <artifactId>junit</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <scope>test</scope> + <!--MRUnit relies on older version of mockito, so cannot manage it globally--> + <version>${mockito.version}</version> + </dependency> </dependencies> </project> http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/query/src/main/java/org/apache/kylin/query/util/PushDownUtil.java ---------------------------------------------------------------------- diff --git a/query/src/main/java/org/apache/kylin/query/util/PushDownUtil.java b/query/src/main/java/org/apache/kylin/query/util/PushDownUtil.java new file mode 100644 index 0000000..f496626 --- /dev/null +++ b/query/src/main/java/org/apache/kylin/query/util/PushDownUtil.java @@ -0,0 +1,294 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +package org.apache.kylin.query.util; + +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.apache.calcite.sql.SqlBasicCall; +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlDataTypeSpec; +import org.apache.calcite.sql.SqlDynamicParam; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlIntervalQualifier; +import org.apache.calcite.sql.SqlJoin; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.util.SqlVisitor; +import org.apache.commons.lang.StringUtils; +import org.apache.commons.lang.text.StrBuilder; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.commons.lang3.tuple.Triple; +import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.util.ClassUtil; +import org.apache.kylin.metadata.MetadataManager; +import org.apache.kylin.metadata.model.ComputedColumnDesc; +import org.apache.kylin.metadata.model.DataModelDesc; +import org.apache.kylin.metadata.model.tool.CalciteParser; +import org.apache.kylin.metadata.querymeta.SelectedColumnMeta; +import org.apache.kylin.query.routing.NoRealizationFoundException; +import org.apache.kylin.source.adhocquery.IPushDownConverter; +import org.apache.kylin.source.adhocquery.IPushDownRunner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; + +public class PushDownUtil { + private static final Logger logger = LoggerFactory.getLogger(PushDownUtil.class); + + public static boolean doPushDownQuery(String project, String sql, String schema, List<List<String>> results, + List<SelectedColumnMeta> columnMetas, SQLException sqlException) throws Exception { + + KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv(); + if (!kylinConfig.isPushDownEnabled()) { + return false; + } + + Throwable rootCause = ExceptionUtils.getRootCause(sqlException); + boolean isExpectedCause = rootCause != null && (rootCause.getClass().equals(NoRealizationFoundException.class)); + + if (isExpectedCause) { + logger.info("Query failed to utilize pre-calculation, routing to other engines", sqlException); + IPushDownRunner runner = (IPushDownRunner) ClassUtil.newInstance(kylinConfig.getPushDownRunnerClassName()); + IPushDownConverter converter = (IPushDownConverter) ClassUtil + .newInstance(kylinConfig.getPushDownConverterClassName()); + + runner.init(kylinConfig); + + logger.debug("Query pushdown runner {}", runner); + + String expandCC = restoreComputedColumnToExpr(sql, project); + if (!StringUtils.equals(expandCC, sql)) { + logger.info("computed column in sql is expanded to: " + expandCC); + } + if (schema != null && !schema.equals("DEFAULT")) { + expandCC = schemaCompletion(expandCC, schema); + } + String adhocSql = converter.convert(expandCC); + if (!adhocSql.equals(expandCC)) { + logger.info("the query is converted to {} according to kylin.query.pushdown.converter-class-name", + adhocSql); + } + + runner.executeQuery(adhocSql, results, columnMetas); + return true; + } else { + return false; + } + } + + static String schemaCompletion(String inputSql, String schema) { + if (inputSql == null || inputSql.equals("")) { + return ""; + } + SqlNode fromNode = CalciteParser.getFromNode(inputSql); + + // get all table node that don't have schema by visitor pattern + FromTablesVisitor ftv = new FromTablesVisitor(); + fromNode.accept(ftv); + List<SqlNode> tablesWithoutSchema = ftv.getTablesWithoutSchema(); + + List<Pair<Integer, Integer>> tablesPos = new ArrayList<>(); + for (SqlNode tables : tablesWithoutSchema) { + tablesPos.add(CalciteParser.getReplacePos(tables, inputSql)); + } + + // make the behind position in the front of the list, so that the front position will not be affected when replaced + Collections.sort(tablesPos); + Collections.reverse(tablesPos); + + StrBuilder afterConvert = new StrBuilder(inputSql); + for (Pair<Integer, Integer> pos : tablesPos) { + String tableWithSchema = schema + "." + inputSql.substring(pos.getLeft(), pos.getRight()); + afterConvert.replace(pos.getLeft(), pos.getRight(), tableWithSchema); + } + return afterConvert.toString(); + } + + private final static Pattern identifierInSqlPattern = Pattern.compile( + //find pattern like "table"."column" or "column" + "((?<![\\p{L}_0-9\\.\\\"])(\\\"[\\p{L}_0-9]+\\\"\\.)?(\\\"[\\p{L}_0-9]+\\\")(?![\\p{L}_0-9\\.\\\"]))" + "|" + //find pattern like table.column or column + + "((?<![\\p{L}_0-9\\.\\\"])([\\p{L}_0-9]+\\.)?([\\p{L}_0-9]+)(?![\\p{L}_0-9\\.\\\"]))"); + + private final static Pattern identifierInExprPattern = Pattern.compile( + // a.b.c + "((?<![\\p{L}_0-9\\.\\\"])([\\p{L}_0-9]+\\.)([\\p{L}_0-9]+\\.)([\\p{L}_0-9]+)(?![\\p{L}_0-9\\.\\\"]))"); + + private final static Pattern endWithAsPattern = Pattern.compile("\\s+as\\s+$", Pattern.CASE_INSENSITIVE); + + public static String restoreComputedColumnToExpr(String beforeSql, String project) { + final MetadataManager metadataManager = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()); + List<DataModelDesc> dataModelDescs = metadataManager.getModels(project); + + String afterSql = beforeSql; + for (DataModelDesc dataModelDesc : dataModelDescs) { + for (ComputedColumnDesc computedColumnDesc : dataModelDesc.getComputedColumnDescs()) { + afterSql = restoreComputedColumnToExpr(afterSql, computedColumnDesc); + } + } + return afterSql; + } + + static String restoreComputedColumnToExpr(String sql, ComputedColumnDesc computedColumnDesc) { + + String ccName = computedColumnDesc.getColumnName(); + List<Triple<Integer, Integer, String>> replacements = Lists.newArrayList(); + Matcher matcher = identifierInSqlPattern.matcher(sql); + + while (matcher.find()) { + if (matcher.group(1) != null) { //with quote case: "TABLE"."COLUMN" + + String quotedColumnName = matcher.group(3); + Preconditions.checkNotNull(quotedColumnName); + String columnName = StringUtils.strip(quotedColumnName, "\""); + if (!columnName.equalsIgnoreCase(ccName)) { + continue; + } + + if (matcher.group(2) != null) { // table name exist + String quotedTableAlias = StringUtils.strip(matcher.group(2), "."); + String tableAlias = StringUtils.strip(quotedTableAlias, "\""); + replacements.add(Triple.of(matcher.start(1), matcher.end(1), + replaceIdentifierInExpr(computedColumnDesc.getExpression(), tableAlias, true))); + } else { //only column + if (endWithAsPattern.matcher(sql.substring(0, matcher.start(1))).find()) { + //select DEAL_AMOUNT as "deal_amount" case + continue; + } + replacements.add(Triple.of(matcher.start(1), matcher.end(1), + replaceIdentifierInExpr(computedColumnDesc.getExpression(), null, true))); + } + } else if (matcher.group(4) != null) { //without quote case: table.column or simply column + String columnName = matcher.group(6); + Preconditions.checkNotNull(columnName); + if (!columnName.equalsIgnoreCase(ccName)) { + continue; + } + + if (matcher.group(5) != null) { //table name exist + String tableAlias = StringUtils.strip(matcher.group(5), "."); + replacements.add(Triple.of(matcher.start(4), matcher.end(4), + replaceIdentifierInExpr(computedColumnDesc.getExpression(), tableAlias, false))); + + } else { //only column + if (endWithAsPattern.matcher(sql.substring(0, matcher.start(4))).find()) { + //select DEAL_AMOUNT as deal_amount case + continue; + } + replacements.add(Triple.of(matcher.start(4), matcher.end(4), + replaceIdentifierInExpr(computedColumnDesc.getExpression(), null, false))); + } + } + } + + Collections.reverse(replacements); + for (Triple<Integer, Integer, String> triple : replacements) { + sql = sql.substring(0, triple.getLeft()) + "(" + triple.getRight() + ")" + + sql.substring(triple.getMiddle()); + } + return sql; + } + + static String replaceIdentifierInExpr(String expr, String tableAlias, boolean quoted) { + if (tableAlias == null) { + return expr; + } + + return CalciteParser.insertAliasInExpr(expr, tableAlias); + } +} + +/** + * Created by jiatao.tao + * Get all the tables from "FROM" clause that without schema + */ +class FromTablesVisitor implements SqlVisitor<SqlNode> { + private List<SqlNode> tables; + + FromTablesVisitor() { + this.tables = new ArrayList<>(); + } + + List<SqlNode> getTablesWithoutSchema() { + return tables; + } + + @Override + public SqlNode visit(SqlNodeList nodeList) { + return null; + } + + @Override + public SqlNode visit(SqlLiteral literal) { + return null; + } + + @Override + public SqlNode visit(SqlCall call) { + if (call instanceof SqlBasicCall) { + SqlBasicCall node = (SqlBasicCall) call; + node.getOperands()[0].accept(this); + return null; + } + if (call instanceof SqlJoin) { + SqlJoin node = (SqlJoin) call; + node.getLeft().accept(this); + node.getRight().accept(this); + return null; + } + for (SqlNode operand : call.getOperandList()) { + if (operand != null) { + operand.accept(this); + } + } + return null; + } + + @Override + public SqlNode visit(SqlIdentifier id) { + if (id.names.size() == 1) { + tables.add(id); + } + return null; + } + + @Override + public SqlNode visit(SqlDataTypeSpec type) { + return null; + } + + @Override + public SqlNode visit(SqlDynamicParam param) { + return null; + } + + @Override + public SqlNode visit(SqlIntervalQualifier intervalQualifier) { + return null; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/query/src/test/java/org/apache/kylin/query/util/PushDownUtilTest.java ---------------------------------------------------------------------- diff --git a/query/src/test/java/org/apache/kylin/query/util/PushDownUtilTest.java b/query/src/test/java/org/apache/kylin/query/util/PushDownUtilTest.java new file mode 100644 index 0000000..e2ccd9e --- /dev/null +++ b/query/src/test/java/org/apache/kylin/query/util/PushDownUtilTest.java @@ -0,0 +1,163 @@ +/* + * 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.kylin.query.util; + +import org.apache.kylin.metadata.model.ComputedColumnDesc; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +public class PushDownUtilTest { + @Test + public void testSchemaCompletion() { + String sql1 = "SELECT a \n"+ + "FROM a.KYLIN_SALES as KYLIN_SALES\n" + + "INNER JOIN \"A\".KYLIN_ACCOUNT as BUYER_ACCOUNT\n" + + "ON KYLIN_SALES.BUYER_ID = BUYER_ACCOUNT.ACCOUNT_ID\n" + + "INNER JOIN \"KYLIN_COUNTRY\" as BUYER_COUNTRY\n" + + "ON BUYER_ACCOUNT.ACCOUNT_COUNTRY = BUYER_COUNTRY.COUNTRY"; + String sql2 = "select * from DB2.t,DB2.tt,ttt"; + + String sql3 = "SELECT t1.week_beg_dt, t1.sum_price, t2.cnt\n" + + "FROM (\n" + + " select test_cal_dt.week_beg_dt, sum(price) as sum_price\n" + + " from DB1.\"test_kylin_fact\"\n" + + " inner JOIN test_cal_dt as test_cal_dt\n" + + " ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt\n" + + " inner JOIN test_category_groupings\n" + + " ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id\n" + + " inner JOIN test_sites as test_sites\n" + + " ON test_kylin_fact.lstg_site_id = test_sites.site_id\n" + + " group by test_cal_dt.week_beg_dt\n" + + ") t1\n" + + "inner join (\n" + + " select test_cal_dt.week_beg_dt, count(*) as cnt\n" + + " from DB1.test_kylin_fact\n" + + " inner JOIN test_cal_dt as test_cal_dt\n" + + " ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt\n" + + " inner JOIN test_category_groupings\n" + + " ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id\n" + + " inner JOIN test_sites as test_sites\n" + + " ON test_kylin_fact.lstg_site_id = test_sites.site_id\n" + + " group by test_cal_dt.week_beg_dt\n" + + ") t2\n" + + "on t1.week_beg_dt=t2.week_beg_dt"; + + String exceptSQL1 = "SELECT a \n" + + "FROM a.KYLIN_SALES as KYLIN_SALES\n" + + "INNER JOIN \"A\".KYLIN_ACCOUNT as BUYER_ACCOUNT\n" + + "ON KYLIN_SALES.BUYER_ID = BUYER_ACCOUNT.ACCOUNT_ID\n" + + "INNER JOIN EDW.\"KYLIN_COUNTRY\" as BUYER_COUNTRY\n" + + "ON BUYER_ACCOUNT.ACCOUNT_COUNTRY = BUYER_COUNTRY.COUNTRY"; + + String exceptSQL2 = "select * from DB2.t,DB2.tt,EDW.ttt"; + + String exceptSQL3 = "SELECT t1.week_beg_dt, t1.sum_price, t2.cnt\n" + + "FROM (\n" + + " select test_cal_dt.week_beg_dt, sum(price) as sum_price\n" + + " from DB1.\"test_kylin_fact\"\n" + + " inner JOIN EDW.test_cal_dt as test_cal_dt\n" + + " ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt\n" + + " inner JOIN EDW.test_category_groupings\n" + + " ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id\n" + + " inner JOIN EDW.test_sites as test_sites\n" + + " ON test_kylin_fact.lstg_site_id = test_sites.site_id\n" + + " group by test_cal_dt.week_beg_dt\n" + + ") t1\n" + + "inner join (\n" + + " select test_cal_dt.week_beg_dt, count(*) as cnt\n" + + " from DB1.test_kylin_fact\n" + + " inner JOIN EDW.test_cal_dt as test_cal_dt\n" + + " ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt\n" + + " inner JOIN EDW.test_category_groupings\n" + + " ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id\n" + + " inner JOIN EDW.test_sites as test_sites\n" + + " ON test_kylin_fact.lstg_site_id = test_sites.site_id\n" + + " group by test_cal_dt.week_beg_dt\n" + + ") t2\n" + + "on t1.week_beg_dt=t2.week_beg_dt"; + Assert.assertEquals(exceptSQL1, PushDownUtil.schemaCompletion(sql1, "EDW")); + Assert.assertEquals(exceptSQL2, PushDownUtil.schemaCompletion(sql2, "EDW")); + Assert.assertEquals(exceptSQL3, PushDownUtil.schemaCompletion(sql3, "EDW")); + } + + @Test + public void testReplaceIdentifierInExpr() { + { + String ret = PushDownUtil.replaceIdentifierInExpr("x * y", null, false); + Assert.assertEquals("x * y", ret); + } + { + String ret = PushDownUtil.replaceIdentifierInExpr("x_3 * y_3", "b_2", false); + Assert.assertEquals("b_2.x_3 * b_2.y_3", ret); + } + { + String ret = PushDownUtil.replaceIdentifierInExpr("substr(x,1,3)>y", "c", true); + Assert.assertEquals("substr(c.x,1,3)>c.y", ret); + } + { + String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", "c", true); + Assert.assertEquals("strcmp(substr(c.x,1,3),c.y)", ret); + } + { + String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", null, true); + Assert.assertEquals("strcmp(substr(x,1,3),y)", ret); + } + { + String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", null, false); + Assert.assertEquals("strcmp(substr(x,1,3),y)", ret); + } + } + + @Test + public void testRestoreComputedColumnToExpr() { + + ComputedColumnDesc computedColumnDesc = Mockito.mock(ComputedColumnDesc.class); + Mockito.when(computedColumnDesc.getColumnName()).thenReturn("DEAL_AMOUNT"); + Mockito.when(computedColumnDesc.getExpression()).thenReturn("price * number"); + { + String ret = PushDownUtil.restoreComputedColumnToExpr( + "select DEAL_AMOUNT from DB.TABLE group by date order by DEAL_AMOUNT", computedColumnDesc); + Assert.assertEquals("select (price * number) from DB.TABLE group by date order by (price * number)", ret); + } + { + String ret = PushDownUtil.restoreComputedColumnToExpr( + "select DEAL_AMOUNT as DEAL_AMOUNT from DB.TABLE group by date order by DEAL_AMOUNT", + computedColumnDesc); + Assert.assertEquals( + "select (price * number) as DEAL_AMOUNT from DB.TABLE group by date order by (price * number)", + ret); + } + { + String ret = PushDownUtil.restoreComputedColumnToExpr( + "select \"DEAL_AMOUNT\" AS deal_amount from DB.TABLE group by date order by DEAL_AMOUNT", + computedColumnDesc); + Assert.assertEquals( + "select (price * number) AS deal_amount from DB.TABLE group by date order by (price * number)", + ret); + } + { + String ret = PushDownUtil.restoreComputedColumnToExpr( + "select x.DEAL_AMOUNT AS deal_amount from DB.TABLE x group by date order by x.DEAL_AMOUNT", + computedColumnDesc); + Assert.assertEquals( + "select (x.price * x.number) AS deal_amount from DB.TABLE x group by date order by (x.price * x.number)", + ret); + } + } +} http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java ---------------------------------------------------------------------- diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java index 1c55f6b..ab47caa 100644 --- a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java +++ b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java @@ -94,7 +94,7 @@ import org.apache.kylin.rest.request.PrepareSqlRequest; import org.apache.kylin.rest.request.SQLRequest; import org.apache.kylin.rest.response.SQLResponse; import org.apache.kylin.rest.util.AclUtil; -import org.apache.kylin.rest.util.PushDownUtil; +import org.apache.kylin.query.util.PushDownUtil; import org.apache.kylin.rest.util.TableauInterceptor; import org.apache.kylin.storage.hybrid.HybridInstance; import org.slf4j.Logger; @@ -791,7 +791,7 @@ public class QueryService extends BasicService { results.add(oneRow); } } catch (SQLException sqlException) { - isPushDown = PushDownUtil.doPushDownQuery(sqlRequest.getProject(), correctedSql, results, columnMetas, + isPushDown = PushDownUtil.doPushDownQuery(sqlRequest.getProject(), correctedSql, conn.getSchema(), results, columnMetas, sqlException); if (!isPushDown) { throw sqlException; http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/server-base/src/main/java/org/apache/kylin/rest/util/PushDownUtil.java ---------------------------------------------------------------------- diff --git a/server-base/src/main/java/org/apache/kylin/rest/util/PushDownUtil.java b/server-base/src/main/java/org/apache/kylin/rest/util/PushDownUtil.java deleted file mode 100644 index c21d5c2..0000000 --- a/server-base/src/main/java/org/apache/kylin/rest/util/PushDownUtil.java +++ /dev/null @@ -1,180 +0,0 @@ -/* - * 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.kylin.rest.util; - -import java.sql.SQLException; -import java.util.Collections; -import java.util.List; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang3.exception.ExceptionUtils; -import org.apache.commons.lang3.tuple.Triple; -import org.apache.kylin.common.KylinConfig; -import org.apache.kylin.common.util.ClassUtil; -import org.apache.kylin.metadata.MetadataManager; -import org.apache.kylin.metadata.model.ComputedColumnDesc; -import org.apache.kylin.metadata.model.DataModelDesc; -import org.apache.kylin.metadata.model.tool.CalciteParser; -import org.apache.kylin.metadata.querymeta.SelectedColumnMeta; -import org.apache.kylin.query.routing.NoRealizationFoundException; -import org.apache.kylin.source.adhocquery.IPushDownConverter; -import org.apache.kylin.source.adhocquery.IPushDownRunner; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; - -public class PushDownUtil { - private static final Logger logger = LoggerFactory.getLogger(PushDownUtil.class); - - public static boolean doPushDownQuery(String project, String sql, List<List<String>> results, - List<SelectedColumnMeta> columnMetas, SQLException sqlException) throws Exception { - - KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv(); - if (!kylinConfig.isPushDownEnabled()) { - return false; - } - - Throwable rootCause = ExceptionUtils.getRootCause(sqlException); - boolean isExpectedCause = rootCause != null && (rootCause.getClass().equals(NoRealizationFoundException.class)); - - if (isExpectedCause) { - logger.info("Query failed to utilize pre-calculation, routing to other engines", sqlException); - IPushDownRunner runner = (IPushDownRunner) ClassUtil.newInstance(kylinConfig.getPushDownRunnerClassName()); - IPushDownConverter converter = (IPushDownConverter) ClassUtil - .newInstance(kylinConfig.getPushDownConverterClassName()); - - runner.init(kylinConfig); - - logger.debug("Query pushdown runner {}", runner); - - String expandCC = restoreComputedColumnToExpr(sql, project); - if (!StringUtils.equals(expandCC, sql)) { - logger.info("computed column in sql is expanded to: " + expandCC); - } - String adhocSql = converter.convert(expandCC); - if (!adhocSql.equals(expandCC)) { - logger.info("the query is converted to {} according to kylin.query.pushdown.converter-class-name", - adhocSql); - } - - runner.executeQuery(adhocSql, results, columnMetas); - - return true; - } else { - return false; - } - } - - private final static Pattern identifierInSqlPattern = Pattern.compile( - //find pattern like "table"."column" or "column" - "((?<![\\p{L}_0-9\\.\\\"])(\\\"[\\p{L}_0-9]+\\\"\\.)?(\\\"[\\p{L}_0-9]+\\\")(?![\\p{L}_0-9\\.\\\"]))" + "|" - //find pattern like table.column or column - + "((?<![\\p{L}_0-9\\.\\\"])([\\p{L}_0-9]+\\.)?([\\p{L}_0-9]+)(?![\\p{L}_0-9\\.\\\"]))"); - - private final static Pattern identifierInExprPattern = Pattern.compile( - // a.b.c - "((?<![\\p{L}_0-9\\.\\\"])([\\p{L}_0-9]+\\.)([\\p{L}_0-9]+\\.)([\\p{L}_0-9]+)(?![\\p{L}_0-9\\.\\\"]))"); - - private final static Pattern endWithAsPattern = Pattern.compile("\\s+as\\s+$", Pattern.CASE_INSENSITIVE); - - public static String restoreComputedColumnToExpr(String beforeSql, String project) { - final MetadataManager metadataManager = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()); - List<DataModelDesc> dataModelDescs = metadataManager.getModels(project); - - String afterSql = beforeSql; - for (DataModelDesc dataModelDesc : dataModelDescs) { - for (ComputedColumnDesc computedColumnDesc : dataModelDesc.getComputedColumnDescs()) { - afterSql = restoreComputedColumnToExpr(afterSql, computedColumnDesc); - } - } - return afterSql; - } - - static String restoreComputedColumnToExpr(String sql, ComputedColumnDesc computedColumnDesc) { - - String ccName = computedColumnDesc.getColumnName(); - List<Triple<Integer, Integer, String>> replacements = Lists.newArrayList(); - Matcher matcher = identifierInSqlPattern.matcher(sql); - - while (matcher.find()) { - if (matcher.group(1) != null) { //with quote case: "TABLE"."COLUMN" - - String quotedColumnName = matcher.group(3); - Preconditions.checkNotNull(quotedColumnName); - String columnName = StringUtils.strip(quotedColumnName, "\""); - if (!columnName.equalsIgnoreCase(ccName)) { - continue; - } - - if (matcher.group(2) != null) { // table name exist - String quotedTableAlias = StringUtils.strip(matcher.group(2), "."); - String tableAlias = StringUtils.strip(quotedTableAlias, "\""); - replacements.add(Triple.of(matcher.start(1), matcher.end(1), - replaceIdentifierInExpr(computedColumnDesc.getExpression(), tableAlias, true))); - } else { //only column - if (endWithAsPattern.matcher(sql.substring(0, matcher.start(1))).find()) { - //select DEAL_AMOUNT as "deal_amount" case - continue; - } - replacements.add(Triple.of(matcher.start(1), matcher.end(1), - replaceIdentifierInExpr(computedColumnDesc.getExpression(), null, true))); - } - } else if (matcher.group(4) != null) { //without quote case: table.column or simply column - String columnName = matcher.group(6); - Preconditions.checkNotNull(columnName); - if (!columnName.equalsIgnoreCase(ccName)) { - continue; - } - - if (matcher.group(5) != null) { //table name exist - String tableAlias = StringUtils.strip(matcher.group(5), "."); - replacements.add(Triple.of(matcher.start(4), matcher.end(4), - replaceIdentifierInExpr(computedColumnDesc.getExpression(), tableAlias, false))); - - } else { //only column - if (endWithAsPattern.matcher(sql.substring(0, matcher.start(4))).find()) { - //select DEAL_AMOUNT as deal_amount case - continue; - } - replacements.add(Triple.of(matcher.start(4), matcher.end(4), - replaceIdentifierInExpr(computedColumnDesc.getExpression(), null, false))); - } - } - } - - Collections.reverse(replacements); - for (Triple<Integer, Integer, String> triple : replacements) { - sql = sql.substring(0, triple.getLeft()) + "(" + triple.getRight() + ")" - + sql.substring(triple.getMiddle()); - } - return sql; - } - - static String replaceIdentifierInExpr(String expr, String tableAlias, boolean quoted) { - if (tableAlias == null) { - return expr; - } - - return CalciteParser.insertAliasInExpr(expr, tableAlias); - } -} http://git-wip-us.apache.org/repos/asf/kylin/blob/bc347b7b/server-base/src/test/java/org/apache/kylin/rest/util/PushDownUtilTest.java ---------------------------------------------------------------------- diff --git a/server-base/src/test/java/org/apache/kylin/rest/util/PushDownUtilTest.java b/server-base/src/test/java/org/apache/kylin/rest/util/PushDownUtilTest.java deleted file mode 100644 index 5302a70..0000000 --- a/server-base/src/test/java/org/apache/kylin/rest/util/PushDownUtilTest.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * 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.kylin.rest.util; - -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import org.apache.kylin.metadata.model.ComputedColumnDesc; -import org.junit.Assert; -import org.junit.Test; - -public class PushDownUtilTest { - - @Test - public void testReplaceIdentifierInExpr() { - { - String ret = PushDownUtil.replaceIdentifierInExpr("x * y", null, false); - Assert.assertEquals("x * y", ret); - } - { - String ret = PushDownUtil.replaceIdentifierInExpr("x_3 * y_3", "b_2", false); - Assert.assertEquals("b_2.x_3 * b_2.y_3", ret); - } - { - String ret = PushDownUtil.replaceIdentifierInExpr("substr(x,1,3)>y", "c", true); - Assert.assertEquals("substr(c.x,1,3)>c.y", ret); - } - { - String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", "c", true); - Assert.assertEquals("strcmp(substr(c.x,1,3),c.y)", ret); - } - { - String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", null, true); - Assert.assertEquals("strcmp(substr(x,1,3),y)", ret); - } - { - String ret = PushDownUtil.replaceIdentifierInExpr("strcmp(substr(x,1,3),y)", null, false); - Assert.assertEquals("strcmp(substr(x,1,3),y)", ret); - } - } - - @Test - public void testRestoreComputedColumnToExpr() { - - ComputedColumnDesc computedColumnDesc = mock(ComputedColumnDesc.class); - when(computedColumnDesc.getColumnName()).thenReturn("DEAL_AMOUNT"); - when(computedColumnDesc.getExpression()).thenReturn("price * number"); - - { - String ret = PushDownUtil.restoreComputedColumnToExpr( - "select DEAL_AMOUNT from DB.TABLE group by date order by DEAL_AMOUNT", computedColumnDesc); - Assert.assertEquals("select (price * number) from DB.TABLE group by date order by (price * number)", ret); - } - { - String ret = PushDownUtil.restoreComputedColumnToExpr( - "select DEAL_AMOUNT as DEAL_AMOUNT from DB.TABLE group by date order by DEAL_AMOUNT", - computedColumnDesc); - Assert.assertEquals( - "select (price * number) as DEAL_AMOUNT from DB.TABLE group by date order by (price * number)", - ret); - } - { - String ret = PushDownUtil.restoreComputedColumnToExpr( - "select \"DEAL_AMOUNT\" AS deal_amount from DB.TABLE group by date order by DEAL_AMOUNT", - computedColumnDesc); - Assert.assertEquals( - "select (price * number) AS deal_amount from DB.TABLE group by date order by (price * number)", - ret); - } - { - String ret = PushDownUtil.restoreComputedColumnToExpr( - "select x.DEAL_AMOUNT AS deal_amount from DB.TABLE x group by date order by x.DEAL_AMOUNT", - computedColumnDesc); - Assert.assertEquals( - "select (x.price * x.number) AS deal_amount from DB.TABLE x group by date order by (x.price * x.number)", - ret); - } - } -}
