You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2016/10/16 05:29:31 UTC
[1/5] kylin git commit: KYLIN-2030 bug fix
Repository: kylin
Updated Branches:
refs/heads/orderedbytes3 [created] 7bce09614
KYLIN-2030 bug fix
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/7bce0961
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/7bce0961
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/7bce0961
Branch: refs/heads/orderedbytes3
Commit: 7bce09614abfeea7c567c047929a9c985f5094d0
Parents: ae73be5
Author: Hongbin Ma <ma...@apache.org>
Authored: Sun Oct 9 19:07:52 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Sun Oct 16 13:27:26 2016 +0800
----------------------------------------------------------------------
.../kylin/cube/CubeCapabilityChecker.java | 23 ++++++++++----------
.../kylin/query/relnode/OLAPAggregateRel.java | 22 +++++++++++--------
2 files changed, 25 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/7bce0961/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 ee21b1c..e509d98 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
@@ -81,7 +81,7 @@ public class CubeCapabilityChecker {
//1. dimension as measure
if (!unmatchedAggregations.isEmpty()) {
- tryDimensionAsMeasures(unmatchedAggregations, digest, cube, result, cube.getDescriptor().listDimensionColumnsIncludingDerived());
+ tryDimensionAsMeasures(unmatchedAggregations, result, cube.getDescriptor().listDimensionColumnsIncludingDerived());
}
} else {
//for non query-on-facttable
@@ -92,10 +92,18 @@ public class CubeCapabilityChecker {
dimCols.add(columnDesc.getRef());
}
- //1. dimension as measure, like max(cal_dt) or count( distinct col) from lookup
+ //1. all aggregations on lookup table can be done. For distinct count, mark them all DimensionAsMeasures
+ // so that the measure has a chance to be upgraded to DimCountDistinctMeasureType in org.apache.kylin.metadata.model.FunctionDesc#reInitMeasureType
if (!unmatchedAggregations.isEmpty()) {
- tryDimensionAsMeasures(unmatchedAggregations, digest, cube, result, dimCols);
+ Iterator<FunctionDesc> itr = unmatchedAggregations.iterator();
+ while (itr.hasNext()) {
+ FunctionDesc functionDesc = itr.next();
+ if (dimCols.containsAll(functionDesc.getParameter().getColRefs())) {
+ itr.remove();
+ }
+ }
}
+ tryDimensionAsMeasures(Lists.newArrayList(aggrFunctions), result, dimCols);
//2. more "dimensions" contributed by snapshot
if (!unmatchedDimensions.isEmpty()) {
@@ -159,19 +167,12 @@ public class CubeCapabilityChecker {
return result;
}
- private static void tryDimensionAsMeasures(Collection<FunctionDesc> unmatchedAggregations, SQLDigest digest, CubeInstance cube, CapabilityResult result, Set<TblColRef> dimCols) {
- CubeDesc cubeDesc = cube.getDescriptor();
- Collection<FunctionDesc> cubeFuncs = cubeDesc.listAllFunctions();
+ private static void tryDimensionAsMeasures(Collection<FunctionDesc> unmatchedAggregations, CapabilityResult result, Set<TblColRef> dimCols) {
Iterator<FunctionDesc> it = unmatchedAggregations.iterator();
while (it.hasNext()) {
FunctionDesc functionDesc = it.next();
- if (cubeFuncs.contains(functionDesc)) {
- it.remove();
- continue;
- }
-
// let calcite handle count
if (functionDesc.isCount()) {
it.remove();
http://git-wip-us.apache.org/repos/asf/kylin/blob/7bce0961/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 97efb27..c7a1eff 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
@@ -285,16 +285,20 @@ public class OLAPAggregateRel extends Aggregate implements OLAPRel {
}
private void translateAggregation() {
- // now the realization is known, replace aggregations with what's defined on MeasureDesc
- List<MeasureDesc> measures = this.context.realization.getMeasures();
- List<FunctionDesc> newAggrs = Lists.newArrayList();
- for (FunctionDesc aggFunc : this.aggregations) {
- newAggrs.add(findInMeasures(aggFunc, measures));
+ if (!noPrecaculatedFieldsAvailable()) {
+ // now the realization is known, replace aggregations with what's defined on MeasureDesc
+ List<MeasureDesc> measures = this.context.realization.getMeasures();
+ List<FunctionDesc> newAggrs = Lists.newArrayList();
+ for (FunctionDesc aggFunc : this.aggregations) {
+ newAggrs.add(findInMeasures(aggFunc, measures));
+ }
+ this.aggregations.clear();
+ this.aggregations.addAll(newAggrs);
+ this.context.aggregations.clear();
+ this.context.aggregations.addAll(newAggrs);
+ } else {
+ //the realization is not contributing pre-calculated fields at all
}
- this.aggregations.clear();
- this.aggregations.addAll(newAggrs);
- this.context.aggregations.clear();
- this.context.aggregations.addAll(newAggrs);
}
private FunctionDesc findInMeasures(FunctionDesc aggFunc, List<MeasureDesc> measures) {
[2/5] kylin git commit: KYLIN-1726 fix BuildCubeWithStream
Posted by ma...@apache.org.
KYLIN-1726 fix BuildCubeWithStream
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/8f1fb6ac
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/8f1fb6ac
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/8f1fb6ac
Branch: refs/heads/orderedbytes3
Commit: 8f1fb6ac47b9fcf22090080b86d2ea98cc9d6132
Parents: 3d0954d
Author: Hongbin Ma <ma...@apache.org>
Authored: Sat Oct 8 14:08:41 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Sun Oct 16 13:27:26 2016 +0800
----------------------------------------------------------------------
.../java/org/apache/kylin/provision/BuildCubeWithStream.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/8f1fb6ac/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
index fed14d8..c2fb180 100644
--- a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
@@ -18,6 +18,8 @@
package org.apache.kylin.provision;
+import static java.lang.Thread.sleep;
+
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
@@ -32,7 +34,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.Lists;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.requests.MetadataResponse;
@@ -64,7 +65,7 @@ import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static java.lang.Thread.sleep;
+import com.google.common.collect.Lists;
/**
* for streaming cubing case "test_streaming_table"
@@ -323,6 +324,5 @@ public class BuildCubeWithStream {
logger.error("error", e);
System.exit(1);
}
-
}
}
[3/5] kylin git commit: invoke after() in buildcubewithstream
Posted by ma...@apache.org.
invoke after() in buildcubewithstream
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/c60c2d62
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/c60c2d62
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/c60c2d62
Branch: refs/heads/orderedbytes3
Commit: c60c2d628ed0a3a4baaaa852129727aa8f42d77c
Parents: 61a08d4
Author: Hongbin Ma <ma...@apache.org>
Authored: Sun Oct 16 13:25:33 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Sun Oct 16 13:27:26 2016 +0800
----------------------------------------------------------------------
.../test/java/org/apache/kylin/provision/BuildCubeWithStream.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/c60c2d62/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
index 2faa8d0..fed14d8 100644
--- a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
@@ -316,6 +316,7 @@ public class BuildCubeWithStream {
buildCubeWithStream.build();
logger.info("Build is done");
buildCubeWithStream.cleanup();
+ buildCubeWithStream.after();
logger.info("Going to exit");
System.exit(0);
} catch (Throwable e) {
[5/5] kylin git commit: improve integer type upgrade logic
Posted by ma...@apache.org.
improve integer type upgrade logic
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/3d0954d1
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/3d0954d1
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/3d0954d1
Branch: refs/heads/orderedbytes3
Commit: 3d0954d1e902755ce6f33719c4e74a368ee537d4
Parents: c60c2d6
Author: Hongbin Ma <ma...@apache.org>
Authored: Fri Sep 30 18:31:04 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Sun Oct 16 13:27:26 2016 +0800
----------------------------------------------------------------------
.../filter/EvaluatableFunctionTupleFilter.java | 151 -------------------
.../metadata/filter/TupleFilterSerializer.java | 20 ++-
.../apache/kylin/metadata/model/ColumnDesc.java | 21 ++-
.../apache/kylin/metadata/tuple/TupleInfo.java | 4 +-
.../apache/kylin/query/schema/OLAPTable.java | 31 +++-
5 files changed, 63 insertions(+), 164 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/3d0954d1/core-metadata/src/main/java/org/apache/kylin/metadata/filter/EvaluatableFunctionTupleFilter.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/EvaluatableFunctionTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/EvaluatableFunctionTupleFilter.java
deleted file mode 100644
index ff24172..0000000
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/EvaluatableFunctionTupleFilter.java
+++ /dev/null
@@ -1,151 +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.metadata.filter;
-
-import java.lang.reflect.InvocationTargetException;
-import java.nio.ByteBuffer;
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.kylin.common.util.ByteArray;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.metadata.datatype.DataType;
-import org.apache.kylin.metadata.datatype.StringSerializer;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
-
-import com.google.common.collect.Lists;
-
-public class EvaluatableFunctionTupleFilter extends BuiltInFunctionTupleFilter {
-
- private boolean constantsInitted = false;
-
- //about non-like
- private List<Object> values;
- private Object tupleValue;
-
- public EvaluatableFunctionTupleFilter(String name) {
- super(name, FilterOperatorEnum.EVAL_FUNC);
- values = Lists.newArrayListWithCapacity(1);
- values.add(null);
- }
-
- @Override
- public boolean evaluate(IEvaluatableTuple tuple, IFilterCodeSystem cs) {
-
- // extract tuple value
- Object tupleValue = null;
- for (TupleFilter filter : this.children) {
- if (!isConstant(filter)) {
- filter.evaluate(tuple, cs);
- tupleValue = filter.getValues().iterator().next();
- }
- }
-
- TblColRef tblColRef = this.getColumn();
- DataType strDataType = DataType.getType("string");
- if (tblColRef.getType() != strDataType) {
- throw new IllegalStateException("Only String type is allow in BuiltInFunction");
- }
- ByteArray valueByteArray = (ByteArray) tupleValue;
- StringSerializer serializer = new StringSerializer(strDataType);
- String value = serializer.deserialize(ByteBuffer.wrap(valueByteArray.array(), valueByteArray.offset(), valueByteArray.length()));
-
- try {
- if (isLikeFunction()) {
- return (Boolean) invokeFunction(value);
- } else {
- this.tupleValue = invokeFunction(value);
- //convert back to ByteArray format because the outer EvaluatableFunctionTupleFilter assumes input as ByteArray
- ByteBuffer buffer = ByteBuffer.allocate(valueByteArray.length() * 2);
- serializer.serialize((String) this.tupleValue, buffer);
- this.tupleValue = new ByteArray(buffer.array(), 0, buffer.position());
-
- return true;
- }
- } catch (InvocationTargetException | IllegalAccessException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public Collection<?> getValues() {
- this.values.set(0, tupleValue);
- return values;
- }
-
- @Override
- public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
- if (!isValid()) {
- throw new IllegalStateException("must be valid");
- }
- BytesUtil.writeUTFString(name, buffer);
- }
-
- @Override
- public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
- this.name = BytesUtil.readUTFString(buffer);
- this.initMethod();
- }
-
- @Override
- public boolean isEvaluable() {
- return true;
- }
-
- private boolean isConstant(TupleFilter filter) {
- return (filter instanceof ConstantTupleFilter) || (filter instanceof DynamicTupleFilter);
- }
-
- @Override
- public Object invokeFunction(Object input) throws InvocationTargetException, IllegalAccessException {
- if (isLikeFunction())
- initConstants();
- return super.invokeFunction(input);
- }
-
- private void initConstants() {
- if (constantsInitted) {
- return;
- }
- //will replace the ByteArray pattern to String type
- ByteArray byteArray = (ByteArray) methodParams.get(constantPosition);
- StringSerializer s = new StringSerializer(DataType.getType("string"));
- String pattern = s.deserialize(ByteBuffer.wrap(byteArray.array(), byteArray.offset(), byteArray.length()));
- //TODO
- //pattern = pattern.toLowerCase();//to remove upper case
- methodParams.set(constantPosition, pattern);
- constantsInitted = true;
- }
-
- //even for "tolower(s)/toupper(s)/substring(like) like pattern", the like pattern can be used for index searching
- public String getLikePattern() {
- if (!isLikeFunction()) {
- return null;
- }
-
- initConstants();
- return (String) methodParams.get(1);
- }
-
- public boolean isLikeFunction() {
- return "like".equalsIgnoreCase(this.getName());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/3d0954d1/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java
index a051ea9..2df474e 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java
@@ -18,6 +18,7 @@
package org.apache.kylin.metadata.filter;
+import java.lang.reflect.InvocationTargetException;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.util.HashMap;
@@ -29,6 +30,8 @@ import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Maps;
+
/**
* http://eli.thegreenplace.net/2011/09/29/an-interesting-tree-serialization-algorithm-from-dwarf
*
@@ -45,6 +48,7 @@ public class TupleFilterSerializer {
private static final int BUFFER_SIZE = 65536;
private static final Map<Integer, TupleFilter.FilterOperatorEnum> ID_OP_MAP = new HashMap<Integer, TupleFilter.FilterOperatorEnum>();
+ protected static final Map<TupleFilter.FilterOperatorEnum, Class> extendedTupleFilters = Maps.newHashMap();
static {
for (TupleFilter.FilterOperatorEnum op : TupleFilter.FilterOperatorEnum.values()) {
@@ -191,14 +195,20 @@ public class TupleFilterSerializer {
case UNSUPPORTED:
filter = new UnsupportedTupleFilter(op);
break;
- case EVAL_FUNC:
- filter = new EvaluatableFunctionTupleFilter(null);
- break;
case MASSIN:
filter = new MassInTupleFilter();
break;
- default:
- throw new IllegalStateException("Error FilterOperatorEnum: " + op.getValue());
+ default: {
+ if (extendedTupleFilters.containsKey(op)) {
+ try {
+ filter = (TupleFilter) extendedTupleFilters.get(op).getConstructor().newInstance();
+ } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new IllegalStateException("Error FilterOperatorEnum: " + op.getValue());
+ }
+ }
}
return filter;
http://git-wip-us.apache.org/repos/asf/kylin/blob/3d0954d1/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 d131b92..c93e630 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
@@ -25,8 +25,8 @@ import org.apache.commons.lang.StringUtils;
import org.apache.kylin.metadata.datatype.DataType;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
/**
* Column Metadata from Source. All name should be uppercase.
@@ -58,6 +58,7 @@ public class ColumnDesc implements Serializable {
// parsed from data type
private DataType type;
+ private DataType upgradedType;
private TableDesc table;
private int zeroBasedIndex = -1;
@@ -89,6 +90,18 @@ public class ColumnDesc implements Serializable {
type = DataType.getType(datatype);
}
+ public void setUpgradedType(String datatype) {
+ this.upgradedType = DataType.getType(datatype);
+ }
+
+ public DataType getUpgradedType() {
+ if (this.upgradedType == null) {
+ return this.type;
+ } else {
+ return this.upgradedType;
+ }
+ }
+
public String getId() {
return id;
}
@@ -185,19 +198,19 @@ public class ColumnDesc implements Serializable {
if (getClass() != obj.getClass())
return false;
ColumnDesc other = (ColumnDesc) obj;
-
+
if (name == null) {
if (other.name != null)
return false;
} else if (!name.equals(other.name))
return false;
-
+
if (datatype == null) {
if (other.datatype != null)
return false;
} else if (!datatype.equals(other.datatype))
return false;
-
+
return true;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/3d0954d1/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TupleInfo.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TupleInfo.java b/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TupleInfo.java
index f7c3b57..8970124 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TupleInfo.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TupleInfo.java
@@ -94,9 +94,9 @@ public class TupleInfo {
columns.add(index, col);
if (dataTypeNames.size() > index)
- dataTypeNames.set(index, col.getType().getName());
+ dataTypeNames.set(index, col.getColumnDesc().getUpgradedType().getName());
else
- dataTypeNames.add(index, col.getType().getName());
+ dataTypeNames.add(index, col.getColumnDesc().getUpgradedType().getName());
}
public List<String> getAllFields() {
http://git-wip-us.apache.org/repos/asf/kylin/blob/3d0954d1/query/src/main/java/org/apache/kylin/query/schema/OLAPTable.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/schema/OLAPTable.java b/query/src/main/java/org/apache/kylin/query/schema/OLAPTable.java
index b9d2fd0..7d3b2bd 100644
--- a/query/src/main/java/org/apache/kylin/query/schema/OLAPTable.java
+++ b/query/src/main/java/org/apache/kylin/query/schema/OLAPTable.java
@@ -43,15 +43,21 @@ import org.apache.calcite.schema.impl.AbstractTableQueryable;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.type.SqlTypeUtil;
import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.datatype.DataType;
import org.apache.kylin.metadata.model.ColumnDesc;
+import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.project.ProjectManager;
+import org.apache.kylin.metadata.realization.IRealization;
+import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.query.enumerator.OLAPQuery;
import org.apache.kylin.query.enumerator.OLAPQuery.EnumeratorTypeEnum;
import org.apache.kylin.query.relnode.OLAPTableScan;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
@@ -61,6 +67,8 @@ import com.google.common.collect.Sets;
*/
public class OLAPTable extends AbstractQueryableTable implements TranslatableTable {
+ protected static final Logger logger = LoggerFactory.getLogger(OLAPTable.class);
+
private static Map<String, SqlTypeName> SQLTYPE_MAPPING = new HashMap<String, SqlTypeName>();
static {
@@ -129,7 +137,7 @@ public class OLAPTable extends AbstractQueryableTable implements TranslatableTab
private RelDataType deriveRowType(RelDataTypeFactory typeFactory) {
RelDataTypeFactory.FieldInfoBuilder fieldInfo = typeFactory.builder();
for (ColumnDesc column : exposedColumns) {
- RelDataType sqlType = createSqlType(typeFactory, column.getType(), column.isNullable());
+ RelDataType sqlType = createSqlType(typeFactory, column.getUpgradedType(), column.isNullable());
sqlType = SqlTypeUtil.addCharsetAndCollation(sqlType, typeFactory);
fieldInfo.add(column.getName(), sqlType);
}
@@ -182,6 +190,8 @@ public class OLAPTable extends AbstractQueryableTable implements TranslatableTab
//if exist sum(x), where x is integer/short/byte
//to avoid overflow we upgrade x's type to long
+ //this includes checking two parts:
+ //1. sum measures in cubes:
HashSet<ColumnDesc> updateColumns = Sets.newHashSet();
for (MeasureDesc m : mgr.listEffectiveMeasures(olapSchema.getProjectName(), sourceTable.getIdentity())) {
if (m.getFunction().isSum()) {
@@ -193,9 +203,26 @@ public class OLAPTable extends AbstractQueryableTable implements TranslatableTab
}
}
}
+ //2. All integer measures in non-cube realizations
+ MetadataManager metadataManager = MetadataManager.getInstance(olapSchema.getConfig());
+ for (IRealization realization : mgr.listAllRealizations(olapSchema.getProjectName())) {
+ if (realization.getType() == RealizationType.INVERTED_INDEX && sourceTable.getIdentity().equalsIgnoreCase(realization.getFactTable())) {
+ DataModelDesc dataModelDesc = realization.getDataModelDesc();
+ for (String metricColumn : dataModelDesc.getMetrics()) {
+ ColumnDesc columnDesc = metadataManager.getColumnDesc(dataModelDesc.getFactTable() + "." + metricColumn);
+ if (columnDesc.getType().isIntegerFamily() && !columnDesc.getType().isBigInt())
+ updateColumns.add(columnDesc);
+ }
+ }
+ }
+
for (ColumnDesc upgrade : updateColumns) {
int index = tableColumns.indexOf(upgrade);
- tableColumns.get(index).setDatatype("bigint");
+ if (index < 0) {
+ throw new IllegalStateException("Metric column " + upgrade + " is not found in the the project's columns");
+ }
+ tableColumns.get(index).setUpgradedType("bigint");
+ logger.info("To avoid overflow, upgraded {}'s type from {} to {}", tableColumns.get(index), tableColumns.get(index).getType(), tableColumns.get(index).getUpgradedType());
}
return Lists.newArrayList(Iterables.concat(tableColumns, metricColumns));
[4/5] kylin git commit: KYLIN-2030 enhancement
Posted by ma...@apache.org.
KYLIN-2030 enhancement
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/ae73be52
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/ae73be52
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/ae73be52
Branch: refs/heads/orderedbytes3
Commit: ae73be5204452face5f28a6b5799de2e22b0a198
Parents: 8f1fb6a
Author: Hongbin Ma <ma...@apache.org>
Authored: Sat Oct 8 16:14:26 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Sun Oct 16 13:27:26 2016 +0800
----------------------------------------------------------------------
.../kylin/cube/CubeCapabilityChecker.java | 38 +++++++++++++-------
.../org/apache/kylin/cube/model/CubeDesc.java | 10 +++---
.../kylin/metadata/model/FunctionDesc.java | 1 +
.../test/resources/query/sql_lookup/query06.sql | 1 +
.../test/resources/query/sql_lookup/query07.sql | 1 +
5 files changed, 32 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/ae73be52/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 1eada16..ee21b1c 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
@@ -75,22 +75,34 @@ public class CubeCapabilityChecker {
tryCustomMeasureTypes(unmatchedDimensions, unmatchedAggregations, digest, cube, result);
// }
- // try dimension-as-measure
- if (!unmatchedAggregations.isEmpty()) {
- if (cube.getDescriptor().getFactTable().equals(digest.factTable)) {
+ //more tricks
+ if (cube.getDescriptor().getFactTable().equals(digest.factTable)) {
+ //for query-on-facttable
+ //1. dimension as measure
+
+ if (!unmatchedAggregations.isEmpty()) {
tryDimensionAsMeasures(unmatchedAggregations, digest, cube, result, cube.getDescriptor().listDimensionColumnsIncludingDerived());
- } else {
- //deal with query on lookup table, like https://issues.apache.org/jira/browse/KYLIN-2030
- if (cube.getSegments().get(0).getSnapshots().containsKey(digest.factTable)) {
- TableDesc tableDesc = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()).getTableDesc(digest.factTable);
- Set<TblColRef> dimCols = Sets.newHashSet();
- for (ColumnDesc columnDesc : tableDesc.getColumns()) {
- dimCols.add(columnDesc.getRef());
- }
+ }
+ } else {
+ //for non query-on-facttable
+ if (cube.getSegments().get(0).getSnapshots().containsKey(digest.factTable)) {
+ TableDesc tableDesc = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()).getTableDesc(digest.factTable);
+ Set<TblColRef> dimCols = Sets.newHashSet();
+ for (ColumnDesc columnDesc : tableDesc.getColumns()) {
+ dimCols.add(columnDesc.getRef());
+ }
+
+ //1. dimension as measure, like max(cal_dt) or count( distinct col) from lookup
+ if (!unmatchedAggregations.isEmpty()) {
tryDimensionAsMeasures(unmatchedAggregations, digest, cube, result, dimCols);
- } else {
- logger.info("Skip tryDimensionAsMeasures because current cube {} does not touch lookup table {} at all", cube.getName(), digest.factTable);
}
+
+ //2. more "dimensions" contributed by snapshot
+ if (!unmatchedDimensions.isEmpty()) {
+ unmatchedDimensions.removeAll(dimCols);
+ }
+ } else {
+ logger.info("cube {} does not touch lookup table {} at all", cube.getName(), digest.factTable);
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/ae73be52/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 4195451..7e055be 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
@@ -33,9 +33,9 @@ import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
+import java.util.Map.Entry;
import javax.annotation.Nullable;
@@ -68,9 +68,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
@@ -559,9 +559,7 @@ public class CubeDesc extends RootPersistentEntity implements IEngineAware {
// check all dimension columns are presented on rowkey
List<TblColRef> dimCols = listDimensionColumnsExcludingDerived(true);
- checkState(rowkey.getRowKeyColumns().length == dimCols.size(),
- "RowKey columns count (%d) doesn't match dimensions columns count (%d)",
- rowkey.getRowKeyColumns().length, dimCols.size());
+ checkState(rowkey.getRowKeyColumns().length == dimCols.size(), "RowKey columns count (%d) doesn't match dimensions columns count (%d)", rowkey.getRowKeyColumns().length, dimCols.size());
initDictionaryDesc();
}
@@ -679,7 +677,7 @@ public class CubeDesc extends RootPersistentEntity implements IEngineAware {
private boolean hasSingle(ArrayList<Set<String>> dimsList) {
boolean hasSingle = false;
for (Set<String> dims : dimsList) {
- if (dims.size() < 2)
+ if (dims.size() == 1)
hasSingle = true;
}
return hasSingle;
http://git-wip-us.apache.org/repos/asf/kylin/blob/ae73be52/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 36fff5c..27c7c69 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
@@ -120,6 +120,7 @@ public class FunctionDesc {
}
public MeasureType<?> getMeasureType() {
+ //like max(cal_dt)
if (isDimensionAsMetric && !isCountDistinct()) {
return null;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/ae73be52/kylin-it/src/test/resources/query/sql_lookup/query06.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_lookup/query06.sql b/kylin-it/src/test/resources/query/sql_lookup/query06.sql
new file mode 100644
index 0000000..e2b8469
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_lookup/query06.sql
@@ -0,0 +1 @@
+select max(cal_dt) as x from edw.test_cal_dt
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/ae73be52/kylin-it/src/test/resources/query/sql_lookup/query07.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_lookup/query07.sql b/kylin-it/src/test/resources/query/sql_lookup/query07.sql
new file mode 100644
index 0000000..6be3439
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_lookup/query07.sql
@@ -0,0 +1 @@
+select site_id,count(*) as y,count(DISTINCT site_name) as x from edw.test_sites group by site_id