Skip to content

Commit

Permalink
KYLIN-2820 Query can't read window function's result from subquery
Browse files Browse the repository at this point in the history
  • Loading branch information
@nichunen @luguosheng1314
nichunen authored and luguosheng1314 committed Aug 11, 2019
1 parent afe076a commit abc0ab8
Show file tree
Hide file tree
Showing 7 changed files with 153 additions and 81 deletions.
104 changes: 58 additions & 46 deletions core-metadata/src/main/java/org/apache/kylin/metadata/model/TblColRef.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,40 +21,50 @@
import static com.google.common.base.Preconditions.checkArgument;

import java.io.Serializable;

import java.util.List;
import java.util.Locale;
import java.util.function.Function;

import org.apache.commons.lang.StringUtils;
import org.apache.kylin.metadata.datatype.DataType;
import org.apache.kylin.metadata.expression.TupleExpression;

/**
*/
@SuppressWarnings({ "serial" })
public class TblColRef implements Serializable {

private static final String INNER_TABLE_NAME = "_kylin_table";
private static final DataModelDesc UNKNOWN_MODEL = new DataModelDesc();

// used by projection rewrite, see OLAPProjectRel
public enum InnerDataTypeEnum {

LITERAL("_literal_type"), DERIVED("_derived_type");

private final String dateType;
static {
UNKNOWN_MODEL.setName("UNKNOWN_MODEL");
}

private InnerDataTypeEnum(String name) {
this.dateType = name;
}
private TableRef table;
private TableRef backupTable;// only used in fixTableRef()
private ColumnDesc column;
private String identity;
private String parserDescription;
//used in window function
private List<TupleExpression> subTupleExps;
/**
* Function used to get quoted identitier
*/
private transient Function<TblColRef, String> quotedFunc;

public String getDataType() {
return dateType;
}
TblColRef(ColumnDesc column) {
this.column = column;
}

public static boolean contains(String name) {
return LITERAL.getDataType().equals(name) || DERIVED.getDataType().equals(name);
}
TblColRef(TableRef table, ColumnDesc column) {
checkArgument(table.getTableDesc().getIdentity().equals(column.getTable().getIdentity()));
this.table = table;
this.column = column;
}

// ============================================================================

// used by projection rewrite, see OLAPProjectRel
public static TblColRef newInnerColumn(String columnName, InnerDataTypeEnum dataType) {
return newInnerColumn(columnName, dataType, null);
Expand All @@ -72,11 +82,6 @@ public static TblColRef newInnerColumn(String columnName, InnerDataTypeEnum data
return colRef;
}

private static final DataModelDesc UNKNOWN_MODEL = new DataModelDesc();
static {
UNKNOWN_MODEL.setName("UNKNOWN_MODEL");
}

public static TableRef tableForUnknownModel(String tempTableAlias, TableDesc table) {
return new TableRef(UNKNOWN_MODEL, tempTableAlias, table, false);
}
Expand All @@ -103,7 +108,8 @@ public static TblColRef mockup(TableDesc table, int oneBasedColumnIndex, String
}

// for test mainly
public static TblColRef mockup(TableDesc table, int oneBasedColumnIndex, String name, String datatype, String comment) {
public static TblColRef mockup(TableDesc table, int oneBasedColumnIndex, String name, String datatype,
String comment) {
ColumnDesc desc = new ColumnDesc();
String id = "" + oneBasedColumnIndex;
desc.setId(id);
Expand All @@ -114,33 +120,10 @@ public static TblColRef mockup(TableDesc table, int oneBasedColumnIndex, String
return new TblColRef(desc);
}

// ============================================================================

private TableRef table;
private TableRef backupTable;// only used in fixTableRef()
private ColumnDesc column;
private String identity;
private String parserDescription;

/**
* Function used to get quoted identitier
*/
private transient Function<TblColRef, String> quotedFunc;

public void setQuotedFunc(Function<TblColRef, String> quotedFunc) {
this.quotedFunc = quotedFunc;
}

TblColRef(ColumnDesc column) {
this.column = column;
}

TblColRef(TableRef table, ColumnDesc column) {
checkArgument(table.getTableDesc().getIdentity().equals(column.getTable().getIdentity()));
this.table = table;
this.column = column;
}

public void fixTableRef(TableRef tableRef) {
this.backupTable = this.table;
this.table = tableRef;
Expand Down Expand Up @@ -199,9 +182,18 @@ public DataType getType() {
return column.getType();
}

public String getBackupTableAlias(){
public List<TupleExpression> getSubTupleExps() {
return subTupleExps;
}

public void setSubTupleExps(List<TupleExpression> subTubleExps) {
this.subTupleExps = subTubleExps;
}

public String getBackupTableAlias() {
return backupTable.getAlias();
}

private void markInnerColumn(InnerDataTypeEnum dataType) {
this.column.setDatatype(dataType.getDataType());
this.column.getTable().setName(INNER_TABLE_NAME);
Expand Down Expand Up @@ -286,4 +278,24 @@ public String getTableWithSchema() {
public String getColumWithTableAndSchema() {
return (getTableWithSchema() + "." + column.getName()).toUpperCase(Locale.ROOT);
}

// used by projection rewrite, see OLAPProjectRel
public enum InnerDataTypeEnum {

LITERAL("_literal_type"), DERIVED("_derived_type");

private final String dateType;

private InnerDataTypeEnum(String name) {
this.dateType = name;
}

public static boolean contains(String name) {
return LITERAL.getDataType().equals(name) || DERIVED.getDataType().equals(name);
}

public String getDataType() {
return dateType;
}
}
}
22 changes: 22 additions & 0 deletions kylin-it/src/test/resources/query/sql_window/query13.sql
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
--
-- 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.
--
select t.first_seller_id as first_seller_id_test, count(*) from (
select first_value(seller_id) over (partition by buyer_id) as first_seller_id
from test_kylin_fact inner join test_order on test_kylin_fact.order_id=test_order.order_id
)
as t group by t.first_seller_id
14 changes: 13 additions & 1 deletion query/src/main/java/org/apache/kylin/query/relnode/ColumnRowType.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.kylin.query.relnode;

import java.util.ArrayList;
import java.util.List;

import org.apache.kylin.common.util.Pair;
Expand Down Expand Up @@ -81,7 +82,7 @@ public Pair<TblColRef, TupleExpression> replaceColumnByIndex(int index, TblColRe
return new Pair<>(oldCol, oldExpr);
}

public TupleExpression getSourceColumnsByIndex(int i) {
public TupleExpression getTupleExpressionByIndex(int i) {
TupleExpression result = null;
if (sourceColumns != null) {
result = sourceColumns.get(i);
Expand All @@ -92,6 +93,17 @@ public TupleExpression getSourceColumnsByIndex(int i) {
return result;
}

public List<TupleExpression> getSourceColumns() {
if (sourceColumns == null) {
List<TupleExpression> sources = new ArrayList<>();
for (int i = 0; i < columns.size(); i++) {
sources.add(getTupleExpressionByIndex(i));
}
sourceColumns = sources;
}
return sourceColumns;
}

public List<TblColRef> getAllColumns() {
return columns;
}
Expand Down
50 changes: 26 additions & 24 deletions query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@

package org.apache.kylin.query.relnode;

import static org.apache.kylin.metadata.expression.TupleExpression.ExpressionOperatorEnum.COLUMN;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
Expand Down Expand Up @@ -85,7 +87,6 @@
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import static org.apache.kylin.metadata.expression.TupleExpression.ExpressionOperatorEnum.COLUMN;
/**
*/
public class OLAPAggregateRel extends Aggregate implements OLAPRel {
Expand All @@ -111,30 +112,14 @@ public class OLAPAggregateRel extends Aggregate implements OLAPRel {
for (String func : udafs.keySet()) {
try {
AGGR_FUNC_PARAM_AS_MEASURE_MAP.put(func,
((ParamAsMeasureCount) (udafs.get(func).getDeclaredConstructor().newInstance())).getParamAsMeasureCount());
((ParamAsMeasureCount) (udafs.get(func).getDeclaredConstructor().newInstance()))
.getParamAsMeasureCount());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}

static String getSqlFuncName(AggregateCall aggCall) {
String sqlName = aggCall.getAggregation().getName();
if (aggCall.isDistinct()) {
sqlName = sqlName + "_DISTINCT";
}
return sqlName;
}

public static String getAggrFuncName(AggregateCall aggCall) {
String sqlName = getSqlFuncName(aggCall);
String funcName = AGGR_FUNC_MAP.get(sqlName);
if (funcName == null) {
throw new IllegalStateException("Non-support aggregation " + sqlName);
}
return funcName;
}

OLAPContext context;
ColumnRowType columnRowType;
private boolean afterAggregate;
Expand All @@ -143,7 +128,6 @@ public static String getAggrFuncName(AggregateCall aggCall) {
private List<TblColRef> groups;
private List<FunctionDesc> aggregations;
private boolean rewriting;

public OLAPAggregateRel(RelOptCluster cluster, RelTraitSet traits, RelNode child, boolean indicator,
ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls)
throws InvalidRelException {
Expand All @@ -154,6 +138,23 @@ public OLAPAggregateRel(RelOptCluster cluster, RelTraitSet traits, RelNode child
this.rowType = getRowType();
}

static String getSqlFuncName(AggregateCall aggCall) {
String sqlName = aggCall.getAggregation().getName();
if (aggCall.isDistinct()) {
sqlName = sqlName + "_DISTINCT";
}
return sqlName;
}

public static String getAggrFuncName(AggregateCall aggCall) {
String sqlName = getSqlFuncName(aggCall);
String funcName = AGGR_FUNC_MAP.get(sqlName);
if (funcName == null) {
throw new IllegalStateException("Non-support aggregation " + sqlName);
}
return funcName;
}

@Override
public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, ImmutableBitSet groupSet,
List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
Expand Down Expand Up @@ -268,10 +269,11 @@ void buildGroups() {
ColumnRowType inputColumnRowType = ((OLAPRel) getInput()).getColumnRowType();
this.groups = Lists.newArrayList();
for (int i = getGroupSet().nextSetBit(0); i >= 0; i = getGroupSet().nextSetBit(i + 1)) {
TupleExpression tupleExpression = inputColumnRowType.getSourceColumnsByIndex(i);
TupleExpression tupleExpression = inputColumnRowType.getTupleExpressionByIndex(i);

// group by column with operator
if (this.context.groupByExpression == false && !(COLUMN.equals(tupleExpression.getOperator()) && tupleExpression.getChildren().isEmpty())) {
if (this.context.groupByExpression == false
&& !(COLUMN.equals(tupleExpression.getOperator()) && tupleExpression.getChildren().isEmpty())) {
this.context.groupByExpression = true;
}

Expand Down Expand Up @@ -350,7 +352,7 @@ void buildAggregations() {
// Check dynamic aggregation
if (this.context.isDynamicColumnEnabled() && !afterAggregate && !rewriting && argList.size() == 1) {
int iRowIdx = argList.get(0);
TupleExpression tupleExpr = inputColumnRowType.getSourceColumnsByIndex(iRowIdx);
TupleExpression tupleExpr = inputColumnRowType.getTupleExpressionByIndex(iRowIdx);
if (aggCall.getAggregation() instanceof SqlSumAggFunction
|| aggCall.getAggregation() instanceof SqlSumEmptyIsZeroAggFunction) {
// sum (expression)
Expand Down Expand Up @@ -571,7 +573,7 @@ private AggregateCall rewriteAggregateCall(AggregateCall aggCall, FunctionDesc f
String callName = getSqlFuncName(aggCall);
RelDataType fieldType = aggCall.getType();
SqlAggFunction newAgg = aggCall.getAggregation();

Map<String, Class<?>> udafMap = func.getMeasureType().getRewriteCalciteAggrFunctions();
if (func.isCount()) {
newAgg = SqlStdOperatorTable.SUM0;
Expand Down
Loading

0 comments on commit abc0ab8

Please sign in to comment.