You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/11/20 04:25:01 UTC
incubator-kylin git commit: KYLIN-942-review
Repository: incubator-kylin
Updated Branches:
refs/heads/KYLIN-942-review [created] ba484d9eb
KYLIN-942-review
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/ba484d9e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/ba484d9e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/ba484d9e
Branch: refs/heads/KYLIN-942-review
Commit: ba484d9ebf0da71be7f13c14c9c7447592d982e9
Parents: d5a676a
Author: Li, Yang <ya...@ebay.com>
Authored: Fri Nov 20 11:24:47 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Fri Nov 20 11:24:47 2015 +0800
----------------------------------------------------------------------
.../kylin/cube/CubeCapabilityChecker.java | 2 +-
.../org/apache/kylin/cube/model/CubeDesc.java | 50 +++----------
.../model/validation/rule/FunctionRule.java | 2 +-
.../apache/kylin/metadata/model/ColumnDesc.java | 1 +
.../kylin/metadata/model/DatabaseDesc.java | 1 +
.../kylin/metadata/model/FunctionDesc.java | 58 ++++++++++++---
.../kylin/metadata/model/MeasureDesc.java | 12 ++++
.../kylin/metadata/model/ParameterDesc.java | 41 ++++-------
.../kylin/storage/cache/StorageMockUtils.java | 2 -
.../kylin/invertedindex/model/IIDesc.java | 14 +++-
.../invertedindex/model/IIKeyValueCodec.java | 1 -
.../kylin/query/relnode/OLAPAggregateRel.java | 2 +-
.../storage/hbase/cube/v1/CubeStorageQuery.java | 74 ++++++--------------
.../storage/hbase/cube/v2/CubeStorageQuery.java | 72 ++++++-------------
.../endpoint/EndpointTupleIterator.java | 1 -
15 files changed, 139 insertions(+), 194 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java
index 3bb246a..16353b3 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java
@@ -136,7 +136,7 @@ public class CubeCapabilityChecker {
if (digest.groupbyColumns.contains(displayCol)) {
dimensionColumnsCopy.remove(displayCol);
if (isMatchedWithDimensions(dimensionColumnsCopy, cube)) {
- if (measure.getFunction().isCompatible(onlyFunction)) {
+ if (measure.getFunction().isTopNCompatibleSum(onlyFunction)) {
return true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index 2250945..2fd560b 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -50,7 +50,6 @@ import org.apache.kylin.metadata.model.IEngineAware;
import org.apache.kylin.metadata.model.IStorageAware;
import org.apache.kylin.metadata.model.JoinDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.ParameterDesc;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.model.TblColRef;
@@ -130,7 +129,6 @@ public class CubeDesc extends RootPersistentEntity {
private LinkedHashSet<TblColRef> allColumns = new LinkedHashSet<TblColRef>();
private LinkedHashSet<TblColRef> dimensionColumns = new LinkedHashSet<TblColRef>();
- private LinkedHashSet<TblColRef> measureDisplayColumns = new LinkedHashSet<TblColRef>();
private Map<TblColRef, DeriveInfo> derivedToHostMap = Maps.newHashMap();
private Map<Array<TblColRef>, List<DeriveInfo>> hostToDerivedMap = Maps.newHashMap();
@@ -414,7 +412,7 @@ public class CubeDesc extends RootPersistentEntity {
}
return calculateSignature().equals(getSignature());
}
-
+
public String calculateSignature() {
MessageDigest md = null;
try {
@@ -646,37 +644,12 @@ public class CubeDesc extends RootPersistentEntity {
m.setDependentMeasureRef(m.getDependentMeasureRef().toUpperCase());
}
- FunctionDesc f = m.getFunction();
- f.setExpression(f.getExpression().toUpperCase());
- f.initReturnDataType();
-
- ParameterDesc p = f.getParameter();
- p.normalizeColumnValue();
-
- ArrayList<TblColRef> colRefs = Lists.newArrayList();
- if (p.isColumnType()) {
- for (String cName : p.getValue().split("\\s*,\\s*")) {
- ColumnDesc sourceColumn = factTable.findColumnByName(cName);
- TblColRef colRef = new TblColRef(sourceColumn);
- colRefs.add(colRef);
- allColumns.add(colRef);
- }
- }
-
- // for topN
- if (StringUtils.isNotEmpty(p.getDisplayColumn())) {
- ColumnDesc sourceColumn = factTable.findColumnByName(p.getDisplayColumn());
- TblColRef colRef = new TblColRef(sourceColumn);
- colRefs.add(colRef);
- measureDisplayColumns.add(colRef);
- allColumns.add(colRef);
- }
-
- if (colRefs.isEmpty() == false)
- p.setColRefs(colRefs);
+ FunctionDesc func = m.getFunction();
+ func.init(factTable);
+ allColumns.addAll(func.getParameter().getColRefs());
// verify holistic count distinct as a dependent measure
- if (m.getFunction().isHolisticCountDistinct() && StringUtils.isBlank(m.getDependentMeasureRef())) {
+ if (func.isHolisticCountDistinct() && StringUtils.isBlank(m.getDependentMeasureRef())) {
throw new IllegalStateException(m + " is a holistic count distinct but it has no DependentMeasureRef defined!");
}
}
@@ -844,24 +817,17 @@ public class CubeDesc extends RootPersistentEntity {
}
}
- for (TblColRef colRef : measureDisplayColumns) {
- if (!result.contains(colRef))
- result.add(colRef);
+ for (MeasureDesc measure : measures) {
+ result.addAll(measure.getColumnsNeedDictionary());
}
return result;
}
- public LinkedHashSet<TblColRef> getMeasureDisplayColumns() {
- return measureDisplayColumns;
- }
-
-
public boolean hasMeasureUsingDictionary() {
for (MeasureDesc measureDesc : this.getMeasures()) {
- if (measureDesc.getFunction().isTopN())
+ if (measureDesc.getColumnsNeedDictionary().size() > 0)
return true;
}
-
return false;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
index 80bd2f7..1920fc7 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
@@ -91,7 +91,7 @@ public class FunctionRule implements IValidatorRule<CubeDesc> {
if (StringUtils.equalsIgnoreCase(FunctionDesc.PARAMETER_TYPE_COLUMN, type)) {
validateColumnParameter(context, cube, value);
- } else if (StringUtils.equals(FunctionDesc.PARAMTER_TYPE_CONSTANT, type)) {
+ } else if (StringUtils.equals(FunctionDesc.PARAMETER_TYPE_CONSTANT, type)) {
validateCostantParameter(context, cube, value);
}
validateReturnType(context, cube, func);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-metadata/src/main/java/org/apache/kylin/metadata/model/ColumnDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ColumnDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ColumnDesc.java
index 12371ce..6162477 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ColumnDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ColumnDesc.java
@@ -30,6 +30,7 @@ import java.io.Serializable;
* Column Metadata from Source. All name should be uppercase.
* <p/>
*/
+@SuppressWarnings("serial")
@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
public class ColumnDesc implements Serializable {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-metadata/src/main/java/org/apache/kylin/metadata/model/DatabaseDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/DatabaseDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/DatabaseDesc.java
index 215e86c..6b8447d 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/DatabaseDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/DatabaseDesc.java
@@ -27,6 +27,7 @@ import java.util.Set;
/**
* @author xjiang
*/
+@SuppressWarnings("serial")
public class DatabaseDesc implements Serializable {
private String name;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-metadata/src/main/java/org/apache/kylin/metadata/model/FunctionDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/FunctionDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/FunctionDesc.java
index d10f395..b8cefa2 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/FunctionDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/FunctionDesc.java
@@ -18,11 +18,13 @@
package org.apache.kylin.metadata.model;
+import java.util.ArrayList;
import java.util.Collection;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Lists;
/**
*/
@@ -36,9 +38,9 @@ public class FunctionDesc {
public static final String FUNC_COUNT_DISTINCT = "COUNT_DISTINCT";
public static final String FUNC_TOP_N = "TOP_N";
- public static final String PARAMTER_TYPE_CONSTANT = "constant";
+ public static final String PARAMETER_TYPE_CONSTANT = "constant";
public static final String PARAMETER_TYPE_COLUMN = "column";
-
+
@JsonProperty("expression")
private String expression;
@JsonProperty("parameter")
@@ -49,6 +51,26 @@ public class FunctionDesc {
private DataType returnDataType;
private boolean isDimensionAsMetric = false;
+ public void init(TableDesc factTable) {
+ expression = expression.toUpperCase();
+ returnDataType = DataType.getInstance(returnType);
+
+ for (ParameterDesc p = parameter; p != null; p = p.getNextParameter()) {
+ p.setValue(p.getValue().toUpperCase());
+ }
+
+ ArrayList<TblColRef> colRefs = Lists.newArrayList();
+ for (ParameterDesc p = parameter; p != null; p = p.getNextParameter()) {
+ if (p.isColumnType()) {
+ ColumnDesc sourceColumn = factTable.findColumnByName(p.getValue());
+ TblColRef colRef = new TblColRef(sourceColumn);
+ colRefs.add(colRef);
+ }
+ }
+
+ parameter.setColRefs(colRefs);
+ }
+
public String getRewriteFieldName() {
if (isSum()) {
return getParameter().getValue();
@@ -161,11 +183,6 @@ public class FunctionDesc {
public void setReturnType(String returnType) {
this.returnType = returnType;
- this.initReturnDataType();
- }
-
- // Jackson does not provide object post-processing currently
- public void initReturnDataType() {
this.returnDataType = DataType.getInstance(returnType);
}
@@ -225,13 +242,32 @@ public class FunctionDesc {
return "FunctionDesc [expression=" + expression + ", parameter=" + parameter + ", returnType=" + returnType + "]";
}
- public boolean isCompatible(FunctionDesc another) {
- if (another == null) {
+ // cols[0] is numeric (e.g. GMV), cols[1] is literal (e.g. SELLER)
+ public TblColRef getTopNNumericColumn() {
+ if (isTopN() == false)
+ throw new IllegalStateException();
+
+ return parameter.getColRefs().get(0);
+ }
+
+ // cols[0] is numeric (e.g. GMV), cols[1] is literal (e.g. SELLER)
+ public TblColRef getTopNLiteralColumn() {
+ if (isTopN() == false)
+ throw new IllegalStateException();
+
+ return parameter.getColRefs().get(1);
+ }
+
+ public boolean isTopNCompatibleSum(FunctionDesc sum) {
+ if (isTopN() == false)
+ throw new IllegalStateException();
+
+ if (sum == null) {
return false;
}
- if (this.isTopN() && another.isSum()) {
- if (this.getParameter().getColRefs().get(0).equals(another.getParameter().getColRefs().get(0)))
+ if (this.isTopN() && sum.isSum()) {
+ if (this.getParameter().getColRefs().get(0).equals(sum.getParameter().getColRefs().get(0)))
return true;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-metadata/src/main/java/org/apache/kylin/metadata/model/MeasureDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/MeasureDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/MeasureDesc.java
index 1561b1f..618d25a 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/MeasureDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/MeasureDesc.java
@@ -18,6 +18,9 @@
package org.apache.kylin.metadata.model;
+import java.util.Collections;
+import java.util.List;
+
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -37,6 +40,15 @@ public class MeasureDesc {
@JsonProperty("dependent_measure_ref")
private String dependentMeasureRef;
+ public List<TblColRef> getColumnsNeedDictionary() {
+ // measure could store literal values using dictionary encoding to save space, like TopN
+ if (function.isTopN()) {
+ return Collections.singletonList(function.getTopNLiteralColumn());
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
public int getId() {
return id;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
index 9773b84..2cf4374 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
@@ -19,11 +19,8 @@
package org.apache.kylin.metadata.model;
import java.io.UnsupportedEncodingException;
-import java.util.Arrays;
import java.util.List;
-import org.apache.commons.lang.StringUtils;
-
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -33,15 +30,13 @@ import com.fasterxml.jackson.annotation.JsonProperty;
@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
public class ParameterDesc {
- public static final String COLUMN_TYPE = "column";
-
@JsonProperty("type")
private String type;
@JsonProperty("value")
private String value;
- @JsonProperty("displaycolumn")
- private String displayColumn;
+ @JsonProperty("next_parameter")
+ private ParameterDesc nextParameter;
private List<TblColRef> colRefs;
@@ -65,14 +60,6 @@ public class ParameterDesc {
this.value = value;
}
- public String getDisplayColumn() {
- return displayColumn;
- }
-
- public void setDisplayColumn(String displayColumn) {
- this.displayColumn = displayColumn;
- }
-
public List<TblColRef> getColRefs() {
return colRefs;
}
@@ -80,19 +67,17 @@ public class ParameterDesc {
public void setColRefs(List<TblColRef> colRefs) {
this.colRefs = colRefs;
}
+
+ public ParameterDesc getNextParameter() {
+ return nextParameter;
+ }
- public boolean isColumnType() {
- return COLUMN_TYPE.equals(type);
+ public void setNextParameter(ParameterDesc nextParameter) {
+ this.nextParameter = nextParameter;
}
- public void normalizeColumnValue() {
- if (isColumnType()) {
- String values[] = value.split("\\s*,\\s*");
- for (int i = 0; i < values.length; i++)
- values[i] = values[i].toUpperCase();
- Arrays.sort(values);
- value = StringUtils.join(values, ",");
- }
+ public boolean isColumnType() {
+ return FunctionDesc.PARAMETER_TYPE_COLUMN.equals(type);
}
@Override
@@ -102,7 +87,7 @@ public class ParameterDesc {
ParameterDesc that = (ParameterDesc) o;
- if (displayColumn != null ? !displayColumn.equals(that.displayColumn) : that.displayColumn != null) return false;
+ if (nextParameter != null ? !nextParameter.equals(that.nextParameter) : that.nextParameter != null) return false;
if (type != null ? !type.equals(that.type) : that.type != null) return false;
if (value != null ? !value.equals(that.value) : that.value != null) return false;
@@ -113,13 +98,13 @@ public class ParameterDesc {
public int hashCode() {
int result = type != null ? type.hashCode() : 0;
result = 31 * result + (value != null ? value.hashCode() : 0);
- result = 31 * result + (displayColumn != null ? displayColumn.hashCode() : 0);
+ result = 31 * result + (nextParameter != null ? nextParameter.hashCode() : 0);
return result;
}
@Override
public String toString() {
- return "ParameterDesc [type=" + type + ", value=" + value + ", displayColumn=" + displayColumn + "]";
+ return "ParameterDesc [type=" + type + ", value=" + value + ", nextParam=" + nextParameter + "]";
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/core-storage/src/test/java/org/apache/kylin/storage/cache/StorageMockUtils.java
----------------------------------------------------------------------
diff --git a/core-storage/src/test/java/org/apache/kylin/storage/cache/StorageMockUtils.java b/core-storage/src/test/java/org/apache/kylin/storage/cache/StorageMockUtils.java
index 2b5ceee..2898f93 100644
--- a/core-storage/src/test/java/org/apache/kylin/storage/cache/StorageMockUtils.java
+++ b/core-storage/src/test/java/org/apache/kylin/storage/cache/StorageMockUtils.java
@@ -117,7 +117,6 @@ public class StorageMockUtils {
return compareFilter;
}
- @SuppressWarnings("unused")
public static TupleFilter buildAndFilter(List<TblColRef> columns) {
CompareTupleFilter compareFilter1 = buildFilter1(columns.get(0));
CompareTupleFilter compareFilter2 = buildFilter2(columns.get(1));
@@ -127,7 +126,6 @@ public class StorageMockUtils {
return andFilter;
}
- @SuppressWarnings("unused")
public static TupleFilter buildOrFilter(List<TblColRef> columns) {
CompareTupleFilter compareFilter1 = buildFilter1(columns.get(0));
CompareTupleFilter compareFilter2 = buildFilter2(columns.get(1));
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIDesc.java
----------------------------------------------------------------------
diff --git a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIDesc.java b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIDesc.java
index 71737dc..452e3a3 100644
--- a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIDesc.java
+++ b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIDesc.java
@@ -35,7 +35,16 @@ import org.apache.kylin.common.util.JsonUtil;
import org.apache.kylin.common.util.StringUtil;
import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.metadata.MetadataManager;
-import org.apache.kylin.metadata.model.*;
+import org.apache.kylin.metadata.model.ColumnDesc;
+import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.metadata.model.DimensionDesc;
+import org.apache.kylin.metadata.model.FunctionDesc;
+import org.apache.kylin.metadata.model.IEngineAware;
+import org.apache.kylin.metadata.model.IStorageAware;
+import org.apache.kylin.metadata.model.MeasureDesc;
+import org.apache.kylin.metadata.model.ParameterDesc;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.model.TblColRef;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
@@ -49,6 +58,7 @@ import com.google.common.collect.Sets;
/**
* @author yangli9
*/
+@SuppressWarnings("serial")
@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
public class IIDesc extends RootPersistentEntity {
@@ -231,7 +241,7 @@ public class IIDesc extends RootPersistentEntity {
/**
*
- * @param hllType represents the presision
+ * @param hllType represents the precision
*/
private MeasureDesc makeHLLMeasure(ColumnDesc columnDesc, String hllType) {
String columnName = columnDesc.getName();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
----------------------------------------------------------------------
diff --git a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
index 7e54a98..e17133f 100644
--- a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
+++ b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
@@ -19,7 +19,6 @@
package org.apache.kylin.invertedindex.model;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java
index e950911..cbc0c56 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPAggregateRel.java
@@ -210,7 +210,7 @@ public class OLAPAggregateRel extends Aggregate implements OLAPRel {
if (!column.isInnerColumn()) {
parameter = new ParameterDesc();
parameter.setValue(column.getName());
- parameter.setType("column");
+ parameter.setType(FunctionDesc.PARAMETER_TYPE_COLUMN);
parameter.setColRefs(Arrays.asList(column));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
index f84e4e6..cf4fd46 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
@@ -87,32 +87,20 @@ public class CubeStorageQuery implements ICachableStorageQuery {
private final CubeInstance cubeInstance;
private final CubeDesc cubeDesc;
private final String uuid;
- private Collection<TblColRef> topNColumns;
public CubeStorageQuery(CubeInstance cube) {
this.cubeInstance = cube;
this.cubeDesc = cube.getDescriptor();
this.uuid = cube.getUuid();
- this.topNColumns = Lists.newArrayList();
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
- if (measureDesc.getFunction().isTopN()) {
- List<TblColRef> colRefs = measureDesc.getFunction().getParameter().getColRefs();
- topNColumns.add(colRefs.get(colRefs.size() - 1));
- }
- }
}
@Override
public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- // check whether this is a TopN query;
- checkAndRewriteTopN(context, sqlDigest, returnTupleInfo);
+ // check whether this is a TopN query
+ checkAndRewriteTopN(sqlDigest);
Collection<TblColRef> groups = sqlDigest.groupbyColumns;
- TblColRef topNCol = extractTopNCol(groups);
- if (topNCol != null)
- groups.remove(topNCol);
-
TupleFilter filter = sqlDigest.filter;
// build dimension & metrics
@@ -196,11 +184,6 @@ public class CubeStorageQuery implements ICachableStorageQuery {
continue;
}
- // skip topN display col
- if (topNColumns.contains(column)) {
- continue;
- }
-
dimensions.add(column);
}
}
@@ -767,48 +750,33 @@ public class CubeStorageQuery implements ICachableStorageQuery {
ObserverEnabler.enableCoprocessorIfBeneficial(cubeInstance, groupsCopD, valueDecoders, context);
}
- private void checkAndRewriteTopN(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- Collection<TblColRef> groups = sqlDigest.groupbyColumns;
- TblColRef topNDisplayCol = extractTopNCol(groups);
- boolean hasTopN = topNDisplayCol != null;
-
- if (hasTopN == false)
+ private void checkAndRewriteTopN(SQLDigest sqlDigest) {
+ FunctionDesc topnFunc = null;
+ TblColRef topnLiteralCol = null;
+ for (MeasureDesc measure : cubeDesc.getMeasures()) {
+ FunctionDesc func = measure.getFunction();
+ if (func.isTopN() && sqlDigest.groupbyColumns.contains(func.getTopNLiteralColumn())) {
+ topnFunc = func;
+ topnLiteralCol = func.getTopNLiteralColumn();
+ }
+ }
+
+ // if TopN is not involved
+ if (topnFunc == null)
return;
if (sqlDigest.aggregations.size() != 1) {
throw new IllegalStateException("When query with topN, only one metrics is allowed.");
}
- FunctionDesc functionDesc = sqlDigest.aggregations.iterator().next();
- if (functionDesc.isSum() == false) {
+ FunctionDesc origFunc = sqlDigest.aggregations.iterator().next();
+ if (origFunc.isSum() == false) {
throw new IllegalStateException("When query with topN, only SUM function is allowed.");
}
- FunctionDesc rewriteFunction = null;
- // replace the SUM to the TopN function
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
- if (measureDesc.getFunction().isCompatible(functionDesc) && topNDisplayCol.getName().equalsIgnoreCase(measureDesc.getFunction().getParameter().getDisplayColumn())) {
- rewriteFunction = measureDesc.getFunction();
- break;
- }
- }
-
- if (rewriteFunction == null) {
- throw new IllegalStateException("Didn't find topN measure for " + functionDesc);
- }
-
- sqlDigest.aggregations = Lists.newArrayList(rewriteFunction);
- logger.info("Rewrite function " + functionDesc + " to " + rewriteFunction);
+ sqlDigest.aggregations = Lists.newArrayList(topnFunc);
+ sqlDigest.groupbyColumns.remove(topnLiteralCol);
+ sqlDigest.metricColumns.add(topnLiteralCol);
+ logger.info("Rewrite function " + origFunc + " to " + topnFunc);
}
-
- private TblColRef extractTopNCol(Collection<TblColRef> colRefs) {
- for (TblColRef colRef : colRefs) {
- if (topNColumns.contains(colRef)) {
- return colRef;
- }
- }
-
- return null;
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
index 4ced852..258e20e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
@@ -46,30 +46,18 @@ public class CubeStorageQuery implements ICachableStorageQuery {
private final CubeInstance cubeInstance;
private final CubeDesc cubeDesc;
- private Collection<TblColRef> topNColumns;
public CubeStorageQuery(CubeInstance cube) {
this.cubeInstance = cube;
this.cubeDesc = cube.getDescriptor();
- this.topNColumns = Lists.newArrayList();
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
- if (measureDesc.getFunction().isTopN()) {
- List<TblColRef> colRefs = measureDesc.getFunction().getParameter().getColRefs();
- topNColumns.add(colRefs.get(colRefs.size() - 1));
- }
- }
}
@Override
public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- // check whether this is a TopN query;
- checkAndRewriteTopN(context, sqlDigest, returnTupleInfo);
+ // check whether this is a TopN query
+ checkAndRewriteTopN(sqlDigest);
Collection<TblColRef> groups = sqlDigest.groupbyColumns;
- TblColRef topNCol = extractTopNCol(groups);
- if (topNCol != null)
- groups.remove(topNCol);
-
TupleFilter filter = sqlDigest.filter;
// build dimension & metrics
@@ -145,10 +133,6 @@ public class CubeStorageQuery implements ICachableStorageQuery {
continue;
}
- // skip topN display col
- if (topNColumns.contains(column)) {
- continue;
- }
dimensions.add(column);
}
}
@@ -399,47 +383,33 @@ public class CubeStorageQuery implements ICachableStorageQuery {
}
- private void checkAndRewriteTopN(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- Collection<TblColRef> groups = sqlDigest.groupbyColumns;
- TblColRef topNDisplayCol = extractTopNCol(groups);
- boolean hasTopN = topNDisplayCol != null;
-
- if (hasTopN == false)
+ private void checkAndRewriteTopN(SQLDigest sqlDigest) {
+ FunctionDesc topnFunc = null;
+ TblColRef topnLiteralCol = null;
+ for (MeasureDesc measure : cubeDesc.getMeasures()) {
+ FunctionDesc func = measure.getFunction();
+ if (func.isTopN() && sqlDigest.groupbyColumns.contains(func.getTopNLiteralColumn())) {
+ topnFunc = func;
+ topnLiteralCol = func.getTopNLiteralColumn();
+ }
+ }
+
+ // if TopN is not involved
+ if (topnFunc == null)
return;
if (sqlDigest.aggregations.size() != 1) {
throw new IllegalStateException("When query with topN, only one metrics is allowed.");
}
- FunctionDesc functionDesc = sqlDigest.aggregations.iterator().next();
- if (functionDesc.isSum() == false) {
+ FunctionDesc origFunc = sqlDigest.aggregations.iterator().next();
+ if (origFunc.isSum() == false) {
throw new IllegalStateException("When query with topN, only SUM function is allowed.");
}
- FunctionDesc rewriteFunction = null;
- // replace the SUM to the TopN function
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
- if (measureDesc.getFunction().isCompatible(functionDesc) && topNDisplayCol.getName().equalsIgnoreCase(measureDesc.getFunction().getParameter().getDisplayColumn())) {
- rewriteFunction = measureDesc.getFunction();
- break;
- }
- }
-
- if (rewriteFunction == null) {
- throw new IllegalStateException("Didn't find topN measure for " + functionDesc);
- }
-
- sqlDigest.aggregations = Lists.newArrayList(rewriteFunction);
- logger.info("Rewrite function " + functionDesc + " to " + rewriteFunction);
- }
-
- private TblColRef extractTopNCol(Collection<TblColRef> colRefs) {
- for (TblColRef colRef : colRefs) {
- if (topNColumns.contains(colRef)) {
- return colRef;
- }
- }
-
- return null;
+ sqlDigest.aggregations = Lists.newArrayList(topnFunc);
+ sqlDigest.groupbyColumns.remove(topnLiteralCol);
+ sqlDigest.metricColumns.add(topnLiteralCol);
+ logger.info("Rewrite function " + origFunc + " to " + topnFunc);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ba484d9e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/coprocessor/endpoint/EndpointTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/coprocessor/endpoint/EndpointTupleIterator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/coprocessor/endpoint/EndpointTupleIterator.java
index 99db123..2fd0b4f 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/coprocessor/endpoint/EndpointTupleIterator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/coprocessor/endpoint/EndpointTupleIterator.java
@@ -29,7 +29,6 @@ import java.util.Map;
import javax.annotation.Nullable;
-import com.google.protobuf.ByteString;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.hadoop.hbase.client.HConnection;