You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kylin.apache.org by GitBox <gi...@apache.org> on 2018/06/14 09:28:22 UTC
[GitHub] shaofengshi closed pull request #138: Apache kylin 3359
shaofengshi closed pull request #138: Apache kylin 3359
URL: https://github.com/apache/kylin/pull/138
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/atopcalcite/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java b/atopcalcite/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
index fb5bb1976f..519a73b647 100644
--- a/atopcalcite/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
+++ b/atopcalcite/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
@@ -4996,7 +4996,7 @@ private void translateAgg(SqlCall call, SqlNode filter, SqlCall outerCall) {
// special case for COUNT(*): delete the *
if (operand instanceof SqlIdentifier) {
SqlIdentifier id = (SqlIdentifier) operand;
- if (id.isStar() || isSimpleCount(call)) { /* OVERRIDE POINT */
+ if (id.isStar()) {
assert call.operandCount() == 1;
assert args.isEmpty();
break;
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index a971f14136..8dbc42bf37 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -1227,6 +1227,12 @@ public boolean isConvertCreateTableToWith() {
return Lists.newArrayList(rules.split(","));
}
+ // check KYLIN-3358, need deploy coprocessor if enabled
+ // finally should be deprecated
+ public boolean isDynamicColumnEnabled() {
+ return Boolean.valueOf(getOptional("kylin.query.enable-dynamic-column", "false"));
+ }
+
//check KYLIN-1684, in most cases keep the default value
public boolean isSkippingEmptySegments() {
return Boolean.valueOf(getOptional("kylin.query.skip-empty-segments", "true"));
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/DecimalUtil.java b/core-common/src/main/java/org/apache/kylin/common/util/DecimalUtil.java
new file mode 100644
index 0000000000..08919606c2
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/util/DecimalUtil.java
@@ -0,0 +1,45 @@
+/*
+ * 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.common.util;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+public class DecimalUtil {
+
+ // Copy from org.apache.calcite.runtime.SqlFunctions.toBigDecimal
+ public static BigDecimal toBigDecimal(String s) {
+ return new BigDecimal(s.trim());
+ }
+
+ public static BigDecimal toBigDecimal(Number number) {
+ // There are some values of "long" that cannot be represented as "double".
+ // Not so "int". If it isn't a long, go straight to double.
+ return number instanceof BigDecimal ? (BigDecimal) number
+ : number instanceof BigInteger ? new BigDecimal((BigInteger) number)
+ : number instanceof Long ? new BigDecimal(number.longValue())
+ : new BigDecimal(number.doubleValue());
+ }
+
+ public static BigDecimal toBigDecimal(Object o) {
+ if (o == null)
+ return null;
+ return o instanceof Number ? toBigDecimal((Number) o) : toBigDecimal(o.toString());
+ }
+}
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 5dffd96458..40746d163f 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
@@ -29,6 +29,7 @@
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.measure.basic.BasicMeasureType;
import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.IStorageAware;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -155,10 +156,12 @@ public MeasureDesc getInvolvedMeasure() {
private static Collection<TblColRef> getDimensionColumns(SQLDigest sqlDigest) {
Collection<TblColRef> groupByColumns = sqlDigest.groupbyColumns;
Collection<TblColRef> filterColumns = sqlDigest.filterColumns;
+ Collection<TblColRef> rtDimColumns = sqlDigest.rtDimensionColumns;
Collection<TblColRef> dimensionColumns = new HashSet<TblColRef>();
dimensionColumns.addAll(groupByColumns);
dimensionColumns.addAll(filterColumns);
+ dimensionColumns.addAll(rtDimColumns);
return dimensionColumns;
}
@@ -171,8 +174,39 @@ public MeasureDesc getInvolvedMeasure() {
private static Set<FunctionDesc> unmatchedAggregations(Collection<FunctionDesc> aggregations, CubeInstance cube) {
HashSet<FunctionDesc> result = Sets.newHashSet(aggregations);
+
CubeDesc cubeDesc = cube.getDescriptor();
- result.removeAll(cubeDesc.listAllFunctions());
+ List<FunctionDesc> definedFuncs = cubeDesc.listAllFunctions();
+
+ // check normal aggregations
+ result.removeAll(definedFuncs);
+
+ // check dynamic aggregations
+ Iterator<FunctionDesc> funcIterator = result.iterator();
+ while (funcIterator.hasNext()) {
+ FunctionDesc entry = funcIterator.next();
+ if (entry instanceof DynamicFunctionDesc) {
+ DynamicFunctionDesc dynFunc = (DynamicFunctionDesc) entry;
+ // Filter columns cannot be derived
+ Collection<TblColRef> definedCols = dynFunc.ifFriendlyForDerivedFilter()
+ ? cubeDesc.listDimensionColumnsIncludingDerived()
+ : cubeDesc.listDimensionColumnsExcludingDerived(true);
+ Set<TblColRef> filterCols = Sets.newHashSet(dynFunc.getFilterColumnSet());
+ filterCols.removeAll(definedCols);
+ if (!filterCols.isEmpty()) {
+ continue;
+ }
+
+ // All inner funcs should be defined
+ Set<FunctionDesc> innerFuncSet = Sets.newHashSet(dynFunc.getRuntimeFuncs());
+ innerFuncSet.removeAll(definedFuncs);
+ if (!innerFuncSet.isEmpty()) {
+ continue;
+ }
+
+ funcIterator.remove();
+ }
+ }
return result;
}
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
index 9eae6f3f0b..357747663d 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
@@ -34,6 +34,7 @@
import org.apache.kylin.gridtable.IGTComparator;
import org.apache.kylin.measure.MeasureAggregator;
import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+import org.apache.kylin.metadata.datatype.DynamicDimSerializer;
/**
* defines how column values will be encoded to/ decoded from GTRecord
@@ -68,7 +69,7 @@ public TrimmedCubeCodeSystem trimForCoprocessor() {
@Override
public void init(GTInfo info) {
this.info = info;
-
+ ImmutableBitSet dDims = info.getDynamicDims();
this.serializers = new DataTypeSerializer[info.getColumnCount()];
for (int i = 0; i < serializers.length; i++) {
DimensionEncoding dimEnc = i < dimEncs.length ? dimEncs[i] : null;
@@ -77,8 +78,14 @@ public void init(GTInfo info) {
// for dimensions
serializers[i] = dimEnc.asDataTypeSerializer();
} else {
- // for measures
- serializers[i] = DataTypeSerializer.create(info.getColumnType(i));
+ DataTypeSerializer dSerializer = DataTypeSerializer.create(info.getColumnType(i));
+ if (dDims != null && dDims.get(i)) {
+ // for dynamic dimensions
+ dSerializer = new DynamicDimSerializer(dSerializer);
+ } else {
+ // for measures
+ }
+ serializers[i] = dSerializer;
}
}
}
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeGridTable.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeGridTable.java
index 5cee9df12a..2a819bcd7f 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeGridTable.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeGridTable.java
@@ -28,10 +28,25 @@ public static GTInfo newGTInfo(Cuboid cuboid, IDimensionEncodingMap dimEncMap) {
GTInfo.Builder builder = GTInfo.builder();
builder.setTableName("Cuboid " + cuboid.getId());
- builder.setCodeSystem(new CubeCodeSystem(mapping.getDimensionEncodings(dimEncMap), mapping.getDependentMetricsMap()));
+ builder.setCodeSystem(
+ new CubeCodeSystem(mapping.getDimensionEncodings(dimEncMap), mapping.getDependentMetricsMap()));
builder.setColumns(mapping.getDataTypes());
builder.setPrimaryKey(mapping.getPrimaryKey());
builder.enableColumnBlock(mapping.getColumnBlocks());
return builder.build();
}
+
+ public static GTInfo newGTInfo(Cuboid cuboid, IDimensionEncodingMap dimEncMap, CuboidToGridTableMapping mapping) {
+ GTInfo.Builder builder = GTInfo.builder();
+ builder.setTableName("Cuboid " + cuboid.getId());
+ builder.setCodeSystem(
+ new CubeCodeSystem(mapping.getDimensionEncodings(dimEncMap), mapping.getDependentMetricsMap()));
+ builder.setColumns(mapping.getDataTypes());
+ builder.setPrimaryKey(mapping.getPrimaryKey());
+ builder.enableColumnBlock(mapping.getColumnBlocks());
+ if (mapping instanceof CuboidToGridTableMappingExt) {
+ builder.enableDynamicDims(((CuboidToGridTableMappingExt) mapping).getDynamicDims());
+ }
+ return builder.build();
+ }
}
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMapping.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMapping.java
index 6879687b4c..05256cc09a 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMapping.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMapping.java
@@ -38,6 +38,7 @@
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.TblColRef;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -201,17 +202,17 @@ public int getIndexOf(FunctionDesc metric) {
public ImmutableBitSet makeGridTableColumns(Set<TblColRef> dimensions) {
BitSet result = new BitSet();
for (TblColRef dim : dimensions) {
- int idx = this.getIndexOf(dim);
+ int idx = getIndexOf(dim);
if (idx >= 0)
result.set(idx);
}
return new ImmutableBitSet(result);
}
- public ImmutableBitSet makeGridTableColumns(Collection<FunctionDesc> metrics) {
+ public ImmutableBitSet makeGridTableColumns(Collection<? extends FunctionDesc> metrics) {
BitSet result = new BitSet();
for (FunctionDesc metric : metrics) {
- int idx = this.getIndexOf(metric);
+ int idx = getIndexOf(metric);
if (idx < 0)
throw new IllegalStateException(metric + " not found in " + this);
result.set(idx);
@@ -227,8 +228,8 @@ public ImmutableBitSet makeGridTableColumns(Collection<FunctionDesc> metrics) {
Collections.sort(metricList, new Comparator<FunctionDesc>() {
@Override
public int compare(FunctionDesc o1, FunctionDesc o2) {
- int a = CuboidToGridTableMapping.this.getIndexOf(o1);
- int b = CuboidToGridTableMapping.this.getIndexOf(o2);
+ int a = getIndexOf(o1);
+ int b = getIndexOf(o2);
return a - b;
}
});
@@ -241,4 +242,7 @@ public int compare(FunctionDesc o1, FunctionDesc o2) {
return result;
}
+ public Map<TblColRef, Integer> getDim2gt() {
+ return ImmutableMap.copyOf(dim2gt);
+ }
}
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMappingExt.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMappingExt.java
new file mode 100644
index 0000000000..32c4ca0fb8
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMappingExt.java
@@ -0,0 +1,134 @@
+/*
+ * 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.cube.gridtable;
+
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
+import org.apache.kylin.metadata.model.FunctionDesc;
+import org.apache.kylin.metadata.model.TblColRef;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public class CuboidToGridTableMappingExt extends CuboidToGridTableMapping {
+ private final List<TblColRef> dynDims;
+ private final List<DynamicFunctionDesc> dynFuncs;
+
+ private ImmutableBitSet dynamicDims;
+
+ private List<DataType> dynGtDataTypes;
+ private List<ImmutableBitSet> dynGtColBlocks;
+
+ private Map<TblColRef, Integer> dynDim2gt;
+
+ private Map<FunctionDesc, Integer> dynMetrics2gt;
+
+ public CuboidToGridTableMappingExt(Cuboid cuboid, List<TblColRef> dynDims, List<DynamicFunctionDesc> dynFuncs) {
+ super(cuboid);
+ this.dynDims = dynDims;
+ this.dynFuncs = dynFuncs;
+ init();
+ }
+
+ private void init() {
+ dynGtDataTypes = Lists.newArrayList();
+ dynGtColBlocks = Lists.newArrayList();
+ dynDim2gt = Maps.newHashMap();
+ dynMetrics2gt = Maps.newHashMap();
+
+ int gtColIdx = super.getColumnCount();
+
+ BitSet rtColBlock = new BitSet();
+ // dynamic dimensions
+ for (TblColRef rtDim : dynDims) {
+ dynDim2gt.put(rtDim, gtColIdx);
+ dynGtDataTypes.add(rtDim.getType());
+ rtColBlock.set(gtColIdx);
+ gtColIdx++;
+ }
+ dynamicDims = new ImmutableBitSet(rtColBlock);
+
+ // dynamic metrics
+ for (DynamicFunctionDesc rtFunc : dynFuncs) {
+ dynMetrics2gt.put(rtFunc, gtColIdx);
+ dynGtDataTypes.add(rtFunc.getReturnDataType());
+ rtColBlock.set(gtColIdx);
+ gtColIdx++;
+ }
+
+ dynGtColBlocks.add(new ImmutableBitSet(rtColBlock));
+ }
+
+ public ImmutableBitSet getDynamicDims() {
+ return dynamicDims;
+ }
+
+ @Override
+ public int getColumnCount() {
+ return super.getColumnCount() + dynDims.size() + dynFuncs.size();
+ }
+
+ @Override
+ public DataType[] getDataTypes() {
+ return (DataType[]) ArrayUtils.addAll(super.getDataTypes(),
+ dynGtDataTypes.toArray(new DataType[dynGtDataTypes.size()]));
+ }
+
+ @Override
+ public ImmutableBitSet[] getColumnBlocks() {
+ return (ImmutableBitSet[]) ArrayUtils.addAll(super.getColumnBlocks(),
+ dynGtColBlocks.toArray(new ImmutableBitSet[dynGtColBlocks.size()]));
+ }
+
+ @Override
+ public int getIndexOf(TblColRef dimension) {
+ Integer i = super.getIndexOf(dimension);
+ if (i < 0) {
+ i = dynDim2gt.get(dimension);
+ }
+ return i == null ? -1 : i;
+ }
+
+ @Override
+ public int getIndexOf(FunctionDesc metric) {
+ Integer r = super.getIndexOf(metric);
+ if (r < 0) {
+ r = dynMetrics2gt.get(metric);
+ }
+ return r == null ? -1 : r;
+ }
+
+ @Override
+ public int[] getMetricsIndexes(Collection<FunctionDesc> metrics) {
+ int[] result = new int[metrics.size()];
+ int i = 0;
+ for (FunctionDesc metric : metrics) {
+ result[i++] = getIndexOf(metric);
+ }
+ return result;
+ }
+}
\ No newline at end of file
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/ScanRangePlannerBase.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/ScanRangePlannerBase.java
index ec86d42de3..811d51282b 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/ScanRangePlannerBase.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/ScanRangePlannerBase.java
@@ -32,6 +32,7 @@
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.gridtable.GTInfo;
import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.ConstantTupleFilter;
import org.apache.kylin.metadata.filter.LogicalTupleFilter;
@@ -61,6 +62,10 @@
protected RecordComparator rangeEndComparator;
protected RecordComparator rangeStartEndComparator;
+ protected ImmutableBitSet gtDynColumns;
+ protected ImmutableBitSet gtRtAggrMetrics;
+ protected Map<Integer, TupleExpression> tupleExpressionMap;
+
public abstract GTScanRequest planScanRequest();
protected TupleFilter flattenToOrAndFilter(TupleFilter filter) {
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
index 6dc5de3675..75e79d0a40 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
@@ -354,7 +354,7 @@ public boolean shouldBypass(GTRecord record) {
for (int i = 0; i < dimensions.trueBitCount(); i++) {
int c = dimensions.trueBitAt(i);
int l = info.codeSystem.maxCodeLength(c);
- boolean m = groupBy.get(c) ? true : false;
+ boolean m = groupBy.get(c);
for (int j = 0; j < l; j++) {
mask[p++] = m;
}
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTFunctionScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTFunctionScanner.java
new file mode 100644
index 0000000000..65466f1cbe
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTFunctionScanner.java
@@ -0,0 +1,117 @@
+/*
+ * 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.gridtable;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.DecimalUtil;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.metadata.expression.TupleExpression;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GTFunctionScanner implements IGTScanner {
+ private static final Logger logger = LoggerFactory.getLogger(GTFunctionScanner.class);
+
+ protected IGTScanner rawScanner;
+ private final ImmutableBitSet rtAggrMetrics;
+ private final Map<Integer, TupleExpression> tupleExpressionMap;
+ private final IEvaluatableTuple oneTuple; // avoid instance creation
+ private final IFilterCodeSystem<ByteArray> filterCodeSystem;
+
+ private GTRecord next = null;
+
+ protected GTFunctionScanner(IGTScanner rawScanner, GTScanRequest req) {
+ this.rawScanner = rawScanner;
+ this.tupleExpressionMap = req.getTupleExpressionMap();
+ this.rtAggrMetrics = req.getRtAggrMetrics();
+ this.oneTuple = new IEvaluatableTuple() {
+ @Override
+ public Object getValue(TblColRef col) {
+ int idx = col.getColumnDesc().getZeroBasedIndex();
+ return rtAggrMetrics.get(idx) ? DecimalUtil.toBigDecimal(next.getValue(idx)) : next.get(idx);
+ }
+ };
+ this.filterCodeSystem = GTUtil.wrap(getInfo().codeSystem.getComparator());
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return rawScanner.getInfo();
+ }
+
+ @Override
+ public void close() throws IOException {
+ rawScanner.close();
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return new Iterator<GTRecord>() {
+ private Iterator<GTRecord> inputIterator = rawScanner.iterator();
+
+ @Override
+ public boolean hasNext() {
+ if (next != null)
+ return true;
+
+ if (inputIterator.hasNext()) {
+ next = inputIterator.next();
+ calculateDynamics();
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public GTRecord next() {
+ // fetch next record
+ if (next == null) {
+ hasNext();
+ if (next == null)
+ throw new NoSuchElementException();
+ }
+
+ GTRecord result = next;
+ next = null;
+ return result;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ private void calculateDynamics() {
+ for (int c : tupleExpressionMap.keySet()) {
+ TupleExpression tupleExpr = tupleExpressionMap.get(c);
+ Object rtResult = tupleExpr.calculate(oneTuple, filterCodeSystem);
+ next.setValue(c, rtResult);
+ }
+ }
+ };
+ }
+}
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
index d10d6e7cac..739adf875d 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
@@ -57,6 +57,9 @@ public static Builder builder() {
int rowBlockSize; // 0: disable row block
ImmutableBitSet colBlocksAll;
+ // not included during serialization, only used for loadColumns
+ ImmutableBitSet dynamicDims;
+
// must create from builder
private GTInfo() {
}
@@ -93,6 +96,10 @@ public ImmutableBitSet getAllColumns() {
return colAll;
}
+ public ImmutableBitSet getDynamicDims() {
+ return dynamicDims;
+ }
+
public boolean isRowBlockEnabled() {
return rowBlockSize > 0;
}
@@ -214,6 +221,10 @@ private void validateColumnBlocks() {
it.remove();
}
colBlocks = list.toArray(new ImmutableBitSet[list.size()]);
+
+ // for dynamic dimensions
+ if (dynamicDims == null)
+ dynamicDims = ImmutableBitSet.EMPTY;
}
public static class Builder {
@@ -269,6 +280,12 @@ public Builder setColumnPreferIndex(ImmutableBitSet colPreferIndex) {
return this;
}
+ /** optional */
+ public Builder enableDynamicDims(ImmutableBitSet dynamicDims) {
+ info.dynamicDims = dynamicDims;
+ return this;
+ }
+
public GTInfo build() {
info.validate();
return info;
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
index 36bd09503b..b4a57c76e9 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
@@ -61,6 +61,10 @@ public ByteArray get(int i) {
return cols[i];
}
+ public Object getValue(int i) {
+ return info.codeSystem.decodeColumnValue(i, cols[i].asBuffer());
+ }
+
public ByteArray[] getInternal() {
return cols;
}
@@ -69,6 +73,14 @@ public void set(int i, ByteArray data) {
cols[i].reset(data.array(), data.offset(), data.length());
}
+ public void setValue(int i, Object value) {
+ ByteArray space = new ByteArray(info.codeSystem.maxCodeLength(i));
+ ByteBuffer buf = space.asBuffer();
+ info.codeSystem.encodeColumnValue(i, value, buf);
+ set(i, space);
+ cols[i].reset(buf.array(), buf.arrayOffset(), buf.position());
+ }
+
/** set record to the codes of specified values, new space allocated to hold the codes */
public GTRecord setValues(Object... values) {
setValues(info.colAll, new ByteArray(info.getMaxRecordLength()), values);
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
index 412ff7f7d7..e3a2234e65 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
@@ -23,6 +23,7 @@
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.commons.io.IOUtils;
@@ -32,8 +33,11 @@
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.common.util.SerializeToByteBuffer;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.measure.BufferedMeasureCodec;
import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.expression.TupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpressionSerializer;
import org.apache.kylin.metadata.filter.StringCodeSystem;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilterSerializer;
@@ -42,6 +46,7 @@
import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
public class GTScanRequest {
@@ -52,11 +57,18 @@
//changing it might break org.apache.kylin.query.ITKylinQueryTest.testTimeoutQuery()
public static final int terminateCheckInterval = 100;
+ private CuboidToGridTableMapping mapping;
private GTInfo info;
private List<GTScanRange> ranges;
private ImmutableBitSet columns;
private transient ImmutableBitSet selectedColBlocks;
+ // optional expression
+ private ImmutableBitSet rtAggrMetrics;
+
+ private ImmutableBitSet dynamicCols;
+ private Map<Integer, TupleExpression> tupleExpressionMap;
+
// optional filtering
private TupleFilter filterPushDown;
private TupleFilter havingFilterPushDown;
@@ -81,8 +93,9 @@
private transient boolean doingStorageAggregation = false;
GTScanRequest(GTInfo info, List<GTScanRange> ranges, ImmutableBitSet dimensions, ImmutableBitSet aggrGroupBy, //
- ImmutableBitSet aggrMetrics, String[] aggrMetricsFuncs, TupleFilter filterPushDown,
- TupleFilter havingFilterPushDown, //
+ ImmutableBitSet aggrMetrics, String[] aggrMetricsFuncs, ImmutableBitSet rtAggrMetrics, //
+ ImmutableBitSet dynamicCols, Map<Integer, TupleExpression> tupleExpressionMap, //
+ TupleFilter filterPushDown, TupleFilter havingFilterPushDown, //
boolean allowStorageAggregation, double aggCacheMemThreshold, int storageScanRowNumThreshold, //
int storagePushDownLimit, StorageLimitLevel storageLimitLevel, String storageBehavior, long startTime,
long timeout) {
@@ -100,6 +113,11 @@
this.aggrMetrics = aggrMetrics;
this.aggrMetricsFuncs = aggrMetricsFuncs;
+ this.rtAggrMetrics = rtAggrMetrics;
+
+ this.dynamicCols = dynamicCols;
+ this.tupleExpressionMap = tupleExpressionMap;
+
this.storageBehavior = storageBehavior;
this.startTime = startTime;
this.timeout = timeout;
@@ -114,7 +132,7 @@
private void validate(GTInfo info) {
if (hasAggregation()) {
- if (aggrGroupBy.intersects(aggrMetrics))
+ if (aggrGroupBy.intersects(aggrMetrics) || aggrGroupBy.intersects(rtAggrMetrics))
throw new IllegalStateException();
if (aggrMetrics.cardinality() != aggrMetricsFuncs.length)
throw new IllegalStateException();
@@ -133,7 +151,7 @@ private void validate(GTInfo info) {
validateFilterPushDown(info);
}
- this.selectedColBlocks = info.selectColumnBlocks(columns);
+ this.selectedColBlocks = info.selectColumnBlocks(columns.or(rtAggrMetrics).andNot(dynamicCols));
}
@@ -201,6 +219,11 @@ public IGTScanner decorateScanner(IGTScanner scanner, boolean filterToggledOn, b
result = new GTForwardingScanner(result);//need its check function
}
+ if (tupleExpressionMap != null && !tupleExpressionMap.isEmpty()) {
+ logger.info("GTFunctionScanner will be used with expressions " + tupleExpressionMap);
+ result = new GTFunctionScanner(result, this);
+ }
+
if (!aggrToggledOn) {//Skip reading this section if you're not profiling!
lookAndForget(result);
return new EmptyGTScanner();
@@ -263,6 +286,14 @@ public boolean hasAggregation() {
return !aggrGroupBy.isEmpty() || !aggrMetrics.isEmpty();
}
+ public CuboidToGridTableMapping getMapping() {
+ return mapping;
+ }
+
+ public void setMapping(CuboidToGridTableMapping mapping) {
+ this.mapping = mapping;
+ }
+
public GTInfo getInfo() {
return info;
}
@@ -292,7 +323,7 @@ public TupleFilter getHavingFilterPushDown() {
}
public ImmutableBitSet getDimensions() {
- return this.getColumns().andNot(this.getAggrMetrics());
+ return this.columns.andNot(this.aggrMetrics);
}
public ImmutableBitSet getAggrGroupBy() {
@@ -307,6 +338,18 @@ public ImmutableBitSet getAggrMetrics() {
return aggrMetricsFuncs;
}
+ public ImmutableBitSet getDynamicCols() {
+ return dynamicCols;
+ }
+
+ public ImmutableBitSet getRtAggrMetrics() {
+ return rtAggrMetrics;
+ }
+
+ public Map<Integer, TupleExpression> getTupleExpressionMap() {
+ return tupleExpressionMap;
+ }
+
public boolean isAllowStorageAggregation() {
return allowStorageAggregation;
}
@@ -402,6 +445,18 @@ public void serialize(GTScanRequest value, ByteBuffer out) {
BytesUtil.writeVLong(value.startTime, out);
BytesUtil.writeVLong(value.timeout, out);
BytesUtil.writeUTFString(value.storageBehavior, out);
+
+ // for dynamic related info
+ ImmutableBitSet.serializer.serialize(value.dynamicCols, out);
+
+ BytesUtil.writeVInt(value.tupleExpressionMap.size(), out);
+ for (int c : value.tupleExpressionMap.keySet()) {
+ TupleExpression tupleExpr = value.tupleExpressionMap.get(c);
+ BytesUtil.writeVInt(c, out);
+ BytesUtil.writeByteArray(TupleExpressionSerializer.serialize(tupleExpr,
+ GTUtil.wrap(value.info.codeSystem.getComparator())), out);
+ }
+ ImmutableBitSet.serializer.serialize(value.rtAggrMetrics, out);
}
@Override
@@ -445,8 +500,22 @@ public GTScanRequest deserialize(ByteBuffer in) {
long timeout = BytesUtil.readVLong(in);
String storageBehavior = BytesUtil.readUTFString(in);
+ ImmutableBitSet aDynCols = ImmutableBitSet.serializer.deserialize(in);
+
+ int nTupleExprs = BytesUtil.readVInt(in);
+ Map<Integer, TupleExpression> sTupleExpressionMap = Maps.newHashMapWithExpectedSize(nTupleExprs);
+ for (int i = 0; i < nTupleExprs; i++) {
+ int sC = BytesUtil.readVInt(in);
+ TupleExpression sTupleExpr = TupleExpressionSerializer.deserialize(BytesUtil.readByteArray(in),
+ GTUtil.wrap(sInfo.codeSystem.getComparator()));
+ sTupleExpressionMap.put(sC, sTupleExpr);
+ }
+ ImmutableBitSet aRuntimeAggrMetrics = ImmutableBitSet.serializer.deserialize(in);
+
return new GTScanRequestBuilder().setInfo(sInfo).setRanges(sRanges).setDimensions(sColumns)
.setAggrGroupBy(sAggGroupBy).setAggrMetrics(sAggrMetrics).setAggrMetricsFuncs(sAggrMetricFuncs)
+ .setRtAggrMetrics(aRuntimeAggrMetrics).setDynamicColumns(aDynCols)
+ .setExprsPushDown(sTupleExpressionMap)
.setFilterPushDown(sGTFilter).setHavingFilterPushDown(sGTHavingFilter)
.setAllowStorageAggregation(sAllowPreAggr).setAggCacheMemThreshold(sAggrCacheGB)
.setStorageScanRowNumThreshold(storageScanRowNumThreshold)
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequestBuilder.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequestBuilder.java
index 28fe40a3d7..94a89e6c5e 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequestBuilder.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequestBuilder.java
@@ -20,11 +20,15 @@
import java.util.BitSet;
import java.util.List;
+import java.util.Map;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
+import com.google.common.collect.Maps;
+
public class GTScanRequestBuilder {
private GTInfo info;
private List<GTScanRange> ranges;
@@ -34,6 +38,9 @@
private ImmutableBitSet aggrGroupBy = null;
private ImmutableBitSet aggrMetrics = null;
private String[] aggrMetricsFuncs = null;
+ private ImmutableBitSet dynamicColumns;
+ private ImmutableBitSet rtAggrMetrics;
+ private Map<Integer, TupleExpression> exprsPushDown;
private boolean allowStorageAggregation = true;
private double aggCacheMemThreshold = 0;
private int storageScanRowNumThreshold = Integer.MAX_VALUE;// storage should terminate itself when $storageScanRowNumThreshold cuboid rows are scanned, and throw exception.
@@ -53,6 +60,21 @@ public GTScanRequestBuilder setRanges(List<GTScanRange> ranges) {
return this;
}
+ public GTScanRequestBuilder setDynamicColumns(ImmutableBitSet dynamicColumns) {
+ this.dynamicColumns = dynamicColumns;
+ return this;
+ }
+
+ public GTScanRequestBuilder setRtAggrMetrics(ImmutableBitSet rtAggrMetrics) {
+ this.rtAggrMetrics = rtAggrMetrics;
+ return this;
+ }
+
+ public GTScanRequestBuilder setExprsPushDown(Map<Integer, TupleExpression> exprsPushDown) {
+ this.exprsPushDown = exprsPushDown;
+ return this;
+ }
+
public GTScanRequestBuilder setFilterPushDown(TupleFilter filterPushDown) {
this.filterPushDown = filterPushDown;
return this;
@@ -136,6 +158,18 @@ public GTScanRequest createGTScanRequest() {
aggrMetricsFuncs = new String[0];
}
+ if (rtAggrMetrics == null) {
+ rtAggrMetrics = new ImmutableBitSet(new BitSet());
+ }
+
+ if (dynamicColumns == null) {
+ dynamicColumns = new ImmutableBitSet(new BitSet());
+ }
+
+ if (exprsPushDown == null) {
+ exprsPushDown = Maps.newHashMap();
+ }
+
if (storageBehavior == null) {
storageBehavior = BackdoorToggles.getCoprocessorBehavior() == null
? StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.toString()
@@ -145,8 +179,9 @@ public GTScanRequest createGTScanRequest() {
this.startTime = startTime == -1 ? System.currentTimeMillis() : startTime;
this.timeout = timeout == -1 ? 300000 : timeout;
- return new GTScanRequest(info, ranges, dimensions, aggrGroupBy, aggrMetrics, aggrMetricsFuncs, filterPushDown,
- havingFilterPushDown, allowStorageAggregation, aggCacheMemThreshold, storageScanRowNumThreshold,
- storagePushDownLimit, storageLimitLevel, storageBehavior, startTime, timeout);
+ return new GTScanRequest(info, ranges, dimensions, aggrGroupBy, aggrMetrics, aggrMetricsFuncs, rtAggrMetrics,
+ dynamicColumns, exprsPushDown, filterPushDown, havingFilterPushDown, allowStorageAggregation,
+ aggCacheMemThreshold, storageScanRowNumThreshold, storagePushDownLimit, storageLimitLevel,
+ storageBehavior, startTime, timeout);
}
}
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java
index 5c9dfe3eb3..25e54550a7 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java
@@ -20,13 +20,15 @@
import java.nio.ByteBuffer;
import java.util.Collection;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
+import org.apache.kylin.metadata.expression.TupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpressionSerializer;
import org.apache.kylin.metadata.filter.ColumnTupleFilter;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.ConstantTupleFilter;
@@ -34,9 +36,11 @@
import org.apache.kylin.metadata.filter.IFilterCodeSystem;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilterSerializer;
+import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.model.TblColRef;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
public class GTUtil {
@@ -65,7 +69,12 @@ public static TupleFilter convertFilterUnevaluatable(TupleFilter rootFilter, GTI
public static TupleFilter convertFilterColumnsAndConstants(TupleFilter rootFilter, GTInfo info, //
List<TblColRef> colMapping, Set<TblColRef> unevaluatableColumnCollector) {
Map<TblColRef, Integer> map = colListToMap(colMapping);
- TupleFilter filter = convertFilter(rootFilter, info, map, true, unevaluatableColumnCollector);
+ return convertFilterColumnsAndConstants(rootFilter, info, map, unevaluatableColumnCollector);
+ }
+
+ public static TupleFilter convertFilterColumnsAndConstants(TupleFilter rootFilter, GTInfo info, //
+ Map<TblColRef, Integer> colMapping, Set<TblColRef> unevaluatableColumnCollector) {
+ TupleFilter filter = convertFilter(rootFilter, info, colMapping, true, unevaluatableColumnCollector);
// optimize the filter: after translating with dictionary, some filters become determined
// e.g.
@@ -77,7 +86,7 @@ public static TupleFilter convertFilterColumnsAndConstants(TupleFilter rootFilte
}
protected static Map<TblColRef, Integer> colListToMap(List<TblColRef> colMapping) {
- Map<TblColRef, Integer> map = new HashMap<>();
+ Map<TblColRef, Integer> map = Maps.newHashMap();
for (int i = 0; i < colMapping.size(); i++) {
map.put(colMapping.get(i), i);
}
@@ -98,6 +107,51 @@ private static TupleFilter convertFilter(TupleFilter rootFilter, final GTInfo in
return TupleFilterSerializer.deserialize(bytes, filterCodeSystem);
}
+ public static TupleExpression convertFilterColumnsAndConstants(TupleExpression rootExpression, GTInfo info,
+ CuboidToGridTableMapping mapping, Set<TblColRef> unevaluatableColumnCollector) {
+ Map<TblColRef, FunctionDesc> innerFuncMap = Maps.newHashMap();
+ return convertFilterColumnsAndConstants(rootExpression, info, mapping, innerFuncMap,
+ unevaluatableColumnCollector);
+ }
+
+ public static TupleExpression convertFilterColumnsAndConstants(TupleExpression rootExpression, GTInfo info,
+ CuboidToGridTableMapping mapping, //
+ Map<TblColRef, FunctionDesc> innerFuncMap, Set<TblColRef> unevaluatableColumnCollector) {
+ return convertExpression(rootExpression, info, mapping, innerFuncMap, true, unevaluatableColumnCollector);
+ }
+
+ private static TupleExpression convertExpression(TupleExpression rootExpression, final GTInfo info,
+ final CuboidToGridTableMapping mapping, //
+ final Map<TblColRef, FunctionDesc> innerFuncMap, final boolean encodeConstants,
+ final Set<TblColRef> unevaluatableColumnCollector) {
+ IFilterCodeSystem<ByteArray> filterCodeSystem = wrap(info.codeSystem.getComparator());
+
+ final TupleExpressionSerializer.Decorator decorator = new TupleExpressionSerializer.Decorator() {
+ @Override
+ public TupleFilter convertInnerFilter(TupleFilter filter) {
+ return convertFilter(filter, info, mapping.getDim2gt(), encodeConstants, unevaluatableColumnCollector);
+ }
+
+ @Override
+ public TupleExpression convertInnerExpression(TupleExpression expression) {
+ return convertFilterColumnsAndConstants(expression, info, mapping, innerFuncMap,
+ unevaluatableColumnCollector);
+ }
+
+ @Override
+ public TblColRef mapCol(TblColRef col) {
+ int gtColIdx = mapping.getIndexOf(col);
+ if (gtColIdx < 0 && innerFuncMap.get(col) != null) {
+ gtColIdx = mapping.getIndexOf(innerFuncMap.get(col));
+ }
+ return info.colRef(gtColIdx);
+ }
+ };
+
+ byte[] bytes = TupleExpressionSerializer.serialize(rootExpression, decorator, filterCodeSystem);
+ return TupleExpressionSerializer.deserialize(bytes, filterCodeSystem);
+ }
+
public static IFilterCodeSystem<ByteArray> wrap(final IGTComparator comp) {
return new IFilterCodeSystem<ByteArray>() {
diff --git a/core-metadata/pom.xml b/core-metadata/pom.xml
index 21aa6ee733..791ae86778 100644
--- a/core-metadata/pom.xml
+++ b/core-metadata/pom.xml
@@ -111,6 +111,12 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito</artifactId>
+ <version>${powermock.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/basic/BigDecimalSumAggregator.java b/core-metadata/src/main/java/org/apache/kylin/measure/basic/BigDecimalSumAggregator.java
index 9f6ffc2ec6..fa59c54320 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/basic/BigDecimalSumAggregator.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/basic/BigDecimalSumAggregator.java
@@ -27,15 +27,20 @@
@SuppressWarnings("serial")
public class BigDecimalSumAggregator extends MeasureAggregator<BigDecimal> {
- BigDecimal sum = new BigDecimal(0);
+ BigDecimal sum = null;
@Override
public void reset() {
- sum = new BigDecimal(0);
+ sum = null;
}
@Override
public void aggregate(BigDecimal value) {
+ if (value == null)
+ return;
+ if (sum == null) {
+ sum = new BigDecimal(0);
+ }
sum = sum.add(value);
}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
index ba1c4ff285..967c00de57 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
@@ -46,6 +46,12 @@ public BigDecimalSerializer(DataType type) {
@Override
public void serialize(BigDecimal value, ByteBuffer out) {
+ if (value == null) {
+ BytesUtil.writeVInt(0, out);
+ BytesUtil.writeVInt(-1, out);
+ return;
+ }
+
if (value.scale() > type.getScale()) {
if (avoidVerbose++ % 10000 == 0) {
logger.warn("value's scale has exceeded the " + type.getScale() + ", cut it off, to ensure encoded value do not exceed maxLength " + maxLength + " times:" + (avoidVerbose));
@@ -67,6 +73,10 @@ public BigDecimal deserialize(ByteBuffer in) {
int scale = BytesUtil.readVInt(in);
int n = BytesUtil.readVInt(in);
+ if (n < 0) {
+ return null;
+ }
+
byte[] bytes = new byte[n];
in.get(bytes);
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataType.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataType.java
index de1e9dfaf5..5ccc1f3fc8 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataType.java
@@ -153,6 +153,10 @@ public static DataType getType(String type) {
return cached;
}
+ public static boolean isNumberFamily(String name) {
+ return NUMBER_FAMILY.contains(name);
+ }
+
// ============================================================================
private String name;
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DynamicDimSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DynamicDimSerializer.java
new file mode 100644
index 0000000000..a1c42a813f
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DynamicDimSerializer.java
@@ -0,0 +1,63 @@
+/*
+ * 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.datatype;
+
+import java.nio.ByteBuffer;
+
+/**
+ * For dynamic dimensions, the code length must be fixed
+ */
+public class DynamicDimSerializer<T> extends DataTypeSerializer<T> {
+
+ private final DataTypeSerializer<T> dimDataTypeSerializer;
+
+ public DynamicDimSerializer(DataTypeSerializer<T> dimDataTypeSerializer) {
+ this.dimDataTypeSerializer = dimDataTypeSerializer;
+ }
+
+ public void serialize(T value, ByteBuffer out) {
+ dimDataTypeSerializer.serialize(value, out);
+ }
+
+ public T deserialize(ByteBuffer in) {
+ return dimDataTypeSerializer.deserialize(in);
+ }
+
+ public int peekLength(ByteBuffer in) {
+ return maxLength();
+ }
+
+ public int maxLength() {
+ return dimDataTypeSerializer.maxLength();
+ }
+
+ public int getStorageBytesEstimate() {
+ return dimDataTypeSerializer.getStorageBytesEstimate();
+ }
+
+ /** An optional convenient method that converts a string to this data type (for dimensions) */
+ public T valueOf(String str) {
+ return dimDataTypeSerializer.valueOf(str);
+ }
+
+ /** Convert from obj to string */
+ public String toString(T value) {
+ return dimDataTypeSerializer.toString(value);
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/BinaryTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/BinaryTupleExpression.java
new file mode 100644
index 0000000000..adafd9cf84
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/BinaryTupleExpression.java
@@ -0,0 +1,149 @@
+/*
+ * 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.expression;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.kylin.common.util.DecimalUtil;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+import com.google.common.collect.Lists;
+
+public class BinaryTupleExpression extends TupleExpression {
+
+ public BinaryTupleExpression(ExpressionOperatorEnum op) {
+ this(op, Lists.<TupleExpression> newArrayListWithExpectedSize(2));
+ }
+
+ public BinaryTupleExpression(ExpressionOperatorEnum op, List<TupleExpression> exprs) {
+ super(op, exprs);
+
+ boolean opGood = (op == ExpressionOperatorEnum.PLUS || op == ExpressionOperatorEnum.MINUS
+ || op == ExpressionOperatorEnum.MULTIPLE || op == ExpressionOperatorEnum.DIVIDE);
+ if (opGood == false)
+ throw new IllegalArgumentException("Unsupported operator " + op);
+ }
+
+ @Override
+ public boolean ifForDynamicColumn() {
+ return ifAbleToPushDown();
+ }
+
+ @Override
+ public void verify() {
+ switch (operator) {
+ case MULTIPLE:
+ verifyMultiply();
+ break;
+ case DIVIDE:
+ verifyDivide();
+ break;
+ default:
+ }
+ }
+
+ private void verifyMultiply() {
+ if (ExpressionColCollector.collectMeasureColumns(getLeft()).size() > 0 //
+ && ExpressionColCollector.collectMeasureColumns(getRight()).size() > 0) {
+ throw new IllegalArgumentException(
+ "That both of the two sides of the BinaryTupleExpression own columns is not supported for "
+ + operator.toString());
+ }
+ }
+
+ private void verifyDivide() {
+ if (ExpressionColCollector.collectMeasureColumns(getRight()).size() > 0) {
+ throw new IllegalArgumentException(
+ "That the right side of the BinaryTupleExpression owns columns is not supported for "
+ + operator.toString());
+ }
+ }
+
+ @Override
+ public BigDecimal calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ assert children.size() == 2;
+ BigDecimal left = DecimalUtil.toBigDecimal(getLeft().calculate(tuple, cs));
+ if (left == null)
+ return null;
+ BigDecimal right = DecimalUtil.toBigDecimal(getRight().calculate(tuple, cs));
+ if (right == null)
+ return null;
+ switch (operator) {
+ case PLUS:
+ return left.add(right);
+ case MINUS:
+ return left.subtract(right);
+ case MULTIPLE:
+ return left.multiply(right);
+ case DIVIDE:
+ return left.divide(right);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitBinary(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ }
+
+ public TupleExpression getLeft() {
+ return children.get(0);
+ }
+
+ public TupleExpression getRight() {
+ return children.get(1);
+ }
+
+ public String toString() {
+ return operator.toString() + "(" + getLeft().toString() + "," + getRight().toString() + ")";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ BinaryTupleExpression that = (BinaryTupleExpression) o;
+
+ if (operator != that.operator)
+ return false;
+ return children.equals(that.children);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = operator != null ? operator.hashCode() : 0;
+ result = 31 * result + (children != null ? children.hashCode() : 0);
+ return result;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/CaseTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/CaseTupleExpression.java
new file mode 100644
index 0000000000..6b9034b3cc
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/CaseTupleExpression.java
@@ -0,0 +1,180 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.filter.TupleFilterSerializer;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+public class CaseTupleExpression extends TupleExpression {
+
+ private List<Pair<TupleFilter, TupleExpression>> whenList;
+ private TupleExpression elseExpr;
+
+ public CaseTupleExpression(List<Pair<TupleFilter, TupleExpression>> whenList, TupleExpression elseExpr) {
+ super(ExpressionOperatorEnum.CASE, Collections.<TupleExpression> emptyList());
+ this.whenList = whenList;
+ this.elseExpr = elseExpr;
+ }
+
+ @Override
+ protected boolean ifAbleToPushDown() {
+ if (ifAbleToPushDown == null) {
+ for (Pair<TupleFilter, TupleExpression> whenEntry : whenList) {
+ ifAbleToPushDown = whenEntry.getSecond().ifAbleToPushDown();
+ if (!ifAbleToPushDown) {
+ break;
+ }
+ }
+ if (elseExpr != null && Boolean.TRUE.equals(ifAbleToPushDown)) {
+ ifAbleToPushDown = elseExpr.ifAbleToPushDown();
+ }
+ if (ifAbleToPushDown == null) {
+ ifAbleToPushDown = true;
+ }
+ }
+ return ifAbleToPushDown;
+ }
+
+ @Override
+ public boolean ifForDynamicColumn() {
+ return ifAbleToPushDown();
+ }
+
+ //TODO
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public Object calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ for (Pair<TupleFilter, TupleExpression> entry : whenList) {
+ if (entry.getFirst().evaluate(tuple, cs)) {
+ return entry.getSecond().calculate(tuple, cs);
+ }
+ }
+ if (elseExpr != null) {
+ return elseExpr.calculate(tuple, cs);
+ }
+ return null;
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitCaseCall(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ BytesUtil.writeVInt(whenList.size(), buffer);
+ for (Pair<TupleFilter, TupleExpression> whenEntry : whenList) {
+ byte[] whenBytes = TupleFilterSerializer.serialize(whenEntry.getFirst(), cs);
+ BytesUtil.writeByteArray(whenBytes, buffer);
+
+ byte[] thenBytes = TupleExpressionSerializer.serialize(whenEntry.getSecond(), cs);
+ BytesUtil.writeByteArray(thenBytes, buffer);
+ }
+ if (elseExpr != null) {
+ BytesUtil.writeVInt(1, buffer);
+ byte[] elseBytes = TupleExpressionSerializer.serialize(elseExpr, cs);
+ BytesUtil.writeByteArray(elseBytes, buffer);
+ } else {
+ BytesUtil.writeVInt(-1, buffer);
+ }
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ int nWhenEntries = BytesUtil.readVInt(buffer);
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists.newArrayListWithExpectedSize(nWhenEntries);
+ for (int i = 0; i < nWhenEntries; i++) {
+ TupleFilter tupleFilter = TupleFilterSerializer.deserialize(BytesUtil.readByteArray(buffer), cs);
+ TupleExpression tupleExpression = TupleExpressionSerializer.deserialize(BytesUtil.readByteArray(buffer),
+ cs);
+ whenList.add(new Pair<>(tupleFilter, tupleExpression));
+ }
+ this.whenList = whenList;
+ int flag = BytesUtil.readVInt(buffer);
+ if (flag == 1) {
+ this.elseExpr = TupleExpressionSerializer.deserialize(BytesUtil.readByteArray(buffer), cs);
+ }
+ }
+
+ public List<Pair<TupleFilter, TupleExpression>> getWhenList() {
+ return ImmutableList.copyOf(whenList);
+ }
+
+ public TupleExpression getElseExpr() {
+ return elseExpr;
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append(operator.toString());
+ sb.append("(");
+ boolean ifFirst = true;
+ for (Pair<TupleFilter, TupleExpression> whenEntry : whenList) {
+ if (ifFirst) {
+ ifFirst = false;
+ } else {
+ sb.append(",");
+ }
+ sb.append(whenEntry.getFirst().toString());
+ sb.append(",");
+ sb.append(whenEntry.getSecond().toString());
+ }
+ if (elseExpr != null) {
+ sb.append(",");
+ sb.append(elseExpr.toString());
+ }
+ sb.append(")");
+ return sb.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ CaseTupleExpression that = (CaseTupleExpression) o;
+
+ if (whenList != null ? !whenList.equals(that.whenList) : that.whenList != null)
+ return false;
+ return elseExpr != null ? elseExpr.equals(that.elseExpr) : that.elseExpr == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = whenList != null ? whenList.hashCode() : 0;
+ result = 31 * result + (elseExpr != null ? elseExpr.hashCode() : 0);
+ return result;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ColumnTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ColumnTupleExpression.java
new file mode 100644
index 0000000000..16ebed355f
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ColumnTupleExpression.java
@@ -0,0 +1,150 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.model.ColumnDesc;
+import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.metadata.model.DataModelManager;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.model.TableRef;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+public class ColumnTupleExpression extends TupleExpression {
+
+ private static final String _QUALIFIED_ = "_QUALIFIED_";
+
+ private TblColRef columnRef;
+
+ public ColumnTupleExpression(TblColRef column) {
+ super(ExpressionOperatorEnum.COLUMN, Collections.<TupleExpression> emptyList());
+ this.columnRef = column;
+ }
+
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public Object calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ return tuple.getValue(columnRef);
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitColumn(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ TableRef tableRef = columnRef.getTableRef();
+
+ if (tableRef == null) {
+ // un-qualified column
+ String table = columnRef.getTable();
+ BytesUtil.writeUTFString(table, buffer);
+
+ String columnId = columnRef.getColumnDesc().getId();
+ BytesUtil.writeUTFString(columnId, buffer);
+
+ String columnName = columnRef.getName();
+ BytesUtil.writeUTFString(columnName, buffer);
+
+ String dataType = columnRef.getDatatype();
+ BytesUtil.writeUTFString(dataType, buffer);
+ } else {
+ // qualified column (from model)
+ BytesUtil.writeUTFString(_QUALIFIED_, buffer);
+
+ String model = tableRef.getModel().getName();
+ BytesUtil.writeUTFString(model, buffer);
+
+ String alias = tableRef.getAlias();
+ BytesUtil.writeUTFString(alias, buffer);
+
+ String col = columnRef.getName();
+ BytesUtil.writeUTFString(col, buffer);
+ }
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ String tableName = BytesUtil.readUTFString(buffer);
+
+ if (_QUALIFIED_.equals(tableName)) {
+ // qualified column (from model)
+ String model = BytesUtil.readUTFString(buffer);
+ String alias = BytesUtil.readUTFString(buffer);
+ String col = BytesUtil.readUTFString(buffer);
+
+ KylinConfig config = KylinConfig.getInstanceFromEnv();
+ DataModelDesc modelDesc = DataModelManager.getInstance(config).getDataModelDesc(model);
+ this.columnRef = modelDesc.findColumn(alias, col);
+
+ } else {
+ // un-qualified column
+ TableDesc tableDesc = null;
+
+ if (tableName != null) {
+ tableDesc = new TableDesc();
+ tableDesc.setName(tableName);
+ }
+
+ ColumnDesc column = new ColumnDesc();
+ column.setId(BytesUtil.readUTFString(buffer));
+ column.setName(BytesUtil.readUTFString(buffer));
+ column.setDatatype(BytesUtil.readUTFString(buffer));
+ column.init(tableDesc);
+
+ this.columnRef = column.getRef();
+ }
+ }
+
+ public TblColRef getColumn() {
+ return columnRef;
+ }
+
+ public String toString() {
+ return columnRef.getCanonicalName();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ ColumnTupleExpression that = (ColumnTupleExpression) o;
+
+ return columnRef != null ? columnRef.equals(that.columnRef) : that.columnRef == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ return columnRef != null ? columnRef.hashCode() : 0;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionColCollector.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionColCollector.java
new file mode 100644
index 0000000000..d8246f4ea1
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionColCollector.java
@@ -0,0 +1,120 @@
+/*
+ * 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.expression;
+
+import java.util.Set;
+
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.TblColRef;
+
+import com.google.common.collect.Sets;
+
+public class ExpressionColCollector implements ExpressionVisitor {
+
+ public static Set<TblColRef> collectColumns(TupleExpression tupleExpression) {
+ Pair<Set<TblColRef>, Set<TblColRef>> pairRet = collectColumnsPair(tupleExpression);
+ Set<TblColRef> ret = Sets.newHashSet();
+ ret.addAll(pairRet.getFirst());
+ ret.addAll(pairRet.getSecond());
+ return ret;
+ }
+
+ public static Pair<Set<TblColRef>, Set<TblColRef>> collectColumnsPair(TupleExpression tupleExpression) {
+ ExpressionColCollector collector = new ExpressionColCollector();
+ tupleExpression.accept(collector);
+ return new Pair<>(collector.filterColumns, collector.measureColumns);
+ }
+
+ public static Set<TblColRef> collectFilterColumns(TupleExpression tupleExpression) {
+ ExpressionColCollector collector = new ExpressionColCollector();
+ collector.ifMCols = false;
+ tupleExpression.accept(collector);
+ return collector.filterColumns;
+ }
+
+ public static Set<TblColRef> collectMeasureColumns(TupleExpression tupleExpression) {
+ ExpressionColCollector collector = new ExpressionColCollector();
+ collector.ifFCols = false;
+ tupleExpression.accept(collector);
+ return collector.measureColumns;
+ }
+
+ private final Set<TblColRef> filterColumns = Sets.newHashSet();
+ private final Set<TblColRef> measureColumns = Sets.newHashSet();
+ private boolean ifFCols = true;
+ private boolean ifMCols = true;
+
+ private ExpressionColCollector() {
+ }
+
+ @Override
+ public TupleExpression visitNumber(NumberTupleExpression numExpr) {
+ return numExpr;
+ }
+
+ @Override
+ public TupleExpression visitString(StringTupleExpression strExpr) {
+ return strExpr;
+ }
+
+ @Override
+ public TupleExpression visitColumn(ColumnTupleExpression colExpr) {
+ if (ifMCols) {
+ measureColumns.add(colExpr.getColumn());
+ }
+ return colExpr;
+ }
+
+ @Override
+ public TupleExpression visitBinary(BinaryTupleExpression binaryExpr) {
+ binaryExpr.getLeft().accept(this);
+ binaryExpr.getRight().accept(this);
+ return binaryExpr;
+ }
+
+ @Override
+ public TupleExpression visitCaseCall(CaseTupleExpression caseExpr) {
+ for (Pair<TupleFilter, TupleExpression> entry : caseExpr.getWhenList()) {
+ TupleFilter filter = entry.getFirst();
+ if (ifFCols) {
+ TupleFilter.collectColumns(filter, filterColumns);
+ }
+
+ entry.getSecond().accept(this);
+ }
+ if (caseExpr.getElseExpr() != null) {
+ caseExpr.getElseExpr().accept(this);
+ }
+ return caseExpr;
+ }
+
+ @Override
+ public TupleExpression visitRexCall(RexCallTupleExpression rexCallExpr) {
+ for (TupleExpression child : rexCallExpr.getChildren()) {
+ child.accept(this);
+ }
+ return rexCallExpr;
+ }
+
+ @Override
+ public TupleExpression visitNone(NoneTupleExpression noneExpr) {
+ return noneExpr;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionCountDistributor.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionCountDistributor.java
new file mode 100644
index 0000000000..183481b32d
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionCountDistributor.java
@@ -0,0 +1,140 @@
+/*
+ * 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.expression;
+
+import java.util.List;
+
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.filter.TupleFilter;
+
+import com.google.common.collect.Lists;
+
+public class ExpressionCountDistributor implements ExpressionVisitor {
+
+ private final TupleExpression cntExpr;
+ private boolean ifToCnt;
+ private boolean ifCntSet;
+
+ public ExpressionCountDistributor(TupleExpression cntExpr) {
+ this.cntExpr = cntExpr;
+ this.ifToCnt = true;
+ this.ifCntSet = false;
+ }
+
+ @Override
+ public TupleExpression visitNumber(NumberTupleExpression numExpr) {
+ NumberTupleExpression copyExpr = new NumberTupleExpression(numExpr.getValue());
+ if (ifToCnt) {
+ List<TupleExpression> children = Lists.newArrayList(cntExpr, copyExpr);
+ ifCntSet = true;
+ return new BinaryTupleExpression(TupleExpression.ExpressionOperatorEnum.MULTIPLE, children);
+ }
+ return copyExpr;
+ }
+
+ @Override
+ public TupleExpression visitString(StringTupleExpression strExpr) {
+ return new StringTupleExpression(strExpr.getValue());
+ }
+
+ @Override
+ public TupleExpression visitColumn(ColumnTupleExpression colExpr) {
+ return new ColumnTupleExpression(colExpr.getColumn());
+ }
+
+ @Override
+ public TupleExpression visitBinary(BinaryTupleExpression binaryExpr) {
+ TupleExpression leftCopy;
+ TupleExpression rightCopy;
+
+ boolean ifToCntO = ifToCnt;
+ switch (binaryExpr.getOperator()) {
+ case PLUS:
+ case MINUS:
+ boolean ifCntSetO = ifCntSet;
+ leftCopy = binaryExpr.getLeft().accept(this);
+ ifToCnt = ifToCntO;
+ ifCntSet = ifCntSetO;
+ rightCopy = binaryExpr.getRight().accept(this);
+ break;
+ case MULTIPLE:
+ case DIVIDE:
+ if (ifToCntO) {
+ ifToCnt = ExpressionColCollector.collectMeasureColumns(binaryExpr.getRight()).isEmpty();
+ }
+ leftCopy = binaryExpr.getLeft().accept(this);
+
+ ifToCnt = ifToCntO && !ifCntSet;
+ if (ifToCnt) {
+ ifToCnt = ExpressionColCollector.collectMeasureColumns(binaryExpr.getLeft()).isEmpty();
+ }
+ ifCntSet = false;
+
+ rightCopy = binaryExpr.getRight().accept(this);
+ ifCntSet = ifToCntO && (ifCntSet || !ifToCnt);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported operator " + binaryExpr.getOperator());
+ }
+ return new BinaryTupleExpression(binaryExpr.getOperator(), Lists.newArrayList(leftCopy, rightCopy));
+ }
+
+ @Override
+ public TupleExpression visitCaseCall(CaseTupleExpression caseExpr) {
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists
+ .newArrayListWithExpectedSize(caseExpr.getWhenList().size());
+ for (Pair<TupleFilter, TupleExpression> entry : caseExpr.getWhenList()) {
+ TupleFilter filter = entry.getFirst();
+ TupleExpression expression = visitIndependent(entry.getSecond());
+ whenList.add(new Pair<>(filter, expression));
+ }
+ TupleExpression elseExpr = null;
+ if (caseExpr.getElseExpr() != null) {
+ elseExpr = visitIndependent(caseExpr.getElseExpr());
+ }
+
+ if (ifToCnt) {
+ ifToCnt = ExpressionColCollector.collectMeasureColumns(caseExpr).isEmpty();
+ }
+ return new CaseTupleExpression(whenList, elseExpr);
+ }
+
+ @Override
+ public TupleExpression visitRexCall(RexCallTupleExpression rexCallExpr) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TupleExpression visitNone(NoneTupleExpression noneExpr) {
+ return noneExpr;
+ }
+
+ private TupleExpression visitIndependent(TupleExpression expression) {
+ boolean ifToCntO = ifToCnt;
+ boolean ifCntSetO = ifCntSet;
+ TupleExpression ret = expression.accept(this);
+ ifToCnt = ifToCntO;
+ ifCntSet = ifCntSetO;
+ return ret;
+ }
+
+ public boolean ifCntSet() {
+ return ifCntSet;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionVisitor.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionVisitor.java
new file mode 100644
index 0000000000..ee4cd9f306
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/ExpressionVisitor.java
@@ -0,0 +1,35 @@
+/*
+ * 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.expression;
+
+public interface ExpressionVisitor {
+ TupleExpression visitNumber(NumberTupleExpression numExpr);
+
+ TupleExpression visitString(StringTupleExpression strExpr);
+
+ TupleExpression visitColumn(ColumnTupleExpression colExpr);
+
+ TupleExpression visitBinary(BinaryTupleExpression binaryExpr);
+
+ TupleExpression visitCaseCall(CaseTupleExpression caseExpr);
+
+ TupleExpression visitRexCall(RexCallTupleExpression rexCallExpr);
+
+ TupleExpression visitNone(NoneTupleExpression noneExpr);
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NoneTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NoneTupleExpression.java
new file mode 100644
index 0000000000..73124f392d
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NoneTupleExpression.java
@@ -0,0 +1,61 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+public class NoneTupleExpression extends TupleExpression {
+
+ public NoneTupleExpression() {
+ super(ExpressionOperatorEnum.NONE, Collections.<TupleExpression> emptyList());
+ }
+
+ @Override
+ public boolean ifAbleToPushDown() {
+ return false;
+ }
+
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public Object calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitNone(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NumberTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NumberTupleExpression.java
new file mode 100644
index 0000000000..5114e9a4d0
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/NumberTupleExpression.java
@@ -0,0 +1,95 @@
+/*
+ * 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.expression;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import org.apache.kylin.common.util.DecimalUtil;
+import org.apache.kylin.metadata.datatype.BigDecimalSerializer;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+public class NumberTupleExpression extends TupleExpression {
+
+ public static final BigDecimalSerializer serializer = new BigDecimalSerializer(DataType.getType("decimal"));
+
+ private BigDecimal value;
+
+ public NumberTupleExpression(Object value) {
+ this(DecimalUtil.toBigDecimal(value));
+ }
+
+ public NumberTupleExpression(BigDecimal value) {
+ super(ExpressionOperatorEnum.NUMBER, Collections.<TupleExpression> emptyList());
+ this.value = value;
+ }
+
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public BigDecimal calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ return value;
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitNumber(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ serializer.serialize(value, buffer);
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ value = serializer.deserialize(buffer);
+ }
+
+ public BigDecimal getValue() {
+ return value;
+ }
+
+ public String toString() {
+ return value.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ NumberTupleExpression that = (NumberTupleExpression) o;
+
+ return value != null ? value.equals(that.value) : that.value == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ return value != null ? value.hashCode() : 0;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/RexCallTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/RexCallTupleExpression.java
new file mode 100644
index 0000000000..736f6f22ed
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/RexCallTupleExpression.java
@@ -0,0 +1,61 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+public class RexCallTupleExpression extends TupleExpression {
+
+ public RexCallTupleExpression(List<TupleExpression> children) {
+ super(ExpressionOperatorEnum.REXCALL, children);
+ }
+
+ @Override
+ protected boolean ifAbleToPushDown() {
+ return false;
+ }
+
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public Object calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitRexCall(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/StringTupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/StringTupleExpression.java
new file mode 100644
index 0000000000..8c56521ce2
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/StringTupleExpression.java
@@ -0,0 +1,89 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.datatype.StringSerializer;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+
+public class StringTupleExpression extends TupleExpression {
+
+ public static final StringSerializer serializer = new StringSerializer(DataType.getType("varchar"));
+
+ private String value;
+
+ public StringTupleExpression(String value) {
+ super(ExpressionOperatorEnum.STRING, Collections.<TupleExpression> emptyList());
+ this.value = value;
+ }
+
+ @Override
+ public void verify() {
+ }
+
+ @Override
+ public String calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) {
+ return value;
+ }
+
+ @Override
+ public TupleExpression accept(ExpressionVisitor visitor) {
+ return visitor.visitString(this);
+ }
+
+ @Override
+ public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ serializer.serialize(value, buffer);
+ }
+
+ @Override
+ public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ value = serializer.deserialize(buffer);
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public String toString() {
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ StringTupleExpression that = (StringTupleExpression) o;
+
+ return value != null ? value.equals(that.value) : that.value == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ return value != null ? value.hashCode() : 0;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpression.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpression.java
new file mode 100644
index 0000000000..cd7e69ca65
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpression.java
@@ -0,0 +1,117 @@
+/*
+ * 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.expression;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class TupleExpression {
+ static final Logger logger = LoggerFactory.getLogger(TupleExpression.class);
+
+ public enum ExpressionOperatorEnum {
+ PLUS(0, "+"), MINUS(1, "-"), MULTIPLE(2, "*"), DIVIDE(3, "/"), //
+ CASE(10, "Case"), //
+ COLUMN(20, "InputRef"), NUMBER(21, "Number"), STRING(22, "String"), //
+ REXCALL(30, "RexCall"), NONE(31, "NONE");
+
+ private final int value;
+ private final String name;
+
+ private ExpressionOperatorEnum(int value, String name) {
+ this.value = value;
+ this.name = name;
+ }
+
+ public String toString() {
+ return name;
+ }
+
+ public int getValue() {
+ return value;
+ }
+ }
+
+ protected final ExpressionOperatorEnum operator;
+ protected final List<TupleExpression> children;
+ protected String digest;
+ protected Boolean ifAbleToPushDown = null;
+
+ protected TupleExpression(ExpressionOperatorEnum op, List<TupleExpression> exprs) {
+ this.operator = op;
+ this.children = exprs;
+ }
+
+ protected boolean ifAbleToPushDown() {
+ if (ifAbleToPushDown == null) {
+ for (TupleExpression child : children) {
+ ifAbleToPushDown = child.ifAbleToPushDown();
+ if (!ifAbleToPushDown) {
+ break;
+ }
+ }
+ if (ifAbleToPushDown == null) {
+ ifAbleToPushDown = true;
+ }
+ }
+ return ifAbleToPushDown;
+ }
+
+ public boolean ifForDynamicColumn() {
+ return false;
+ }
+
+ public abstract void verify();
+
+ public abstract Object calculate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs);
+
+ public abstract TupleExpression accept(ExpressionVisitor visitor);
+
+ public abstract void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer);
+
+ public abstract void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer);
+
+ public ExpressionOperatorEnum getOperator() {
+ return operator;
+ }
+
+ public void setDigest(String digest) {
+ this.digest = digest;
+ }
+
+ public String getDigest() {
+ return digest;
+ }
+
+ public boolean hasChildren() {
+ return children != null && !children.isEmpty();
+ }
+
+ public List<? extends TupleExpression> getChildren() {
+ return children;
+ }
+
+ public void addChild(TupleExpression child) {
+ children.add(child);
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpressionSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpressionSerializer.java
new file mode 100644
index 0000000000..435922e5ff
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/expression/TupleExpressionSerializer.java
@@ -0,0 +1,239 @@
+/*
+ * 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.expression;
+
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Stack;
+
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public class TupleExpressionSerializer {
+
+ private static final Logger logger = LoggerFactory.getLogger(TupleExpressionSerializer.class);
+
+ public interface Decorator {
+ TblColRef mapCol(TblColRef col);
+
+ TupleExpression convertInnerExpression(TupleExpression expression);
+
+ TupleFilter convertInnerFilter(TupleFilter filter);
+ }
+
+ private static class Serializer implements ExpressionVisitor {
+ private final Decorator decorator;
+ private final IFilterCodeSystem<?> cs;
+ private final ByteBuffer buffer;
+
+ private Serializer(Decorator decorator, IFilterCodeSystem<?> cs, ByteBuffer buffer) {
+ this.decorator = decorator;
+ this.cs = cs;
+ this.buffer = buffer;
+ }
+
+ @Override
+ public TupleExpression visitNumber(NumberTupleExpression numExpr) {
+ serializeExpression(0, numExpr, buffer, cs);
+ return numExpr;
+ }
+
+ public TupleExpression visitString(StringTupleExpression strExpr) {
+ serializeExpression(0, strExpr, buffer, cs);
+ return strExpr;
+ }
+
+ @Override
+ public TupleExpression visitColumn(ColumnTupleExpression colExpr) {
+ if (decorator != null) {
+ colExpr = new ColumnTupleExpression(decorator.mapCol(colExpr.getColumn()));
+ }
+ serializeExpression(0, colExpr, buffer, cs);
+ return colExpr;
+ }
+
+ @Override
+ public TupleExpression visitBinary(BinaryTupleExpression binaryExpr) {
+ // serialize expression+true
+ serializeExpression(1, binaryExpr, buffer, cs);
+ // serialize children
+ TupleExpression left = binaryExpr.getLeft().accept(this);
+ TupleExpression right = binaryExpr.getRight().accept(this);
+ // serialize none
+ serializeExpression(-1, binaryExpr, buffer, cs);
+ return decorator == null ? binaryExpr
+ : new BinaryTupleExpression(binaryExpr.getOperator(), Lists.newArrayList(left, right));
+ }
+
+ @Override
+ public TupleExpression visitCaseCall(CaseTupleExpression caseExpr) {
+ if (decorator != null) {
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists
+ .newArrayListWithExpectedSize(caseExpr.getWhenList().size());
+ for (Pair<TupleFilter, TupleExpression> entry : caseExpr.getWhenList()) {
+ TupleFilter filter = decorator.convertInnerFilter(entry.getFirst());
+ TupleExpression whenEntry = decorator.convertInnerExpression(entry.getSecond());
+ whenList.add(new Pair<>(filter, whenEntry));
+ }
+
+ TupleExpression elseExpr = caseExpr.getElseExpr();
+ if (elseExpr != null) {
+ elseExpr = decorator.convertInnerExpression(elseExpr);
+ }
+ caseExpr = new CaseTupleExpression(whenList, elseExpr);
+ }
+
+ serializeExpression(0, caseExpr, buffer, cs);
+ return caseExpr;
+ }
+
+ @Override
+ public TupleExpression visitRexCall(RexCallTupleExpression rexCallExpr) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TupleExpression visitNone(NoneTupleExpression noneExpr) {
+ return noneExpr;
+ }
+ }
+
+ private static final int BUFFER_SIZE = 65536;
+ private static final Map<Integer, TupleExpression.ExpressionOperatorEnum> ID_OP_MAP = Maps.newHashMap();
+
+ static {
+ for (TupleExpression.ExpressionOperatorEnum op : TupleExpression.ExpressionOperatorEnum.values()) {
+ ID_OP_MAP.put(op.getValue(), op);
+ }
+ }
+
+ public static byte[] serialize(TupleExpression rootExpr, IFilterCodeSystem<?> cs) {
+ return serialize(rootExpr, null, cs);
+ }
+
+ public static byte[] serialize(TupleExpression rootExpr, Decorator decorator, IFilterCodeSystem<?> cs) {
+ ByteBuffer buffer;
+ int bufferSize = BUFFER_SIZE;
+ while (true) {
+ try {
+ buffer = ByteBuffer.allocate(bufferSize);
+ Serializer serializer = new Serializer(decorator, cs, buffer);
+ rootExpr.accept(serializer);
+ break;
+ } catch (BufferOverflowException e) {
+ logger.info("Buffer size {} cannot hold the expression, resizing to 4 times", bufferSize);
+ bufferSize *= 4;
+ }
+ }
+ byte[] result = new byte[buffer.position()];
+ System.arraycopy(buffer.array(), 0, result, 0, buffer.position());
+ return result;
+ }
+
+ private static void serializeExpression(int flag, TupleExpression expr, ByteBuffer buffer,
+ IFilterCodeSystem<?> cs) {
+ if (flag < 0) {
+ BytesUtil.writeVInt(-1, buffer);
+ } else {
+ int opVal = expr.getOperator().getValue();
+ BytesUtil.writeVInt(opVal, buffer);
+ expr.serialize(cs, buffer);
+ BytesUtil.writeVInt(flag, buffer);
+ }
+ }
+
+ public static TupleExpression deserialize(byte[] bytes, IFilterCodeSystem<?> cs) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ TupleExpression rootTuple = null;
+ Stack<TupleExpression> parentStack = new Stack<>();
+ while (buffer.hasRemaining()) {
+ int opVal = BytesUtil.readVInt(buffer);
+ if (opVal < 0) {
+ parentStack.pop();
+ continue;
+ }
+
+ // deserialize expression
+ TupleExpression tuple = createTupleExpression(opVal);
+ tuple.deserialize(cs, buffer);
+
+ if (rootTuple == null) {
+ // push root to stack
+ rootTuple = tuple;
+ parentStack.push(tuple);
+ BytesUtil.readVInt(buffer);
+ continue;
+ }
+
+ // add expression to parent
+ TupleExpression parentExpression = parentStack.peek();
+ if (parentExpression != null) {
+ parentExpression.addChild(tuple);
+ }
+
+ // push expression to stack or not based on having children or not
+ int hasChild = BytesUtil.readVInt(buffer);
+ if (hasChild == 1) {
+ parentStack.push(tuple);
+ }
+ }
+ return rootTuple;
+ }
+
+ private static TupleExpression createTupleExpression(int opVal) {
+ TupleExpression.ExpressionOperatorEnum op = ID_OP_MAP.get(opVal);
+ if (op == null) {
+ throw new IllegalStateException("operator value is " + opVal);
+ }
+ TupleExpression tuple = null;
+ switch (op) {
+ case PLUS:
+ case MINUS:
+ case MULTIPLE:
+ case DIVIDE:
+ tuple = new BinaryTupleExpression(op);
+ break;
+ case NUMBER:
+ tuple = new NumberTupleExpression(null);
+ break;
+ case STRING:
+ tuple = new StringTupleExpression(null);
+ break;
+ case COLUMN:
+ tuple = new ColumnTupleExpression(null);
+ break;
+ case CASE:
+ tuple = new CaseTupleExpression(null, null);
+ break;
+ default:
+ throw new IllegalStateException("Error ExpressionOperatorEnum: " + op.getValue());
+ }
+ return tuple;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java
index 4875217cbd..bd2baec667 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java
@@ -273,4 +273,41 @@ public TupleFilter acceptOptimizeTransformer(FilterOptimizeTransformer transform
return transformer.visit(this);
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ CompareTupleFilter that = (CompareTupleFilter) o;
+
+ if (operator != that.operator)
+ return false;
+ if (column != null ? !column.equals(that.column) : that.column != null)
+ return false;
+ if (function != null ? !function.equals(that.function) : that.function != null)
+ return false;
+ if (secondColumn != null ? !secondColumn.equals(that.secondColumn) : that.secondColumn != null)
+ return false;
+ if (conditionValues != null ? !conditionValues.equals(that.conditionValues) : that.conditionValues != null)
+ return false;
+ if (firstCondValue != null ? !firstCondValue.equals(that.firstCondValue) : that.firstCondValue != null)
+ return false;
+ return dynamicVariables != null ? dynamicVariables.equals(that.dynamicVariables)
+ : that.dynamicVariables == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = operator != null ? operator.hashCode() : 0;
+ result = 31 * result + (column != null ? column.hashCode() : 0);
+ result = 31 * result + (function != null ? function.hashCode() : 0);
+ result = 31 * result + (secondColumn != null ? secondColumn.hashCode() : 0);
+ result = 31 * result + (conditionValues != null ? conditionValues.hashCode() : 0);
+ result = 31 * result + (firstCondValue != null ? firstCondValue.hashCode() : 0);
+ result = 31 * result + (dynamicVariables != null ? dynamicVariables.hashCode() : 0);
+ return result;
+ }
}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/DynamicFunctionDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/DynamicFunctionDesc.java
new file mode 100644
index 0000000000..a41477afd3
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/DynamicFunctionDesc.java
@@ -0,0 +1,85 @@
+/*
+ * 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.model;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.expression.ExpressionColCollector;
+import org.apache.kylin.metadata.expression.TupleExpression;
+
+import com.google.common.collect.Maps;
+
+public abstract class DynamicFunctionDesc extends FunctionDesc {
+
+ protected final TupleExpression tupleExpression;
+ protected final Set<TblColRef> filterColSet;
+ protected Map<TblColRef, FunctionDesc> runtimeFuncMap;
+
+ public DynamicFunctionDesc(ParameterDesc parameter, TupleExpression tupleExpression) {
+ this.setParameter(parameter);
+ this.tupleExpression = tupleExpression;
+
+ Pair<Set<TblColRef>, Set<TblColRef>> colsPair = ExpressionColCollector.collectColumnsPair(tupleExpression);
+ filterColSet = colsPair.getFirst();
+ Set<TblColRef> measureColumns = colsPair.getSecond();
+ this.runtimeFuncMap = Maps.newHashMapWithExpectedSize(measureColumns.size());
+ for (TblColRef column : measureColumns) {
+ runtimeFuncMap.put(column, constructRuntimeFunction(column));
+ }
+ }
+
+ @Override
+ public boolean needRewrite() {
+ return false;
+ }
+
+ // TODO: this should be referred by the filters in tupleExpression
+ public boolean ifFriendlyForDerivedFilter() {
+ return false;
+ }
+
+ public TupleExpression getTupleExpression() {
+ return tupleExpression;
+ }
+
+ public Set<TblColRef> getFilterColumnSet() {
+ return filterColSet;
+ }
+
+ public Set<TblColRef> getMeasureColumnSet() {
+ return runtimeFuncMap.keySet();
+ }
+
+ public Collection<FunctionDesc> getRuntimeFuncs() {
+ return runtimeFuncMap.values();
+ }
+
+ public Map<TblColRef, FunctionDesc> getRuntimeFuncMap() {
+ return runtimeFuncMap;
+ }
+
+ public void setRuntimeFuncMap(Map<TblColRef, FunctionDesc> funcMap) {
+ this.runtimeFuncMap = funcMap;
+ }
+
+ protected abstract FunctionDesc constructRuntimeFunction(TblColRef column);
+}
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 7597d407d8..d93ada403f 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
@@ -272,6 +272,7 @@ public String getReturnType() {
public void setReturnType(String returnType) {
this.returnType = returnType;
+ this.returnDataType = DataType.getType(returnType);
}
public DataType getReturnDataType() {
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/SumDynamicFunctionDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/SumDynamicFunctionDesc.java
new file mode 100644
index 0000000000..f29c44d3f0
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/SumDynamicFunctionDesc.java
@@ -0,0 +1,77 @@
+/*
+ * 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.model;
+
+import java.util.Set;
+
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.expression.TupleExpression;
+
+import com.google.common.collect.Sets;
+
+public class SumDynamicFunctionDesc extends DynamicFunctionDesc {
+
+ public static final TblColRef mockCntCol = TblColRef.newInnerColumn(FunctionDesc.FUNC_COUNT,
+ TblColRef.InnerDataTypeEnum.DERIVED);
+
+ private Set<TblColRef> measureColumnSet;
+
+ public SumDynamicFunctionDesc(ParameterDesc parameter, TupleExpression tupleExpression) {
+ super(parameter, tupleExpression);
+ setExpression(FUNC_SUM);
+ setReturnType("decimal");
+ }
+
+ @Override
+ public String getRewriteFieldName() {
+ return "_KY_" + FUNC_SUM + "_" + tupleExpression.toString();
+ }
+
+ @Override
+ public DataType getRewriteFieldType() {
+ return getReturnDataType();
+ }
+
+ @Override
+ public Set<TblColRef> getMeasureColumnSet() {
+ if (measureColumnSet == null) {
+ measureColumnSet = Sets.newHashSet(super.getMeasureColumnSet());
+ measureColumnSet.remove(mockCntCol);
+ }
+ return measureColumnSet;
+ }
+
+ @Override
+ protected FunctionDesc constructRuntimeFunction(TblColRef column) {
+ return column == mockCntCol ? constructCountFunction() : constructSumFunction(column);
+ }
+
+ private FunctionDesc constructCountFunction() {
+ return FunctionDesc.newInstance(FunctionDesc.FUNC_COUNT, null, null);
+ }
+
+ private FunctionDesc constructSumFunction(TblColRef column) {
+ FunctionDesc function = new FunctionDesc();
+ function.setParameter(ParameterDesc.newInstance(column));
+ function.setExpression(FUNC_SUM);
+ function.setReturnType("decimal");
+
+ return function;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java
index dbe517e166..0b23e48eed 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java
@@ -19,9 +19,12 @@
package org.apache.kylin.metadata.realization;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.JoinDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -56,11 +59,17 @@ public SQLCall(String function, Iterable<Object> args) {
public List<TblColRef> groupbyColumns;
public Set<TblColRef> subqueryJoinParticipants;
+ public Map<TblColRef, TupleExpression> dynGroupbyColumns;
+
// aggregation
public Set<TblColRef> metricColumns;
public List<FunctionDesc> aggregations; // storage level measure type, on top of which various sql aggr function may apply
public List<SQLCall> aggrSqlCalls; // sql level aggregation function call
+ public List<DynamicFunctionDesc> dynAggregations;
+ public Set<TblColRef> rtDimensionColumns; // dynamic col related dimension columns
+ public Set<TblColRef> rtMetricColumns; // dynamic col related metric columns
+
// filter
public Set<TblColRef> filterColumns;
public TupleFilter filter;
@@ -75,8 +84,11 @@ public SQLCall(String function, Iterable<Object> args) {
public Set<MeasureDesc> involvedMeasure;
public SQLDigest(String factTable, Set<TblColRef> allColumns, List<JoinDesc> joinDescs, // model
- List<TblColRef> groupbyColumns, Set<TblColRef> subqueryJoinParticipants, // group by
+ List<TblColRef> groupbyColumns, Set<TblColRef> subqueryJoinParticipants,
+ Map<TblColRef, TupleExpression> dynGroupByColumns, // group by
Set<TblColRef> metricColumns, List<FunctionDesc> aggregations, List<SQLCall> aggrSqlCalls, // aggregation
+ List<DynamicFunctionDesc> dynAggregations, //
+ Set<TblColRef> rtDimensionColumns, Set<TblColRef> rtMetricColumns, // dynamic col related columns
Set<TblColRef> filterColumns, TupleFilter filter, TupleFilter havingFilter, // filter
List<TblColRef> sortColumns, List<OrderEnum> sortOrders, boolean limitPrecedesAggr, // sort & limit
Set<MeasureDesc> involvedMeasure
@@ -88,10 +100,17 @@ public SQLDigest(String factTable, Set<TblColRef> allColumns, List<JoinDesc> joi
this.groupbyColumns = groupbyColumns;
this.subqueryJoinParticipants = subqueryJoinParticipants;
+ this.dynGroupbyColumns = dynGroupByColumns;
+
this.metricColumns = metricColumns;
this.aggregations = aggregations;
this.aggrSqlCalls = aggrSqlCalls;
+ this.dynAggregations = dynAggregations;
+
+ this.rtDimensionColumns = rtDimensionColumns;
+ this.rtMetricColumns = rtMetricColumns;
+
this.filterColumns = filterColumns;
this.filter = filter;
this.havingFilter = havingFilter;
diff --git a/core-metadata/src/test/java/org/apache/kylin/metadata/datatype/BigDecimalSerializerTest.java b/core-metadata/src/test/java/org/apache/kylin/metadata/datatype/BigDecimalSerializerTest.java
index 5be5806c67..a21b08e4a2 100644
--- a/core-metadata/src/test/java/org/apache/kylin/metadata/datatype/BigDecimalSerializerTest.java
+++ b/core-metadata/src/test/java/org/apache/kylin/metadata/datatype/BigDecimalSerializerTest.java
@@ -74,4 +74,14 @@ public void testOutOfPrecision() {
bigDecimalSerializer.serialize(input, buffer);
}
+ @Test
+ public void testNull() {
+ BigDecimal input = null;
+ ByteBuffer buffer = ByteBuffer.allocate(256);
+ buffer.mark();
+ bigDecimalSerializer.serialize(input, buffer);
+ buffer.reset();
+ BigDecimal output = bigDecimalSerializer.deserialize(buffer);
+ assertEquals(input, output);
+ }
}
diff --git a/core-metadata/src/test/java/org/apache/kylin/metadata/expression/ExpressionCountDistributorTest.java b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/ExpressionCountDistributorTest.java
new file mode 100644
index 0000000000..73424755aa
--- /dev/null
+++ b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/ExpressionCountDistributorTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.expression;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.expression.TupleExpression.ExpressionOperatorEnum;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+import org.apache.kylin.metadata.filter.IFilterCodeSystem;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.powermock.api.mockito.PowerMockito;
+
+import com.google.common.collect.Lists;
+
+public class ExpressionCountDistributorTest extends LocalFileMetadataTestCase {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ staticCleanupTestMetadata();
+ }
+
+ /**
+ * * 3 10
+ * * | |
+ * * 1*(1+2)*(col+1)*(3+4)+(1+2)*(3+1)*(4+5)-4+5 => 1*(1+2)*(col+1*n)*(3+4)+((1+2)*(3+1)*(4+5)-4+5)*n = 1363
+ * <p>
+ * * + +
+ * * / \ / \
+ * * - 5 - *
+ * * / \ / \ / \
+ * * + 4 + * n 5
+ * * / \ / \ / \
+ * * * * ===> * * n 4
+ * * / \ / \ / \ / \
+ * * * + * + * + * +
+ * * / \ / \ / \ / \ / \ / \ / \ / \
+ * * * + 3 4 + + 4 5 * + 3 4 + + 4 5
+ * * / \ / \ / \ / \ / \ / \ / \ / \
+ * * 1 + col 1 1 2 3 1 1 + col * * * 3 1
+ * * / \ / \ / \ / \ / \
+ * * 1 2 1 2 n 1 n 1 n 2
+ */
+ @Test
+ public void testDistribute1() {
+ NumberTupleExpression n = new NumberTupleExpression(10);
+ ExpressionCountDistributor cntDistributor = new ExpressionCountDistributor(n);
+
+ TupleExpression t0 = new NumberTupleExpression(1);
+ TupleExpression t1 = new NumberTupleExpression(1);
+ TupleExpression t2 = new NumberTupleExpression(2);
+
+ TblColRef c = PowerMockito.mock(TblColRef.class);
+ TupleExpression t3 = new ColumnTupleExpression(c);
+ IEvaluatableTuple evaluatableTuple = PowerMockito.mock(IEvaluatableTuple.class);
+ IFilterCodeSystem filterCodeSystem = PowerMockito.mock(IFilterCodeSystem.class);
+ t3 = PowerMockito.spy(t3);
+ PowerMockito.when(t3.calculate(evaluatableTuple, filterCodeSystem)).thenReturn(new BigDecimal(3));
+
+ TupleExpression t4 = new NumberTupleExpression(1);
+ TupleExpression t5 = new NumberTupleExpression(3);
+ TupleExpression t6 = new NumberTupleExpression(4);
+ TupleExpression t7 = new NumberTupleExpression(1);
+ TupleExpression t8 = new NumberTupleExpression(2);
+ TupleExpression t9 = new NumberTupleExpression(3);
+ TupleExpression t10 = new NumberTupleExpression(1);
+ TupleExpression t11 = new NumberTupleExpression(4);
+ TupleExpression t12 = new NumberTupleExpression(5);
+ TupleExpression t13 = new NumberTupleExpression(4);
+ TupleExpression t14 = new NumberTupleExpression(5);
+
+ TupleExpression b0 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t1, t2));
+
+ TupleExpression b1 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(t0, b0));
+ TupleExpression b2 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t3, t4));
+ TupleExpression b3 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t7, t8));
+ TupleExpression b4 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t9, t10));
+
+ TupleExpression b11 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b1, b2));
+ TupleExpression b12 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t5, t6));
+ TupleExpression b13 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b3, b4));
+ TupleExpression b14 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t11, t12));
+
+ TupleExpression b21 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b11, b12));
+ TupleExpression b22 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b13, b14));
+
+ TupleExpression b31 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b21, b22));
+
+ TupleExpression b41 = new BinaryTupleExpression(ExpressionOperatorEnum.MINUS, Lists.newArrayList(b31, t13));
+
+ TupleExpression b51 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b41, t14));
+
+ TupleExpression ret = b51.accept(cntDistributor);
+ assertTrue(cntDistributor.ifCntSet());
+
+ assertEquals(new BigDecimal(1363), ret.calculate(evaluatableTuple, filterCodeSystem));
+ }
+
+ /**
+ * * 3 10
+ * * | |
+ * * (1+2)*(case when f1 = 'c1' then (1+2)*(col+1)+3 (1+2)*(case when f1 = 'c1' then (1+2)*(col+n*1)+n*3
+ * * when f1 = 'c2' then (2+col)*(2+3)+4 => when f1 = 'c2' then (n*2+col)*(2+3)+n*4
+ * * else 6 else n*6
+ * * end) + col*2 + 1 end) + col*2 + n*1
+ */
+ @Test
+ public void testDistribute2() {
+ NumberTupleExpression n = new NumberTupleExpression(10);
+ ExpressionCountDistributor cntDistributor = new ExpressionCountDistributor(n);
+
+ TupleExpression t1 = new NumberTupleExpression(1);
+ TupleExpression t2 = new NumberTupleExpression(2);
+
+ TupleExpression t3 = new NumberTupleExpression(1);
+ TupleExpression t4 = new NumberTupleExpression(2);
+
+ TblColRef c = PowerMockito.mock(TblColRef.class);
+ TupleExpression t5 = new ColumnTupleExpression(c);
+ IEvaluatableTuple evaluatableTuple = PowerMockito.mock(IEvaluatableTuple.class);
+ IFilterCodeSystem filterCodeSystem = PowerMockito.mock(IFilterCodeSystem.class);
+ t5 = PowerMockito.spy(t5);
+ PowerMockito.when(t5.calculate(evaluatableTuple, filterCodeSystem)).thenReturn(new BigDecimal(3));
+
+ TupleExpression t6 = new NumberTupleExpression(1);
+ TupleExpression t7 = new NumberTupleExpression(3);
+
+ TupleExpression t8 = new NumberTupleExpression(2);
+ TupleExpression t9 = new NumberTupleExpression(2);
+ TupleExpression t10 = new NumberTupleExpression(3);
+ TupleExpression t11 = new NumberTupleExpression(4);
+
+ TupleExpression t12 = new NumberTupleExpression(6);
+
+ TupleExpression t13 = new NumberTupleExpression(2);
+ TupleExpression t14 = new NumberTupleExpression(1);
+
+ TupleExpression b1 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t1, t2));
+ TupleExpression b2 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t3, t4));
+ TupleExpression b3 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t5, t6));
+ TupleExpression b4 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t8, t5));
+ TupleExpression b5 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(t9, t10));
+ TupleExpression b6 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(t5, t13));
+
+ TupleExpression b11 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b2, b3));
+ TupleExpression b12 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b4, b5));
+
+ TupleExpression b21 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b11, t7));
+ TupleExpression b22 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b12, t11));
+
+ TupleFilter f1 = PowerMockito.mock(CompareTupleFilter.class);
+ TupleFilter f2 = PowerMockito.mock(CompareTupleFilter.class);
+
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists.newArrayList();
+ whenList.add(new Pair<>(f1, b21));
+ whenList.add(new Pair<>(f2, b22));
+ TupleExpression b31 = new CaseTupleExpression(whenList, t12);
+
+ TupleExpression b41 = new BinaryTupleExpression(ExpressionOperatorEnum.MULTIPLE, Lists.newArrayList(b1, b31));
+
+ TupleExpression b51 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b41, b6));
+
+ TupleExpression b61 = new BinaryTupleExpression(ExpressionOperatorEnum.PLUS, Lists.newArrayList(b51, t14));
+
+ TupleExpression ret = b61.accept(cntDistributor);
+ assertTrue(cntDistributor.ifCntSet());
+
+ PowerMockito.when(f1.evaluate(evaluatableTuple, filterCodeSystem)).thenReturn(true);
+ assertEquals(new BigDecimal(223), ret.calculate(evaluatableTuple, filterCodeSystem));
+
+ PowerMockito.when(f1.evaluate(evaluatableTuple, filterCodeSystem)).thenReturn(false);
+ PowerMockito.when(f2.evaluate(evaluatableTuple, filterCodeSystem)).thenReturn(true);
+ assertEquals(new BigDecimal(481), ret.calculate(evaluatableTuple, filterCodeSystem));
+
+ PowerMockito.when(f1.evaluate(evaluatableTuple, filterCodeSystem)).thenReturn(false);
+ PowerMockito.when(f2.evaluate(evaluatableTuple, filterCodeSystem)).thenReturn(false);
+ assertEquals(new BigDecimal(196), ret.calculate(evaluatableTuple, filterCodeSystem));
+ }
+}
diff --git a/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionSerializerTest.java b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionSerializerTest.java
new file mode 100644
index 0000000000..9797fa7276
--- /dev/null
+++ b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionSerializerTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.expression;
+
+import static org.junit.Assert.assertEquals;
+
+import java.math.BigDecimal;
+
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.filter.ColumnTupleFilter;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+import org.apache.kylin.metadata.filter.ConstantTupleFilter;
+import org.apache.kylin.metadata.filter.StringCodeSystem;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class TupleExpressionSerializerTest extends LocalFileMetadataTestCase {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ staticCleanupTestMetadata();
+ }
+
+ private TableDesc t = TableDesc.mockup("T");
+
+ @Test
+ public void testSerialization() {
+ TblColRef colD = TblColRef.mockup(t, 1, "C1", "decimal");
+ TblColRef colM = TblColRef.mockup(t, 2, "C2", "string");
+ BigDecimal value = BigDecimal.valueOf(10L);
+
+ ColumnTupleFilter colFilter = new ColumnTupleFilter(colD);
+ ConstantTupleFilter constFilter = new ConstantTupleFilter("col");
+ CompareTupleFilter compareFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.EQ);
+ compareFilter.addChild(colFilter);
+ compareFilter.addChild(constFilter);
+
+ ColumnTupleExpression colTuple = new ColumnTupleExpression(colM);
+ NumberTupleExpression constTuple = new NumberTupleExpression(value);
+
+ Pair<TupleFilter, TupleExpression> whenEntry = new Pair<TupleFilter, TupleExpression>(compareFilter, colTuple);
+ CaseTupleExpression caseTuple = new CaseTupleExpression(Lists.newArrayList(whenEntry), constTuple);
+
+ byte[] result = TupleExpressionSerializer.serialize(caseTuple, StringCodeSystem.INSTANCE);
+
+ TupleExpression desTuple = TupleExpressionSerializer.deserialize(result, StringCodeSystem.INSTANCE);
+ assertEquals(caseTuple, desTuple);
+ }
+}
diff --git a/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionTest.java b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionTest.java
new file mode 100644
index 0000000000..7617c5a4d7
--- /dev/null
+++ b/core-metadata/src/test/java/org/apache/kylin/metadata/expression/TupleExpressionTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.expression;
+
+import static org.junit.Assert.fail;
+
+import java.math.BigDecimal;
+
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class TupleExpressionTest extends LocalFileMetadataTestCase {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ staticCleanupTestMetadata();
+ }
+
+ private TableDesc t = TableDesc.mockup("T");
+
+ @Test
+ public void testBinary() {
+ BigDecimal value1 = BigDecimal.valueOf(10L);
+ BigDecimal value2 = BigDecimal.valueOf(10L);
+ TblColRef col1 = TblColRef.mockup(t, 1, "C1", "decimal");
+ TblColRef col2 = TblColRef.mockup(t, 2, "C2", "decimal");
+
+ NumberTupleExpression constTuple1 = new NumberTupleExpression(value1);
+ NumberTupleExpression constTuple2 = new NumberTupleExpression(value2);
+ ColumnTupleExpression colTuple1 = new ColumnTupleExpression(col1);
+ ColumnTupleExpression colTuple2 = new ColumnTupleExpression(col2);
+
+ BinaryTupleExpression biTuple1 = new BinaryTupleExpression(TupleExpression.ExpressionOperatorEnum.MULTIPLE,
+ Lists.newArrayList(constTuple1, colTuple1));
+ biTuple1.verify();
+
+ BinaryTupleExpression biTuple2 = new BinaryTupleExpression(TupleExpression.ExpressionOperatorEnum.DIVIDE,
+ Lists.newArrayList(constTuple2, colTuple2));
+ try {
+ biTuple2.verify();
+ fail("IllegalArgumentException should be thrown");
+ } catch (IllegalArgumentException e) {
+ }
+
+ biTuple2 = new BinaryTupleExpression(TupleExpression.ExpressionOperatorEnum.DIVIDE,
+ Lists.newArrayList(colTuple2, constTuple2));
+ biTuple2.verify();
+
+ BinaryTupleExpression biTuple = new BinaryTupleExpression(TupleExpression.ExpressionOperatorEnum.MULTIPLE,
+ Lists.<TupleExpression> newArrayList(biTuple1, biTuple2));
+ try {
+ biTuple.verify();
+ fail("IllegalArgumentException should be thrown");
+ } catch (IllegalArgumentException e) {
+ }
+ }
+}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java b/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
index 501dff8b59..5d2d06fb0f 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
@@ -23,6 +23,7 @@
import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.gridtable.StorageLimitLevel;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.storage.gtrecord.GTCubeStorageQueryBase;
@@ -56,6 +57,7 @@
private IStorageQuery storageQuery;
private AtomicLong processedRowCount = new AtomicLong();
private Cuboid cuboid;
+ private CuboidToGridTableMapping mapping;
private boolean partialResultReturned = false;
private Range<Long> reusedPeriod;
@@ -189,6 +191,14 @@ public Cuboid getCuboid() {
return cuboid;
}
+ public CuboidToGridTableMapping getMapping() {
+ return mapping;
+ }
+
+ public void setMapping(CuboidToGridTableMapping mapping) {
+ this.mapping = mapping;
+ }
+
public long getProcessedRowCount() {
return processedRowCount.get();
}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeScanRangePlanner.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeScanRangePlanner.java
index 5572e44f83..5f86a45534 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeScanRangePlanner.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeScanRangePlanner.java
@@ -19,6 +19,7 @@
package org.apache.kylin.storage.gtrecord;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
@@ -29,6 +30,7 @@
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.common.FuzzyValueCombination;
@@ -47,7 +49,9 @@
import org.apache.kylin.gridtable.GTScanRequestBuilder;
import org.apache.kylin.gridtable.GTUtil;
import org.apache.kylin.gridtable.IGTComparator;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.StorageContext;
@@ -71,8 +75,10 @@
protected CubeDesc cubeDesc;
protected Cuboid cuboid;
- public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter filter, Set<TblColRef> dimensions, Set<TblColRef> groupByDims, //
- Collection<FunctionDesc> metrics, TupleFilter havingFilter, StorageContext context) {
+ public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter filter, Set<TblColRef> dimensions, //
+ Set<TblColRef> groupByDims, List<TblColRef> dynGroupsDims, List<TupleExpression> dynGroupExprs, //
+ Collection<FunctionDesc> metrics, List<DynamicFunctionDesc> dynFuncs, //
+ TupleFilter havingFilter, StorageContext context) {
this.context = context;
this.maxScanRanges = cubeSegment.getConfig().getQueryStorageVisitScanRangeMax();
@@ -83,11 +89,9 @@ public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter
this.cubeDesc = cubeSegment.getCubeDesc();
this.cuboid = cuboid;
- Set<TblColRef> filterDims = Sets.newHashSet();
- TupleFilter.collectColumns(filter, filterDims);
+ final CuboidToGridTableMapping mapping = context.getMapping();
- this.gtInfo = CubeGridTable.newGTInfo(cuboid, new CubeDimEncMap(cubeSegment));
- CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
+ this.gtInfo = CubeGridTable.newGTInfo(cuboid, new CubeDimEncMap(cubeSegment), mapping);
IGTComparator comp = gtInfo.getCodeSystem().getComparator();
//start key GTRecord compare to start key GTRecord
@@ -99,7 +103,8 @@ public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter
//replace the constant values in filter to dictionary codes
Set<TblColRef> groupByPushDown = Sets.newHashSet(groupByDims);
- this.gtFilter = GTUtil.convertFilterColumnsAndConstants(filter, gtInfo, mapping.getCuboidDimensionsInGTOrder(), groupByPushDown);
+ groupByPushDown.addAll(dynGroupsDims);
+ this.gtFilter = GTUtil.convertFilterColumnsAndConstants(filter, gtInfo, mapping.getDim2gt(), groupByPushDown);
this.havingFilter = havingFilter;
this.gtDimensions = mapping.makeGridTableColumns(dimensions);
@@ -107,6 +112,31 @@ public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter
this.gtAggrMetrics = mapping.makeGridTableColumns(metrics);
this.gtAggrFuncs = mapping.makeAggrFuncs(metrics);
+ // for dynamic cols, which are as appended columns to GTInfo
+ BitSet tmpGtDynCols = new BitSet();
+
+ this.tupleExpressionMap = Maps.newHashMap();
+ // for dynamic dimensions
+ for (int i = 0; i < dynGroupsDims.size(); i++) {
+ int c = mapping.getIndexOf(dynGroupsDims.get(i));
+ tmpGtDynCols.set(c);
+ TupleExpression tupleExpr = GTUtil.convertFilterColumnsAndConstants(dynGroupExprs.get(i), gtInfo, mapping,
+ groupByPushDown);
+ this.tupleExpressionMap.put(c, tupleExpr);
+ }
+
+ // for dynamic measures
+ Set<FunctionDesc> tmpRtAggrMetrics = Sets.newHashSet();
+ for (DynamicFunctionDesc dynFunc : dynFuncs) {
+ tmpRtAggrMetrics.addAll(dynFunc.getRuntimeFuncs());
+ int c = mapping.getIndexOf(dynFunc);
+ tmpGtDynCols.set(c);
+ this.tupleExpressionMap.put(c, GTUtil.convertFilterColumnsAndConstants(dynFunc.getTupleExpression(), gtInfo,
+ mapping, dynFunc.getRuntimeFuncMap(), groupByPushDown));
+ }
+ this.gtDynColumns = new ImmutableBitSet(tmpGtDynCols);
+ this.gtRtAggrMetrics = mapping.makeGridTableColumns(tmpRtAggrMetrics);
+
if (cubeSegment.getModel().getPartitionDesc().isPartitioned()) {
int index = mapping.getIndexOf(cubeSegment.getModel().getPartitionDesc().getPartitionDateColumnRef());
if (index >= 0) {
@@ -148,10 +178,16 @@ public GTScanRequest planScanRequest() {
GTScanRequest scanRequest;
List<GTScanRange> scanRanges = this.planScanRanges();
if (scanRanges != null && scanRanges.size() != 0) {
- scanRequest = new GTScanRequestBuilder().setInfo(gtInfo).setRanges(scanRanges).setDimensions(gtDimensions).//
- setAggrGroupBy(gtAggrGroups).setAggrMetrics(gtAggrMetrics).setAggrMetricsFuncs(gtAggrFuncs).setFilterPushDown(gtFilter).//
- setAllowStorageAggregation(context.isNeedStorageAggregation()).setAggCacheMemThreshold(cubeSegment.getConfig().getQueryCoprocessorMemGB()).//
- setStoragePushDownLimit(context.getFinalPushDownLimit()).setStorageLimitLevel(context.getStorageLimitLevel()).setHavingFilterPushDown(havingFilter).createGTScanRequest();
+ scanRequest = new GTScanRequestBuilder().setInfo(gtInfo).setRanges(scanRanges).setDimensions(gtDimensions)
+ .setAggrGroupBy(gtAggrGroups).setAggrMetrics(gtAggrMetrics).setAggrMetricsFuncs(gtAggrFuncs)
+ .setFilterPushDown(gtFilter)//
+ .setRtAggrMetrics(gtRtAggrMetrics).setDynamicColumns(gtDynColumns)
+ .setExprsPushDown(tupleExpressionMap)//
+ .setAllowStorageAggregation(context.isNeedStorageAggregation())
+ .setAggCacheMemThreshold(cubeSegment.getConfig().getQueryCoprocessorMemGB())//
+ .setStoragePushDownLimit(context.getFinalPushDownLimit())
+ .setStorageLimitLevel(context.getStorageLimitLevel()).setHavingFilterPushDown(havingFilter)
+ .createGTScanRequest();
} else {
scanRequest = null;
}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeSegmentScanner.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeSegmentScanner.java
index ee127439f2..95ffa35335 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeSegmentScanner.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/CubeSegmentScanner.java
@@ -21,6 +21,7 @@
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
+import java.util.List;
import java.util.Set;
import org.apache.kylin.cube.CubeSegment;
@@ -30,10 +31,12 @@
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.ITupleFilterTransformer;
import org.apache.kylin.metadata.filter.StringCodeSystem;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilterSerializer;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.StorageContext;
@@ -50,11 +53,13 @@
final GTScanRequest scanRequest;
- public CubeSegmentScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dimensions, Set<TblColRef> groups, //
- Collection<FunctionDesc> metrics, TupleFilter originalfilter, TupleFilter havingFilter, StorageContext context) {
-
+ public CubeSegmentScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dimensions, //
+ Set<TblColRef> groups, List<TblColRef> dynGroups, List<TupleExpression> dynGroupExprs, //
+ Collection<FunctionDesc> metrics, List<DynamicFunctionDesc> dynFuncs, //
+ TupleFilter originalfilter, TupleFilter havingFilter, StorageContext context) {
+
logger.info("Init CubeSegmentScanner for segment {}", cubeSeg.getName());
-
+
this.cuboid = cuboid;
this.cubeSeg = cubeSeg;
@@ -70,19 +75,20 @@ public CubeSegmentScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dim
CubeScanRangePlanner scanRangePlanner;
try {
- scanRangePlanner = new CubeScanRangePlanner(cubeSeg, cuboid, filter, dimensions, groups, metrics, havingFilter, context);
+ scanRangePlanner = new CubeScanRangePlanner(cubeSeg, cuboid, filter, dimensions, groups, dynGroups,
+ dynGroupExprs, metrics, dynFuncs, havingFilter, context);
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
-
+
scanRequest = scanRangePlanner.planScanRequest();
-
+
String gtStorage = ((GTCubeStorageQueryBase) context.getStorageQuery()).getGTStorage();
scanner = new ScannerWorker(cubeSeg, cuboid, scanRequest, gtStorage, context);
}
-
+
public boolean isSegmentSkipped() {
return scanner.isSegmentSkipped();
}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryBase.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryBase.java
index 2f69b76900..4eb3a58a86 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryBase.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryBase.java
@@ -33,6 +33,8 @@
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.RawQueryLastHacker;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMappingExt;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeDesc.DeriveInfo;
import org.apache.kylin.cube.model.RowKeyColDesc;
@@ -40,12 +42,14 @@
import org.apache.kylin.gridtable.StorageLimitLevel;
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.measure.bitmap.BitmapMeasureType;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.CaseTupleFilter;
import org.apache.kylin.metadata.filter.ColumnTupleFilter;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.LogicalTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.PartitionDesc;
@@ -90,8 +94,10 @@ public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleI
continue;
}
- scanner = new CubeSegmentScanner(cubeSeg, request.getCuboid(), request.getDimensions(), request.getGroups(),
- request.getMetrics(), request.getFilter(), request.getHavingFilter(), request.getContext());
+ scanner = new CubeSegmentScanner(cubeSeg, request.getCuboid(), request.getDimensions(), //
+ request.getGroups(), request.getDynGroups(), request.getDynGroupExprs(), //
+ request.getMetrics(), request.getDynFuncs(), //
+ request.getFilter(), request.getHavingFilter(), request.getContext());
if (!scanner.isSegmentSkipped())
scanners.add(scanner);
}
@@ -100,7 +106,7 @@ public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleI
return ITupleIterator.EMPTY_TUPLE_ITERATOR;
return new SequentialCubeTupleIterator(scanners, request.getCuboid(), request.getDimensions(),
- request.getGroups(), request.getMetrics(), returnTupleInfo, request.getContext(), sqlDigest);
+ request.getDynGroups(), request.getGroups(), request.getMetrics(), returnTupleInfo, request.getContext(), sqlDigest);
}
public GTCubeStorageQueryRequest getStorageQueryRequest(StorageContext context, SQLDigest sqlDigest,
@@ -138,6 +144,23 @@ public GTCubeStorageQueryRequest getStorageQueryRequest(StorageContext context,
Cuboid cuboid = findCuboid(cubeInstance, dimensionsD, metrics);
context.setCuboid(cuboid);
+ // set cuboid to GridTable mapping;
+ boolean noDynamicCols;
+ // dynamic dimensions
+ List<TblColRef> dynGroups = Lists.newArrayList(sqlDigest.dynGroupbyColumns.keySet());
+ noDynamicCols = dynGroups.isEmpty();
+ List<TupleExpression> dynGroupExprs = Lists.newArrayListWithExpectedSize(sqlDigest.dynGroupbyColumns.size());
+ for (TblColRef dynGroupCol : dynGroups) {
+ dynGroupExprs.add(sqlDigest.dynGroupbyColumns.get(dynGroupCol));
+ }
+ // dynamic measures
+ List<DynamicFunctionDesc> dynFuncs = sqlDigest.dynAggregations;
+ noDynamicCols = noDynamicCols && dynFuncs.isEmpty();
+
+ CuboidToGridTableMapping mapping = noDynamicCols ? new CuboidToGridTableMapping(cuboid)
+ : new CuboidToGridTableMappingExt(cuboid, dynGroups, dynFuncs);
+ context.setMapping(mapping);
+
// set whether to aggr at storage
Set<TblColRef> singleValuesD = findSingleValueColumns(filter);
context.setNeedStorageAggregation(isNeedStorageAggregation(cuboid, groupsD, singleValuesD));
@@ -156,8 +179,8 @@ public GTCubeStorageQueryRequest getStorageQueryRequest(StorageContext context,
context.setFilterMask(getQueryFilterMask(filterColumnD));
// set limit push down
- enableStorageLimitIfPossible(cuboid, groups, derivedPostAggregation, groupsD, filterD, loosenedColumnD,
- sqlDigest.aggregations, context);
+ enableStorageLimitIfPossible(cuboid, groups, dynGroups, derivedPostAggregation, groupsD, filterD,
+ loosenedColumnD, sqlDigest.aggregations, context);
// set whether to aggregate results from multiple partitions
enableStreamAggregateIfBeneficial(cuboid, groupsD, context);
// check query deadline
@@ -172,8 +195,8 @@ public GTCubeStorageQueryRequest getStorageQueryRequest(StorageContext context,
cubeInstance.getName(), cuboid.getId(), groupsD, filterColumnD, context.getFinalPushDownLimit(),
context.getStorageLimitLevel(), context.isNeedStorageAggregation());
- return new GTCubeStorageQueryRequest(cuboid, dimensionsD, groupsD, filterColumnD, metrics, filterD,
- havingFilter, context);
+ return new GTCubeStorageQueryRequest(cuboid, dimensionsD, groupsD, dynGroups, dynGroupExprs, filterColumnD,
+ metrics, dynFuncs, filterD, havingFilter, context);
}
protected abstract String getGTStorage();
@@ -198,8 +221,9 @@ protected void buildDimensionsAndMetrics(SQLDigest sqlDigest, Collection<TblColR
for (TblColRef column : sqlDigest.allColumns) {
// skip measure columns
- if (sqlDigest.metricColumns.contains(column)
- && !(sqlDigest.groupbyColumns.contains(column) || sqlDigest.filterColumns.contains(column))) {
+ if ((sqlDigest.metricColumns.contains(column) || sqlDigest.rtMetricColumns.contains(column))
+ && !(sqlDigest.groupbyColumns.contains(column) || sqlDigest.filterColumns.contains(column)
+ || sqlDigest.rtDimensionColumns.contains(column))) {
continue;
}
@@ -222,7 +246,7 @@ private FunctionDesc findAggrFuncFromCubeDesc(FunctionDesc aggrFunc) {
DeriveInfo hostInfo = cubeDesc.getHostInfo(col);
for (TblColRef hostCol : hostInfo.columns) {
expanded.add(hostCol);
- if (hostInfo.isOneToOne == false)
+ if (!hostInfo.isOneToOne)
derivedPostAggregation.add(hostCol);
}
} else {
@@ -359,7 +383,7 @@ private TupleFilter translateDerivedInCompare(CompareTupleFilter compf, Set<TblC
if (cubeDesc.isExtendedColumn(derived)) {
throw new CubeDesc.CannotFilterExtendedColumnException(derived);
}
- if (cubeDesc.isDerived(derived) == false)
+ if (!cubeDesc.isDerived(derived))
return compf;
DeriveInfo hostInfo = cubeDesc.getHostInfo(derived);
@@ -393,7 +417,7 @@ private void collectColumnsRecursively(TupleFilter filter, Set<TblColRef> collec
}
}
- private void enableStorageLimitIfPossible(Cuboid cuboid, Collection<TblColRef> groups,
+ private void enableStorageLimitIfPossible(Cuboid cuboid, Collection<TblColRef> groups, List<TblColRef> dynGroups,
Set<TblColRef> derivedPostAggregation, Collection<TblColRef> groupsD, TupleFilter filter,
Set<TblColRef> loosenedColumnD, Collection<FunctionDesc> functionDescs, StorageContext context) {
@@ -409,6 +433,11 @@ private void enableStorageLimitIfPossible(Cuboid cuboid, Collection<TblColRef> g
+ " with cuboid columns: " + cuboid.getColumns());
}
+ if (!dynGroups.isEmpty()) {
+ storageLimitLevel = StorageLimitLevel.NO_LIMIT;
+ logger.debug("Storage limit push down is impossible because the query has dynamic groupby " + dynGroups);
+ }
+
// derived aggregation is bad, unless expanded columns are already in group by
if (!groups.containsAll(derivedPostAggregation)) {
storageLimitLevel = StorageLimitLevel.NO_LIMIT;
@@ -539,14 +568,14 @@ private boolean isExactAggregation(StorageContext context, Cuboid cuboid, Collec
}
// derived aggregation is bad, unless expanded columns are already in group by
- if (groups.containsAll(derivedPostAggregation) == false) {
+ if (!groups.containsAll(derivedPostAggregation)) {
logger.info("exactAggregation is false because derived column require post aggregation: "
+ derivedPostAggregation);
return false;
}
// other columns (from filter) is bad, unless they are ensured to have single value
- if (singleValuesD.containsAll(othersD) == false) {
+ if (!singleValuesD.containsAll(othersD)) {
logger.info("exactAggregation is false because some column not on group by: " + othersD //
+ " (single value column: " + singleValuesD + ")");
return false;
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryRequest.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryRequest.java
index 7793515c1c..fdc976e208 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryRequest.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/GTCubeStorageQueryRequest.java
@@ -19,10 +19,13 @@
package org.apache.kylin.storage.gtrecord;
import java.io.Serializable;
+import java.util.List;
import java.util.Set;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.StorageContext;
@@ -34,17 +37,25 @@
private Set<TblColRef> groups;
private Set<TblColRef> filterCols;
private Set<FunctionDesc> metrics;
+ private List<TblColRef> dynGroups;
+ private List<TupleExpression> dynGroupExprs;
+ private List<DynamicFunctionDesc> dynFuncs;
private TupleFilter filter;
private TupleFilter havingFilter;
private StorageContext context;
- public GTCubeStorageQueryRequest(Cuboid cuboid, Set<TblColRef> dimensions, Set<TblColRef> groups, //
- Set<TblColRef> filterCols, Set<FunctionDesc> metrics, TupleFilter filter, TupleFilter havingFilter, StorageContext context) {
+ public GTCubeStorageQueryRequest(Cuboid cuboid, Set<TblColRef> dimensions, //
+ Set<TblColRef> groups, List<TblColRef> dynGroups, List<TupleExpression> dynGroupExprs, //
+ Set<TblColRef> filterCols, Set<FunctionDesc> metrics, List<DynamicFunctionDesc> dynFuncs, //
+ TupleFilter filter, TupleFilter havingFilter, StorageContext context) {
this.cuboid = cuboid;
this.dimensions = dimensions;
this.groups = groups;
+ this.dynGroups = dynGroups;
+ this.dynGroupExprs = dynGroupExprs;
this.filterCols = filterCols;
this.metrics = metrics;
+ this.dynFuncs = dynFuncs;
this.filter = filter;
this.havingFilter = havingFilter;
this.context = context;
@@ -74,6 +85,14 @@ public void setGroups(Set<TblColRef> groups) {
this.groups = groups;
}
+ public List<TblColRef> getDynGroups() {
+ return dynGroups;
+ }
+
+ public List<TupleExpression> getDynGroupExprs() {
+ return dynGroupExprs;
+ }
+
public Set<FunctionDesc> getMetrics() {
return metrics;
}
@@ -82,6 +101,10 @@ public void setMetrics(Set<FunctionDesc> metrics) {
this.metrics = metrics;
}
+ public List<DynamicFunctionDesc> getDynFuncs() {
+ return dynFuncs;
+ }
+
public TupleFilter getFilter() {
return filter;
}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SegmentCubeTupleIterator.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SegmentCubeTupleIterator.java
index 3bac5ec5a1..fe4ff2ec60 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SegmentCubeTupleIterator.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SegmentCubeTupleIterator.java
@@ -73,7 +73,7 @@ public SegmentCubeTupleIterator(CubeSegmentScanner scanner, Cuboid cuboid, Set<T
this.tuple = new Tuple(returnTupleInfo);
this.context = context;
- CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
+ CuboidToGridTableMapping mapping = context.getMapping();
int[] gtDimsIdx = mapping.getDimIndexes(selectedDimensions);
int[] gtMetricsIdx = mapping.getMetricsIndexes(selectedMetrics);
// gtColIdx = gtDimsIdx + gtMetricsIdx
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SequentialCubeTupleIterator.java b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SequentialCubeTupleIterator.java
index c067e333fb..b8dff8bb5f 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SequentialCubeTupleIterator.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/SequentialCubeTupleIterator.java
@@ -25,6 +25,7 @@
import java.util.Set;
import java.util.TreeSet;
+import com.google.common.collect.Sets;
import org.apache.kylin.common.QueryContextFacade;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.metadata.model.FunctionDesc;
@@ -53,14 +54,18 @@
private int scanCount;
private int scanCountDelta;
- public SequentialCubeTupleIterator(List<CubeSegmentScanner> scanners, Cuboid cuboid, Set<TblColRef> selectedDimensions, //
- Set<TblColRef> groups, Set<FunctionDesc> selectedMetrics, TupleInfo returnTupleInfo, StorageContext context, SQLDigest sqlDigest) {
+ public SequentialCubeTupleIterator(List<CubeSegmentScanner> scanners, Cuboid cuboid,
+ Set<TblColRef> selectedDimensions, List<TblColRef> rtGroups, Set<TblColRef> groups, //
+ Set<FunctionDesc> selectedMetrics, TupleInfo returnTupleInfo, StorageContext context, SQLDigest sqlDigest) {
this.context = context;
this.scanners = scanners;
+ Set<TblColRef> selectedDims = Sets.newHashSet(selectedDimensions);
+ selectedDims.addAll(rtGroups);
+
segmentCubeTupleIterators = Lists.newArrayList();
for (CubeSegmentScanner scanner : scanners) {
- segmentCubeTupleIterators.add(new SegmentCubeTupleIterator(scanner, cuboid, selectedDimensions, selectedMetrics, returnTupleInfo, context));
+ segmentCubeTupleIterators.add(new SegmentCubeTupleIterator(scanner, cuboid, selectedDims, selectedMetrics, returnTupleInfo, context));
}
if (context.mergeSortPartitionResults() && !sqlDigest.isRawQuery) {
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
index 7616df22e5..f90a644ab1 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
@@ -148,9 +148,9 @@ public void testReducerOnlyAggrInBaseCuboid() throws Exception {
List<Pair<Text, Text>> result = reduceDriver.run();
- Pair<Text, Text> p1 = new Pair<Text, Text>(new Text("72010ustech"), newValueText(codec, "0", "10", "20.34", 3, 600));
- Pair<Text, Text> p2 = new Pair<Text, Text>(new Text("1tech"), newValueText(codec, "0", "15.09", "20.34", 2, 1500));
- Pair<Text, Text> p3 = new Pair<Text, Text>(new Text("0"), newValueText(codec, "0", "146.52", "146.52", 0, 0));
+ Pair<Text, Text> p1 = new Pair<>(new Text("72010ustech"), newValueText(codec, null, "10", "20.34", 3, 600));
+ Pair<Text, Text> p2 = new Pair<>(new Text("1tech"), newValueText(codec, null, "15.09", "20.34", 2, 1500));
+ Pair<Text, Text> p3 = new Pair<>(new Text("0"), newValueText(codec, null, "146.52", "146.52", 0, 0));
assertEquals(3, result.size());
@@ -160,7 +160,8 @@ public void testReducerOnlyAggrInBaseCuboid() throws Exception {
}
private Text newValueText(BufferedMeasureCodec codec, String sum, String min, String max, int count, int item_count) {
- Object[] values = new Object[] { new BigDecimal(sum), new BigDecimal(min), new BigDecimal(max), new Long(count), new Long(item_count) };
+ Object[] values = new Object[] { sum == null ? null : new BigDecimal(sum), //
+ new BigDecimal(min), new BigDecimal(max), new Long(count), new Long(item_count) };
ByteBuffer buf = codec.encode(values);
diff --git a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
index e5d3c6bb2d..2d0c7b5e7e 100644
--- a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
@@ -415,6 +415,12 @@ public void testPercentileQuery() throws Exception {
batchExecuteQuery(getQueryFolderPrefix() + "src/test/resources/query/sql_percentile");
}
+ @Test
+ public void testExpressionQuery() throws Exception {
+ if (config.isDynamicColumnEnabled()) {
+ batchExecuteQuery(getQueryFolderPrefix() + "src/test/resources/query/sql_expression");
+ }
+ }
@Test
public void testValues() throws Exception {
diff --git a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
index c61753c0ff..61aa5602e2 100644
--- a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
@@ -29,7 +29,9 @@
import org.apache.kylin.common.util.HBaseMetadataTestCase;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.TblColRef;
@@ -139,7 +141,11 @@ private int search(List<TblColRef> groups, List<FunctionDesc> aggregations, Tupl
try {
SQLDigest sqlDigest = new SQLDigest("default.test_kylin_fact", /*allCol*/ Collections.<TblColRef> emptySet(), /*join*/ null, //
groups, /*subqueryJoinParticipants*/ Sets.<TblColRef> newHashSet(), //
+ /*dynamicGroupByColumns*/ Collections.<TblColRef, TupleExpression> emptyMap(), //
/*metricCol*/ Collections.<TblColRef> emptySet(), aggregations, /*aggrSqlCalls*/ Collections.<SQLCall> emptyList(), //
+ /*dynamicAggregations*/ Collections.<DynamicFunctionDesc> emptyList(), //
+ /*runtimeDimensionColumns*/ Collections.<TblColRef> emptySet(), //
+ /*runtimeMetricColumns*/ Collections.<TblColRef> emptySet(), //
/*filter col*/ Collections.<TblColRef> emptySet(), filter, null, //
/*sortCol*/ new ArrayList<TblColRef>(), new ArrayList<SQLDigest.OrderEnum>(), false, new HashSet<MeasureDesc>());
iterator = storageEngine.search(context, sqlDigest, mockup.newTupleInfo(groups, aggregations));
diff --git a/kylin-it/src/test/resources/query/sql_expression/query01.sql b/kylin-it/src/test/resources/query/sql_expression/query01.sql
new file mode 100644
index 0000000000..35025ac12b
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_expression/query01.sql
@@ -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 SLR_SEGMENT_CD, sum(case when LSTG_FORMAT_NAME is null then 0 else 1 end)
+FROM test_kylin_fact
+group by SLR_SEGMENT_CD
+order by SLR_SEGMENT_CD
\ No newline at end of file
diff --git a/kylin-it/src/test/resources/query/sql_expression/query02.sql b/kylin-it/src/test/resources/query/sql_expression/query02.sql
new file mode 100644
index 0000000000..a73018bb27
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_expression/query02.sql
@@ -0,0 +1,34 @@
+--
+-- 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 LSTG_FORMAT_NAME,
+ sum(price),
+ sum(case
+ when LSTG_FORMAT_NAME = 'ABIN' then 2*price
+ when LSTG_FORMAT_NAME = 'Auction' then (1+2)*price*(2+3)+(2+3)*(3+2)*(4+5)-4+5
+ else 3
+ end)
+FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ 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
+WHERE SLR_SEGMENT_CD < 16
+group by LSTG_FORMAT_NAME
+having sum((1+2)*price*(2+3)+(2+3)*(3+2)*(4+5)-4+5) > 1800000
+order by LSTG_FORMAT_NAME
\ No newline at end of file
diff --git a/kylin-it/src/test/resources/query/sql_expression/query03.sql b/kylin-it/src/test/resources/query/sql_expression/query03.sql
new file mode 100644
index 0000000000..85fa0a5e14
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_expression/query03.sql
@@ -0,0 +1,26 @@
+--
+-- 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 SLR_SEGMENT_CD, count(LSTG_FORMAT_NAME)
+FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ 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
+group by SLR_SEGMENT_CD
+order by SLR_SEGMENT_CD
\ No newline at end of file
diff --git a/kylin-it/src/test/resources/query/sql_expression/query04.sql b/kylin-it/src/test/resources/query/sql_expression/query04.sql
new file mode 100644
index 0000000000..22bee60e5b
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_expression/query04.sql
@@ -0,0 +1,40 @@
+--
+-- 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 LSTG_FORMAT_NAME, (case
+ when SLR_SEGMENT_CD < 0 then 0
+ when SLR_SEGMENT_CD < 10 then 1
+ else 3
+ end),
+ sum(price)
+FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ 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
+WHERE SLR_SEGMENT_CD < 16
+group by LSTG_FORMAT_NAME, (case
+ when SLR_SEGMENT_CD < 0 then 0
+ when SLR_SEGMENT_CD < 10 then 1
+ else 3
+ end)
+order by LSTG_FORMAT_NAME, (case
+ when SLR_SEGMENT_CD < 0 then 0
+ when SLR_SEGMENT_CD < 10 then 1
+ else 3
+ end)
\ No newline at end of file
diff --git a/kylin-it/src/test/resources/query/sql_expression/query05.sql b/kylin-it/src/test/resources/query/sql_expression/query05.sql
new file mode 100644
index 0000000000..f1ce6c2c76
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_expression/query05.sql
@@ -0,0 +1,40 @@
+--
+-- 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 LSTG_FORMAT_NAME, (case
+ when USER_DEFINED_FIELD1 < 0 then 0
+ when USER_DEFINED_FIELD1 < 10 then 1
+ else 3
+ end),
+ sum(price)
+FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ 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
+WHERE SLR_SEGMENT_CD < 16 and USER_DEFINED_FIELD1 is not null
+group by LSTG_FORMAT_NAME, (case
+ when USER_DEFINED_FIELD1 < 0 then 0
+ when USER_DEFINED_FIELD1 < 10 then 1
+ else 3
+ end)
+order by LSTG_FORMAT_NAME, (case
+ when USER_DEFINED_FIELD1 < 0 then 0
+ when USER_DEFINED_FIELD1 < 10 then 1
+ else 3
+ end)
\ No newline at end of file
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/ColumnRowType.java b/query/src/main/java/org/apache/kylin/query/relnode/ColumnRowType.java
index ce50122e88..65a00e66a2 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/ColumnRowType.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/ColumnRowType.java
@@ -18,10 +18,12 @@
package org.apache.kylin.query.relnode;
-import java.util.Collections;
import java.util.List;
-import java.util.Set;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.expression.ColumnTupleExpression;
+import org.apache.kylin.metadata.expression.NoneTupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.model.TblColRef;
/**
@@ -34,13 +36,13 @@
private List<TblColRef> columns;
// for calculated column, like (CASE LSTG_FORMAT_NAME WHEN 'Auction' THEN '111' ELSE '222' END)
// source columns are the contributing physical columns, here the LSTG_FORMAT_NAME
- private List<Set<TblColRef>> sourceColumns;
+ private List<TupleExpression> sourceColumns;
public ColumnRowType(List<TblColRef> columns) {
this(columns, null);
}
- public ColumnRowType(List<TblColRef> columns, List<Set<TblColRef>> sourceColumns) {
+ public ColumnRowType(List<TblColRef> columns, List<TupleExpression> sourceColumns) {
this.columns = columns;
this.sourceColumns = sourceColumns;
}
@@ -69,13 +71,23 @@ public int getIndexByName(String columnName) {
return -1;
}
- public Set<TblColRef> getSourceColumnsByIndex(int i) {
- Set<TblColRef> result = null;
+ public Pair<TblColRef, TupleExpression> replaceColumnByIndex(int index, TblColRef newColumn,
+ TupleExpression newTupleExpr) {
+ if (index < 0 || index >= columns.size()) {
+ return null;
+ }
+ TblColRef oldCol = columns.set(index, newColumn);
+ TupleExpression oldExpr = sourceColumns.set(index, newTupleExpr);
+ return new Pair<>(oldCol, oldExpr);
+ }
+
+ public TupleExpression getSourceColumnsByIndex(int i) {
+ TupleExpression result = null;
if (sourceColumns != null) {
result = sourceColumns.get(i);
}
- if (result == null || result.isEmpty()) {
- result = Collections.singleton(getColumnByIndex(i));
+ if (result == null || result instanceof NoneTupleExpression) {
+ result = new ColumnTupleExpression(getColumnByIndex(i));
}
return result;
}
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 722d287914..ab8be1b8f1 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
@@ -47,7 +47,10 @@
import org.apache.calcite.schema.impl.AggregateFunctionImpl;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.fun.SqlCountAggFunction;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.fun.SqlSumAggFunction;
+import org.apache.calcite.sql.fun.SqlSumEmptyIsZeroAggFunction;
import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.sql.type.InferTypes;
import org.apache.calcite.sql.type.OperandTypes;
@@ -56,17 +59,30 @@
import org.apache.calcite.sql.validate.SqlUserDefinedAggFunction;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Util;
+import org.apache.kylin.common.util.Pair;
import org.apache.kylin.measure.MeasureTypeFactory;
import org.apache.kylin.measure.ParamAsMeasureCount;
+import org.apache.kylin.metadata.expression.CaseTupleExpression;
+import org.apache.kylin.metadata.expression.ColumnTupleExpression;
+import org.apache.kylin.metadata.expression.ExpressionColCollector;
+import org.apache.kylin.metadata.expression.ExpressionCountDistributor;
+import org.apache.kylin.metadata.expression.NumberTupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpression;
+import org.apache.kylin.metadata.filter.ColumnTupleFilter;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.ParameterDesc;
+import org.apache.kylin.metadata.model.SumDynamicFunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.SQLDigest.SQLCall;
import org.apache.kylin.query.schema.OLAPTable;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
/**
@@ -120,9 +136,11 @@ public static String getAggrFuncName(AggregateCall aggCall) {
OLAPContext context;
ColumnRowType columnRowType;
private boolean afterAggregate;
+ private Map<Integer, AggregateCall> hackAggCalls;
private List<AggregateCall> rewriteAggCalls;
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)
@@ -245,20 +263,62 @@ TblColRef buildRewriteColumn(FunctionDesc aggFunc) {
void buildGroups() {
ColumnRowType inputColumnRowType = ((OLAPRel) getInput()).getColumnRowType();
- this.groups = new ArrayList<TblColRef>();
+ this.groups = Lists.newArrayList();
for (int i = getGroupSet().nextSetBit(0); i >= 0; i = getGroupSet().nextSetBit(i + 1)) {
- Set<TblColRef> columns = inputColumnRowType.getSourceColumnsByIndex(i);
- this.groups.addAll(columns);
+ TupleExpression tupleExpression = inputColumnRowType.getSourceColumnsByIndex(i);
+ TblColRef groupOutCol = inputColumnRowType.getColumnByIndex(i);
+ if (tupleExpression instanceof ColumnTupleExpression) {
+ this.groups.add(((ColumnTupleExpression) tupleExpression).getColumn());
+ } else if (this.context.isDynamicColumnEnabled()) {
+ Pair<Set<TblColRef>, Set<TblColRef>> cols = ExpressionColCollector.collectColumnsPair(tupleExpression);
+
+ // push down only available for the innermost aggregation
+ boolean ifPushDown = !afterAggregate;
+
+ // if measure columns exist, don't do push down
+ if (ifPushDown && !cols.getSecond().isEmpty()) {
+ ifPushDown = false;
+ }
+
+ // if existing a dimension which is a derived column, don't do push down
+ if (ifPushDown) {
+ for (TblColRef dimCol : cols.getFirst()) {
+ if (!this.context.belongToFactTableDims(dimCol)) {
+ ifPushDown = false;
+ break;
+ }
+ }
+ }
+
+ if (ifPushDown) {
+ this.groups.add(groupOutCol);
+ this.context.dynGroupBy.put(groupOutCol, tupleExpression);
+ } else {
+ this.groups.addAll(cols.getFirst());
+ this.groups.addAll(cols.getSecond());
+ this.context.dynamicFields.remove(groupOutCol);
+ }
+ } else {
+ Set<TblColRef> srcCols = ExpressionColCollector.collectColumns(tupleExpression);
+ // if no source columns, use target column instead
+ if (srcCols.isEmpty()) {
+ srcCols.add(groupOutCol);
+ }
+ this.groups.addAll(srcCols);
+ }
}
}
void buildAggregations() {
ColumnRowType inputColumnRowType = ((OLAPRel) getInput()).getColumnRowType();
- this.aggregations = new ArrayList<FunctionDesc>();
- for (AggregateCall aggCall : this.rewriteAggCalls) {
+ this.aggregations = Lists.newArrayList();
+ this.hackAggCalls = Maps.newHashMap();
+ for (int i = 0; i < this.rewriteAggCalls.size(); i++) {
+ AggregateCall aggCall = this.rewriteAggCalls.get(i);
ParameterDesc parameter = null;
+ List<Integer> argList = aggCall.getArgList();
// By default all args are included, UDFs can define their own in getParamAsMeasureCount method.
- if (!aggCall.getArgList().isEmpty()) {
+ if (!argList.isEmpty()) {
List<TblColRef> columns = Lists.newArrayList();
String funcName = getSqlFuncName(aggCall);
int columnsCount = aggCall.getArgList().size();
@@ -278,6 +338,48 @@ void buildAggregations() {
parameter = ParameterDesc.newInstance(columns.toArray(new TblColRef[columns.size()]));
}
}
+ // Check dynamic aggregation
+ if (this.context.isDynamicColumnEnabled() && !afterAggregate && !rewriting && argList.size() == 1) {
+ int iRowIdx = argList.get(0);
+ TupleExpression tupleExpr = inputColumnRowType.getSourceColumnsByIndex(iRowIdx);
+ if (aggCall.getAggregation() instanceof SqlSumAggFunction
+ || aggCall.getAggregation() instanceof SqlSumEmptyIsZeroAggFunction) {
+ // sum (expression)
+ if (!(tupleExpr instanceof NumberTupleExpression || tupleExpr instanceof ColumnTupleExpression)) {
+ ColumnTupleExpression cntExpr = new ColumnTupleExpression(SumDynamicFunctionDesc.mockCntCol);
+ ExpressionCountDistributor cntDistributor = new ExpressionCountDistributor(cntExpr);
+ tupleExpr = tupleExpr.accept(cntDistributor);
+ SumDynamicFunctionDesc sumDynFunc = new SumDynamicFunctionDesc(parameter, tupleExpr);
+ this.aggregations.add(sumDynFunc);
+ continue;
+ }
+ } else if (aggCall.getAggregation() instanceof SqlCountAggFunction && !aggCall.isDistinct()) {
+ // count column
+ if (tupleExpr instanceof ColumnTupleExpression) {
+ TblColRef srcCol = ((ColumnTupleExpression) tupleExpr).getColumn();
+ if (this.context.belongToFactTableDims(srcCol)) {
+ tupleExpr = getCountColumnExpression(srcCol);
+
+ TblColRef column = TblColRef.newInnerColumn(tupleExpr.getDigest(),
+ TblColRef.InnerDataTypeEnum.LITERAL);
+
+ SumDynamicFunctionDesc sumDynFunc = new SumDynamicFunctionDesc(
+ ParameterDesc.newInstance(column), tupleExpr);
+
+ inputColumnRowType.replaceColumnByIndex(iRowIdx, column, tupleExpr);
+
+ AggregateCall newAggCall = AggregateCall.create(SqlStdOperatorTable.SUM, false,
+ aggCall.getArgList(), -1, aggCall.getType(), aggCall.getName());
+ this.hackAggCalls.put(i, newAggCall);
+
+ this.context.dynamicFields.put(column, aggCall.getType());
+
+ this.aggregations.add(sumDynFunc);
+ continue;
+ }
+ }
+ }
+ }
String expression = getAggrFuncName(aggCall);
FunctionDesc aggFunc = FunctionDesc.newInstance(expression, parameter, null);
this.aggregations.add(aggFunc);
@@ -298,12 +400,16 @@ public void implementRewrite(RewriteImplementor implementor) {
}
implementor.visitChild(this, getInput());
+
+ this.rewriting = true;
+
// only rewrite the innermost aggregation
if (needRewrite()) {
// rewrite the aggCalls
- this.rewriteAggCalls = new ArrayList<AggregateCall>(aggCalls.size());
+ this.rewriteAggCalls = Lists.newArrayListWithExpectedSize(aggCalls.size());
for (int i = 0; i < this.aggCalls.size(); i++) {
- AggregateCall aggCall = this.aggCalls.get(i);
+ AggregateCall aggCall = this.hackAggCalls.get(i) != null ? this.hackAggCalls.get(i)
+ : this.aggCalls.get(i);
FunctionDesc cubeFunc = this.context.aggregations.get(i);
// filter needn,t rewrite aggfunc
// if it's not a cube, then the "needRewriteField func" should not resort to any rewrite fields,
@@ -332,6 +438,7 @@ public void implementRewrite(RewriteImplementor implementor) {
this.rowType = this.deriveRowType();
this.columnRowType = this.buildColumnRowType();
+ this.rewriting = false;
}
SQLCall toSqlCall(AggregateCall aggCall) {
@@ -352,7 +459,18 @@ void translateAggregation() {
List<MeasureDesc> measures = this.context.realization.getMeasures();
List<FunctionDesc> newAggrs = Lists.newArrayList();
for (FunctionDesc aggFunc : this.aggregations) {
- newAggrs.add(findInMeasures(aggFunc, measures));
+ if (aggFunc instanceof DynamicFunctionDesc) {
+ DynamicFunctionDesc rtAggFunc = (DynamicFunctionDesc) aggFunc;
+ Map<TblColRef, FunctionDesc> innerOldAggrs = rtAggFunc.getRuntimeFuncMap();
+ Map<TblColRef, FunctionDesc> innerNewAggrs = Maps.newHashMapWithExpectedSize(innerOldAggrs.size());
+ for (TblColRef key : innerOldAggrs.keySet()) {
+ innerNewAggrs.put(key, findInMeasures(innerOldAggrs.get(key), measures));
+ }
+ rtAggFunc.setRuntimeFuncMap(innerNewAggrs);
+ newAggrs.add(rtAggFunc);
+ } else {
+ newAggrs.add(findInMeasures(aggFunc, measures));
+ }
}
this.aggregations.clear();
this.aggregations.addAll(newAggrs);
@@ -406,7 +524,7 @@ void buildRewriteFieldsAndMetricsColumns() {
void addToContextGroupBy(List<TblColRef> colRefs) {
for (TblColRef col : colRefs) {
- if (col.isInnerColumn() == false && this.context.belongToContextTables(col))
+ if (!col.isInnerColumn() && this.context.belongToContextTables(col))
this.context.groupByColumns.add(col);
}
}
@@ -520,4 +638,16 @@ public RelWriter explainTerms(RelWriter pw) {
return super.explainTerms(pw).item("ctx",
context == null ? "" : String.valueOf(context.id) + "@" + context.realization);
}
+
+ private TupleExpression getCountColumnExpression(TblColRef colRef) {
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists.newArrayListWithExpectedSize(1);
+ TupleFilter whenFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.ISNULL);
+ whenFilter.addChild(new ColumnTupleFilter(colRef));
+ whenList.add(new Pair<TupleFilter, TupleExpression>(whenFilter, new NumberTupleExpression(0)));
+
+ TupleExpression elseExpr = new ColumnTupleExpression(SumDynamicFunctionDesc.mockCntCol);
+ TupleExpression ret = new CaseTupleExpression(whenList, elseExpr);
+ ret.setDigest("_KY_COUNT(" + colRef.getName() + ")");
+ return ret;
+ }
}
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
index d041ff12d4..c03bd0b65e 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
@@ -33,14 +33,19 @@
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.DateFormat;
import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.metadata.expression.ExpressionColCollector;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.JoinDesc;
import org.apache.kylin.metadata.model.JoinsTree;
import org.apache.kylin.metadata.model.MeasureDesc;
+import org.apache.kylin.metadata.model.TableRef;
import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.project.ProjectManager;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.metadata.realization.SQLDigest.SQLCall;
@@ -51,6 +56,7 @@
import org.apache.kylin.storage.hybrid.HybridInstance;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
/**
*/
@@ -151,6 +157,11 @@ public OLAPContext(int seq) {
// rewrite info
public Map<String, RelDataType> rewriteFields = new HashMap<>();
+ // dynamic columns info, note that the name of TblColRef will be the field name
+ public Map<TblColRef, RelDataType> dynamicFields = new HashMap<>();
+
+ public Map<TblColRef, TupleExpression> dynGroupBy = new HashMap<>();
+
// hive query
public String sql = "";
@@ -163,16 +174,36 @@ public boolean isSimpleQuery() {
SQLDigest sqlDigest;
public SQLDigest getSQLDigest() {
- if (sqlDigest == null)
+ if (sqlDigest == null) {
+ Set<TblColRef> rtDimColumns = new HashSet<>();
+ for (TupleExpression tupleExpr : dynGroupBy.values()) {
+ rtDimColumns.addAll(ExpressionColCollector.collectColumns(tupleExpr));
+ }
+ Set<TblColRef> rtMetricColumns = new HashSet<>();
+ List<DynamicFunctionDesc> dynFuncs = Lists.newLinkedList();
+ for (FunctionDesc functionDesc : aggregations) {
+ if (functionDesc instanceof DynamicFunctionDesc) {
+ DynamicFunctionDesc dynFunc = (DynamicFunctionDesc) functionDesc;
+ rtMetricColumns.addAll(dynFunc.getMeasureColumnSet());
+ rtDimColumns.addAll(dynFunc.getFilterColumnSet());
+ dynFuncs.add(dynFunc);
+ }
+ }
sqlDigest = new SQLDigest(firstTableScan.getTableName(), allColumns, joins, // model
- groupByColumns, subqueryJoinParticipants, // group by
- metricsColumns, aggregations, aggrSqlCalls, // aggregation
+ groupByColumns, subqueryJoinParticipants, dynGroupBy, // group by
+ metricsColumns, aggregations, aggrSqlCalls, dynFuncs, // aggregation
+ rtDimColumns, rtMetricColumns, // runtime related columns
filterColumns, filter, havingFilter, // filter
sortColumns, sortOrders, limitPrecedesAggr, // sort & limit
involvedMeasure);
+ }
return sqlDigest;
}
+ public boolean isDynamicColumnEnabled() {
+ return olapSchema != null && olapSchema.getConfig().isDynamicColumnEnabled();
+ }
+
public boolean hasPrecalculatedFields() {
return realization instanceof CubeInstance || realization instanceof HybridInstance;
}
@@ -191,6 +222,38 @@ public boolean belongToContextTables(TblColRef tblColRef) {
return false;
}
+ public boolean belongToFactTableDims(TblColRef tblColRef) {
+ if (!belongToContextTables(tblColRef)) {
+ return false;
+ }
+ KylinConfig kylinConfig = olapSchema.getConfig();
+ String projectName = olapSchema.getProjectName();
+ String factTableName = firstTableScan.getOlapTable().getTableName();
+ Set<IRealization> realizations = ProjectManager.getInstance(kylinConfig).getRealizationsByTable(projectName,
+ factTableName);
+ for (IRealization real : realizations) {
+ DataModelDesc model = real.getModel();
+ TblColRef.fixUnknownModel(model, tblColRef.getTableRef().getTableIdentity(), tblColRef);
+
+ // cannot be a measure column
+ Set<String> metrics = Sets.newHashSet(model.getMetrics());
+ if (metrics.contains(tblColRef.getIdentity())) {
+ tblColRef.unfixTableRef();
+ return false;
+ }
+
+ // must belong to a fact table
+ for (TableRef factTable : model.getFactTables()) {
+ if (factTable.getColumns().contains(tblColRef)) {
+ tblColRef.unfixTableRef();
+ return true;
+ }
+ }
+ tblColRef.unfixTableRef();
+ }
+ return false;
+ }
+
public void setReturnTupleInfo(RelDataType rowType, ColumnRowType columnRowType) {
TupleInfo info = new TupleInfo();
List<RelDataTypeField> fieldList = rowType.getFieldList();
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
index eedd814a19..39a266941b 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
@@ -18,18 +18,12 @@
package org.apache.kylin.query.relnode;
-import java.math.BigDecimal;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.LinkedList;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import org.apache.calcite.adapter.enumerable.EnumerableCalc;
import org.apache.calcite.adapter.enumerable.EnumerableConvention;
import org.apache.calcite.adapter.enumerable.EnumerableRel;
-import org.apache.calcite.avatica.util.TimeUnitRange;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptCost;
import org.apache.calcite.plan.RelOptPlanner;
@@ -41,290 +35,21 @@
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
-import org.apache.calcite.rex.RexCall;
-import org.apache.calcite.rex.RexDynamicParam;
-import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexLocalRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexProgram;
import org.apache.calcite.rex.RexProgramBuilder;
-import org.apache.calcite.rex.RexVisitorImpl;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlOperator;
-import org.apache.calcite.sql.type.SqlTypeFamily;
-import org.apache.calcite.util.NlsString;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.metadata.filter.CaseTupleFilter;
-import org.apache.kylin.metadata.filter.ColumnTupleFilter;
-import org.apache.kylin.metadata.filter.CompareTupleFilter;
-import org.apache.kylin.metadata.filter.ConstantTupleFilter;
-import org.apache.kylin.metadata.filter.DynamicTupleFilter;
-import org.apache.kylin.metadata.filter.ExtractTupleFilter;
import org.apache.kylin.metadata.filter.FilterOptimizeTransformer;
-import org.apache.kylin.metadata.filter.LogicalTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
-import org.apache.kylin.metadata.filter.UnsupportedTupleFilter;
-import org.apache.kylin.metadata.filter.function.Functions;
import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.query.relnode.visitor.TupleFilterVisitor;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
/**
*/
public class OLAPFilterRel extends Filter implements OLAPRel {
- static class TupleFilterVisitor extends RexVisitorImpl<TupleFilter> {
-
- final ColumnRowType inputRowType;
-
- public TupleFilterVisitor(ColumnRowType inputRowType) {
- super(true);
- this.inputRowType = inputRowType;
- }
-
- @Override
- public TupleFilter visitCall(RexCall call) {
- TupleFilter filter = null;
- SqlOperator op = call.getOperator();
- switch (op.getKind()) {
- case AND:
- filter = new LogicalTupleFilter(FilterOperatorEnum.AND);
- break;
- case OR:
- filter = new LogicalTupleFilter(FilterOperatorEnum.OR);
- break;
- case NOT:
- filter = new LogicalTupleFilter(FilterOperatorEnum.NOT);
- break;
- case EQUALS:
- filter = new CompareTupleFilter(FilterOperatorEnum.EQ);
- break;
- case GREATER_THAN:
- filter = new CompareTupleFilter(FilterOperatorEnum.GT);
- break;
- case LESS_THAN:
- filter = new CompareTupleFilter(FilterOperatorEnum.LT);
- break;
- case GREATER_THAN_OR_EQUAL:
- filter = new CompareTupleFilter(FilterOperatorEnum.GTE);
- break;
- case LESS_THAN_OR_EQUAL:
- filter = new CompareTupleFilter(FilterOperatorEnum.LTE);
- break;
- case NOT_EQUALS:
- filter = new CompareTupleFilter(FilterOperatorEnum.NEQ);
- break;
- case IS_NULL:
- filter = new CompareTupleFilter(FilterOperatorEnum.ISNULL);
- break;
- case IS_NOT_NULL:
- filter = new CompareTupleFilter(FilterOperatorEnum.ISNOTNULL);
- break;
- case CAST:
- case REINTERPRET:
- // NOTE: use child directly
- break;
- case CASE:
- filter = new CaseTupleFilter();
- break;
- case OTHER:
- if (op.getName().equalsIgnoreCase("extract_date")) {
- filter = new ExtractTupleFilter(FilterOperatorEnum.EXTRACT);
- } else {
- filter = Functions.getFunctionTupleFilter(op.getName());
- }
- break;
- case LIKE:
- case OTHER_FUNCTION:
- filter = Functions.getFunctionTupleFilter(op.getName());
- break;
- case PLUS:
- case MINUS:
- case TIMES:
- case DIVIDE:
- TupleFilter f = dealWithTrivialExpr(call);
- if (f != null) {
- // is a trivial expr
- return f;
- }
- //else go to default
- default:
- filter = new UnsupportedTupleFilter(FilterOperatorEnum.UNSUPPORTED);
- break;
- }
-
- for (RexNode operand : call.operands) {
- TupleFilter childFilter = operand.accept(this);
- if (filter == null) {
- filter = cast(childFilter, call.type);
- } else {
- filter.addChild(childFilter);
- }
- }
-
- if (op.getKind() == SqlKind.OR) {
- CompareTupleFilter inFilter = mergeToInClause(filter);
- if (inFilter != null) {
- filter = inFilter;
- }
- } else if (op.getKind() == SqlKind.NOT) {
- assert (filter.getChildren().size() == 1);
- filter = filter.getChildren().get(0).reverse();
- }
- return filter;
- }
-
- //KYLIN-2597 - Deal with trivial expression in filters like x = 1 + 2
- TupleFilter dealWithTrivialExpr(RexCall call) {
- ImmutableList<RexNode> operators = call.operands;
- if (operators.size() != 2) {
- return null;
- }
-
- BigDecimal left = null;
- BigDecimal right = null;
- for (RexNode rexNode : operators) {
- if (!(rexNode instanceof RexLiteral)) {
- return null;// only trivial expr with constants
- }
-
- RexLiteral temp = (RexLiteral) rexNode;
- if (temp.getType().getFamily() != SqlTypeFamily.NUMERIC || !(temp.getValue() instanceof BigDecimal)) {
- return null;// only numeric constants now
- }
-
- if (left == null) {
- left = (BigDecimal) temp.getValue();
- } else {
- right = (BigDecimal) temp.getValue();
- }
- }
-
- Preconditions.checkNotNull(left);
- Preconditions.checkNotNull(right);
-
- switch (call.op.getKind()) {
- case PLUS:
- return new ConstantTupleFilter(left.add(right).toString());
- case MINUS:
- return new ConstantTupleFilter(left.subtract(right).toString());
- case TIMES:
- return new ConstantTupleFilter(left.multiply(right).toString());
- case DIVIDE:
- return new ConstantTupleFilter(left.divide(right).toString());
- default:
- return null;
- }
- }
-
- TupleFilter cast(TupleFilter filter, RelDataType type) {
- if ((filter instanceof ConstantTupleFilter) == false) {
- return filter;
- }
-
- ConstantTupleFilter constFilter = (ConstantTupleFilter) filter;
-
- if (type.getFamily() == SqlTypeFamily.DATE || type.getFamily() == SqlTypeFamily.DATETIME
- || type.getFamily() == SqlTypeFamily.TIMESTAMP) {
- List<String> newValues = Lists.newArrayList();
- for (Object v : constFilter.getValues()) {
- if (v == null)
- newValues.add(null);
- else
- newValues.add(String.valueOf(DateFormat.stringToMillis(v.toString())));
- }
- constFilter = new ConstantTupleFilter(newValues);
- }
- return constFilter;
- }
-
- CompareTupleFilter mergeToInClause(TupleFilter filter) {
- List<? extends TupleFilter> children = filter.getChildren();
- TblColRef inColumn = null;
- List<Object> inValues = new LinkedList<Object>();
- Map<String, Object> dynamicVariables = new HashMap<>();
- for (TupleFilter child : children) {
- if (child.getOperator() == FilterOperatorEnum.EQ) {
- CompareTupleFilter compFilter = (CompareTupleFilter) child;
- TblColRef column = compFilter.getColumn();
- if (inColumn == null) {
- inColumn = column;
- }
-
- if (column == null || !column.equals(inColumn)) {
- return null;
- }
- inValues.addAll(compFilter.getValues());
- dynamicVariables.putAll(compFilter.getVariables());
- } else {
- return null;
- }
- }
-
- children.clear();
-
- CompareTupleFilter inFilter = new CompareTupleFilter(FilterOperatorEnum.IN);
- inFilter.addChild(new ColumnTupleFilter(inColumn));
- inFilter.addChild(new ConstantTupleFilter(inValues));
- inFilter.getVariables().putAll(dynamicVariables);
- return inFilter;
- }
-
- @Override
- public TupleFilter visitLocalRef(RexLocalRef localRef) {
- throw new UnsupportedOperationException("local ref:" + localRef);
- }
-
- @Override
- public TupleFilter visitInputRef(RexInputRef inputRef) {
- TblColRef column = inputRowType.getColumnByIndex(inputRef.getIndex());
- ColumnTupleFilter filter = new ColumnTupleFilter(column);
- return filter;
- }
-
- @SuppressWarnings("unused")
- String normToTwoDigits(int i) {
- if (i < 10)
- return "0" + i;
- else
- return "" + i;
- }
-
- @Override
- public TupleFilter visitLiteral(RexLiteral literal) {
- String strValue = null;
- Object literalValue = literal.getValue();
- if (literalValue instanceof NlsString) {
- strValue = ((NlsString) literalValue).getValue();
- } else if (literalValue instanceof GregorianCalendar) {
- GregorianCalendar g = (GregorianCalendar) literalValue;
- //strValue = "" + g.get(Calendar.YEAR) + "-" + normToTwoDigits(g.get(Calendar.MONTH) + 1) + "-" + normToTwoDigits(g.get(Calendar.DAY_OF_MONTH));
- strValue = Long.toString(g.getTimeInMillis());
- } else if (literalValue instanceof TimeUnitRange) {
- // Extract(x from y) in where clause
- strValue = ((TimeUnitRange) literalValue).name();
- } else if (literalValue == null) {
- strValue = null;
- } else {
- strValue = literalValue.toString();
- }
- TupleFilter filter = new ConstantTupleFilter(strValue);
- return filter;
- }
-
- @Override
- public TupleFilter visitDynamicParam(RexDynamicParam dynamicParam) {
- String name = dynamicParam.getName();
- TupleFilter filter = new DynamicTupleFilter(name);
- return filter;
- }
- }
-
ColumnRowType columnRowType;
OLAPContext context;
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPJoinRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPJoinRel.java
index 4182453a75..479c8ca00f 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPJoinRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPJoinRel.java
@@ -49,6 +49,7 @@
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeFactory.FieldInfoBuilder;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
@@ -62,6 +63,7 @@
import org.apache.kylin.query.schema.OLAPTable;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
/**
*/
@@ -333,9 +335,8 @@ public void implementRewrite(RewriteImplementor implementor) {
this.rowType = this.deriveRowType();
- if (this.isTopJoin && RewriteImplementor.needRewrite(this.context)) {
- if (this.context.hasPrecalculatedFields()) {
-
+ if (this.isTopJoin) {
+ if (RewriteImplementor.needRewrite(this.context) && this.context.hasPrecalculatedFields()) {
// find missed rewrite fields
int paramIndex = this.rowType.getFieldList().size();
List<RelDataTypeField> newFieldList = new LinkedList<RelDataTypeField>();
@@ -357,6 +358,30 @@ public void implementRewrite(RewriteImplementor implementor) {
// rebuild columns
this.columnRowType = this.buildColumnRowType();
}
+
+ // add dynamic field
+ Map<TblColRef, RelDataType> dynFields = this.context.dynamicFields;
+ if (!dynFields.isEmpty()) {
+ List<TblColRef> newCols = Lists.newArrayList(this.columnRowType.getAllColumns());
+ List<RelDataTypeField> newFieldList = Lists.newArrayList();
+ int paramIndex = this.rowType.getFieldList().size();
+ for (TblColRef fieldCol : dynFields.keySet()) {
+ RelDataType fieldType = dynFields.get(fieldCol);
+
+ RelDataTypeField newField = new RelDataTypeFieldImpl(fieldCol.getName(), paramIndex++, fieldType);
+ newFieldList.add(newField);
+
+ newCols.add(fieldCol);
+ }
+
+ // rebuild row type
+ RelDataTypeFactory.FieldInfoBuilder fieldInfo = getCluster().getTypeFactory().builder();
+ fieldInfo.addAll(this.rowType.getFieldList());
+ fieldInfo.addAll(newFieldList);
+ this.rowType = getCluster().getTypeFactory().createStructType(fieldInfo);
+
+ this.columnRowType = new ColumnRowType(newCols);
+ }
}
}
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPProjectRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPProjectRel.java
index b0938bd66f..39f4bb0813 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPProjectRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPProjectRel.java
@@ -18,11 +18,7 @@
package org.apache.kylin.query.relnode;
-import static org.apache.kylin.metadata.filter.CompareTupleFilter.CompareResultType;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.LinkedList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -43,23 +39,29 @@
import org.apache.calcite.rel.type.RelDataTypeFactory.FieldInfoBuilder;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
-import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexOver;
import org.apache.calcite.rex.RexProgram;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.fun.SqlCaseOperator;
-import org.apache.calcite.sql.fun.SqlStdOperatorTable;
-import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
+import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.tools.RelUtils;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.expression.ColumnTupleExpression;
+import org.apache.kylin.metadata.expression.ExpressionColCollector;
+import org.apache.kylin.metadata.expression.NoneTupleExpression;
+import org.apache.kylin.metadata.expression.NumberTupleExpression;
+import org.apache.kylin.metadata.expression.StringTupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.model.TblColRef.InnerDataTypeEnum;
+import org.apache.kylin.query.relnode.visitor.TupleExpressionVisitor;
+import org.apache.kylin.query.schema.OLAPTable;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
/**
*/
@@ -138,157 +140,61 @@ public void implementOLAP(OLAPImplementor implementor) {
}
ColumnRowType buildColumnRowType() {
- List<TblColRef> columns = new ArrayList<TblColRef>();
- List<Set<TblColRef>> sourceColumns = new ArrayList<Set<TblColRef>>();
+ List<TblColRef> columns = Lists.newArrayList();
+ List<TupleExpression> sourceColumns = Lists.newArrayList();
+
OLAPRel olapChild = (OLAPRel) getInput();
ColumnRowType inputColumnRowType = olapChild.getColumnRowType();
+ TupleExpressionVisitor visitor = new TupleExpressionVisitor(inputColumnRowType, afterAggregate);
for (int i = 0; i < this.rewriteProjects.size(); i++) {
RexNode rex = this.rewriteProjects.get(i);
RelDataTypeField columnField = this.rowType.getFieldList().get(i);
String fieldName = columnField.getName();
- Set<TblColRef> sourceCollector = new HashSet<TblColRef>();
- TblColRef column = translateRexNode(rex, inputColumnRowType, fieldName, sourceCollector);
- if (column == null)
- throw new IllegalStateException("No TblColRef found in " + rex);
- columns.add(column);
- sourceColumns.add(sourceCollector);
- }
- return new ColumnRowType(columns, sourceColumns);
- }
-
- TblColRef translateRexNode(RexNode rexNode, ColumnRowType inputColumnRowType, String fieldName,
- Set<TblColRef> sourceCollector) {
- TblColRef column = null;
- if (rexNode instanceof RexInputRef) {
- RexInputRef inputRef = (RexInputRef) rexNode;
- column = translateRexInputRef(inputRef, inputColumnRowType, fieldName, sourceCollector);
- } else if (rexNode instanceof RexLiteral) {
- RexLiteral literal = (RexLiteral) rexNode;
- column = translateRexLiteral(literal);
- } else if (rexNode instanceof RexCall) {
- RexCall call = (RexCall) rexNode;
- column = translateRexCall(call, inputColumnRowType, fieldName, sourceCollector);
- } else {
- throw new IllegalStateException("Unsupported RexNode " + rexNode);
- }
- return column;
- }
-
- TblColRef translateFirstRexInputRef(RexCall call, ColumnRowType inputColumnRowType, String fieldName,
- Set<TblColRef> sourceCollector) {
- for (RexNode operand : call.getOperands()) {
- if (operand instanceof RexInputRef) {
- return translateRexInputRef((RexInputRef) operand, inputColumnRowType, fieldName, sourceCollector);
- }
- if (operand instanceof RexCall) {
- TblColRef r = translateFirstRexInputRef((RexCall) operand, inputColumnRowType, fieldName,
- sourceCollector);
- if (r != null)
- return r;
- }
- }
- return null;
- }
- TblColRef translateRexInputRef(RexInputRef inputRef, ColumnRowType inputColumnRowType, String fieldName,
- Set<TblColRef> sourceCollector) {
- int index = inputRef.getIndex();
- // check it for rewrite count
- if (index < inputColumnRowType.size()) {
- TblColRef column = inputColumnRowType.getColumnByIndex(index);
- if (!column.isInnerColumn() && context.belongToContextTables(column) && !this.rewriting
- && !this.afterAggregate) {
- if (!isMerelyPermutation) {
- context.allColumns.add(column);
+ TupleExpression tupleExpr = rex.accept(visitor);
+ TblColRef column = translateRexNode(tupleExpr, fieldName);
+ if (!this.rewriting && !this.afterAggregate && !isMerelyPermutation) {
+ Set<TblColRef> srcCols = ExpressionColCollector.collectColumns(tupleExpr);
+ // remove cols not belonging to context tables
+ Iterator<TblColRef> srcColIter = srcCols.iterator();
+ while (srcColIter.hasNext()) {
+ if (!context.belongToContextTables(srcColIter.next())) {
+ srcColIter.remove();
+ }
}
- sourceCollector.add(column);
- }
- return column;
- } else {
- throw new IllegalStateException("Can't find " + inputRef + " from child columnrowtype " + inputColumnRowType
- + " with fieldname " + fieldName);
- }
- }
-
- TblColRef translateRexLiteral(RexLiteral literal) {
- if (RexLiteral.isNullLiteral(literal)) {
- return TblColRef.newInnerColumn("null", InnerDataTypeEnum.LITERAL);
- } else {
- return TblColRef.newInnerColumn(literal.getValue().toString(), InnerDataTypeEnum.LITERAL);
- }
-
- }
-
- TblColRef translateRexCall(RexCall call, ColumnRowType inputColumnRowType, String fieldName,
- Set<TblColRef> sourceCollector) {
- SqlOperator operator = call.getOperator();
- if (operator == SqlStdOperatorTable.EXTRACT_DATE) {
- return translateFirstRexInputRef(call, inputColumnRowType, fieldName, sourceCollector);
- } else if (operator instanceof SqlUserDefinedFunction) {
- if (operator.getName().equals("QUARTER")) {
- return translateFirstRexInputRef(call, inputColumnRowType, fieldName, sourceCollector);
- }
- }
-
- List<RexNode> children = limitTranslateScope(call.getOperands(), operator);
-
- for (RexNode operand : children) {
- translateRexNode(operand, inputColumnRowType, fieldName, sourceCollector);
- }
- return TblColRef.newInnerColumn(fieldName, InnerDataTypeEnum.LITERAL, call.toString());
- }
-
- //in most cases it will return children itself
- List<RexNode> limitTranslateScope(List<RexNode> children, SqlOperator operator) {
-
- //group by case when 1 = 1 then x 1 = 2 then y else z
- if (operator instanceof SqlCaseOperator) {
- int unknownWhenCalls = 0;
- for (int i = 0; i < children.size() - 1; i += 2) {
- if (children.get(i) instanceof RexCall) {
- RexCall whenCall = (RexCall) children.get(i);
- CompareResultType compareResultType = getCompareResultType(whenCall);
- if (compareResultType == CompareResultType.AlwaysTrue) {
- return Lists.newArrayList(children.get(i), children.get(i + 1));
- } else if (compareResultType == CompareResultType.Unknown) {
- unknownWhenCalls++;
+ this.context.allColumns.addAll(srcCols);
+
+ if (this.context.isDynamicColumnEnabled() && tupleExpr.ifForDynamicColumn()) {
+ SqlTypeName fSqlType = columnField.getType().getSqlTypeName();
+ String dataType = OLAPTable.DATATYPE_MAPPING.get(fSqlType);
+ // upgrade data type for number columns
+ if (DataType.isNumberFamily(dataType)) {
+ dataType = "decimal";
}
+ column.getColumnDesc().setDatatype(dataType);
+ this.context.dynamicFields.put(column, columnField.getType());
}
+ } else {
+ tupleExpr = new NoneTupleExpression();
}
- if (unknownWhenCalls == 0) {
- return Lists.newArrayList(children.get(children.size() - 1));
- }
- }
-
- return children;
- }
-
- CompareResultType getCompareResultType(RexCall whenCall) {
- List<RexNode> operands = whenCall.getOperands();
- if (SqlKind.EQUALS == whenCall.getKind() && operands != null && operands.size() == 2) {
- if (operands.get(0).equals(operands.get(1))) {
- return CompareResultType.AlwaysTrue;
- }
-
- if (isConstant(operands.get(0)) && isConstant(operands.get(1))) {
- return CompareResultType.AlwaysFalse;
- }
+ columns.add(column);
+ sourceColumns.add(tupleExpr);
}
- return CompareResultType.Unknown;
+ return new ColumnRowType(columns, sourceColumns);
}
- boolean isConstant(RexNode rexNode) {
- if (rexNode instanceof RexLiteral) {
- return true;
+ private TblColRef translateRexNode(TupleExpression tupleExpr, String fieldName) {
+ if (tupleExpr instanceof ColumnTupleExpression) {
+ return ((ColumnTupleExpression) tupleExpr).getColumn();
+ } else if (tupleExpr instanceof NumberTupleExpression) {
+ Object value = ((NumberTupleExpression) tupleExpr).getValue();
+ return TblColRef.newInnerColumn(value == null ? "null" : value.toString(), InnerDataTypeEnum.LITERAL);
+ } else if (tupleExpr instanceof StringTupleExpression) {
+ Object value = ((StringTupleExpression) tupleExpr).getValue();
+ return TblColRef.newInnerColumn(value == null ? "null" : value.toString(), InnerDataTypeEnum.LITERAL);
}
-
- if (rexNode instanceof RexCall && SqlKind.CAST.equals(rexNode.getKind())
- && ((RexCall) rexNode).getOperands().get(0) instanceof RexLiteral) {
- return true;
- }
-
- return false;
+ return TblColRef.newInnerColumn(fieldName, InnerDataTypeEnum.LITERAL, tupleExpr.getDigest());
}
@Override
@@ -329,12 +235,15 @@ public void implementRewrite(RewriteImplementor implementor) {
return;
}
- // find missed rewrite fields
- int paramIndex = this.rowType.getFieldList().size();
- List<RelDataTypeField> newFieldList = new LinkedList<RelDataTypeField>();
- List<RexNode> newExpList = new LinkedList<RexNode>();
+ List<RelDataTypeField> newFieldList = Lists.newLinkedList();
+ List<RexNode> newExpList = Lists.newLinkedList();
+ Map<Integer, Pair<RelDataTypeField, RexNode>> replaceFieldMap = Maps
+ .newHashMapWithExpectedSize(this.context.dynamicFields.size());
+
ColumnRowType inputColumnRowType = ((OLAPRel) getInput()).getColumnRowType();
+ // find missed rewrite fields
+ int paramIndex = this.rowType.getFieldList().size();
for (Map.Entry<String, RelDataType> rewriteField : this.context.rewriteFields.entrySet()) {
String rewriteFieldName = rewriteField.getKey();
int rowIndex = this.columnRowType.getIndexByName(rewriteFieldName);
@@ -353,15 +262,42 @@ public void implementRewrite(RewriteImplementor implementor) {
}
}
- if (!newFieldList.isEmpty()) {
+ // replace projects with dynamic fields
+ Map<TblColRef, RelDataType> dynFields = this.context.dynamicFields;
+ for (TblColRef dynFieldCol : dynFields.keySet()) {
+ String replaceFieldName = dynFieldCol.getName();
+ int rowIndex = this.columnRowType.getIndexByName(replaceFieldName);
+ if (rowIndex >= 0) {
+ int inputIndex = inputColumnRowType.getIndexByName(replaceFieldName);
+ if (inputIndex >= 0) {
+ // field to be replaced
+ RelDataType fieldType = dynFields.get(dynFieldCol);
+ RelDataTypeField newField = new RelDataTypeFieldImpl(replaceFieldName, rowIndex, fieldType);
+ // project to be replaced
+ RelDataTypeField inputField = getInput().getRowType().getFieldList().get(inputIndex);
+ RexInputRef newFieldRef = new RexInputRef(inputField.getIndex(), inputField.getType());
+
+ replaceFieldMap.put(rowIndex, new Pair<RelDataTypeField, RexNode>(newField, newFieldRef));
+ }
+ }
+ }
+
+ if (!newFieldList.isEmpty() || !replaceFieldMap.isEmpty()) {
+ List<RexNode> newProjects = Lists.newArrayList(this.rewriteProjects);
+ List<RelDataTypeField> newFields = Lists.newArrayList(this.rowType.getFieldList());
+ for (int rowIndex : replaceFieldMap.keySet()) {
+ Pair<RelDataTypeField, RexNode> entry = replaceFieldMap.get(rowIndex);
+ newProjects.set(rowIndex, entry.getSecond());
+ newFields.set(rowIndex, entry.getFirst());
+ }
+
// rebuild projects
- List<RexNode> newProjects = new ArrayList<RexNode>(this.rewriteProjects);
newProjects.addAll(newExpList);
this.rewriteProjects = newProjects;
// rebuild row type
FieldInfoBuilder fieldInfo = getCluster().getTypeFactory().builder();
- fieldInfo.addAll(this.rowType.getFieldList());
+ fieldInfo.addAll(newFields);
fieldInfo.addAll(newFieldList);
this.rowType = getCluster().getTypeFactory().createStructType(fieldInfo);
}
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPTableScan.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPTableScan.java
index 67d3851fca..c23f1c5f4b 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPTableScan.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPTableScan.java
@@ -65,6 +65,7 @@
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.metadata.model.ColumnDesc;
@@ -88,6 +89,7 @@
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
/**
*/
@@ -439,6 +441,27 @@ public void implementRewrite(RewriteImplementor implementor) {
rewriteField.setValue(fieldType);
}
}
+ // add dynamic field to the table scan if join not exist
+ if (!this.context.hasJoin && !this.context.dynamicFields.isEmpty()) {
+ Map<TblColRef, RelDataType> dynFields = this.context.dynamicFields;
+ List<TblColRef> newCols = Lists.newArrayList(this.columnRowType.getAllColumns());
+ List<RelDataTypeField> newFieldList = Lists.newArrayList(this.rowType.getFieldList());
+ int paramIndex = this.rowType.getFieldList().size();
+ for (TblColRef fieldCol : dynFields.keySet()) {
+ newCols.add(fieldCol);
+
+ RelDataType fieldType = dynFields.get(fieldCol);
+ RelDataTypeField newField = new RelDataTypeFieldImpl(fieldCol.getName(), paramIndex++, fieldType);
+ newFieldList.add(newField);
+ }
+
+ // rebuild row type
+ RelDataTypeFactory.FieldInfoBuilder fieldInfo = getCluster().getTypeFactory().builder();
+ fieldInfo.addAll(newFieldList);
+ this.rowType = getCluster().getTypeFactory().createStructType(fieldInfo);
+
+ this.columnRowType = new ColumnRowType(newCols);
+ }
}
@Override
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPUnionRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPUnionRel.java
index 4276449164..11f0a92a59 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPUnionRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPUnionRel.java
@@ -19,9 +19,7 @@
package org.apache.kylin.query.relnode;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.List;
-import java.util.Set;
import org.apache.calcite.adapter.enumerable.EnumerableConvention;
import org.apache.calcite.adapter.enumerable.EnumerableRel;
@@ -36,9 +34,13 @@
import org.apache.calcite.rel.core.SetOp;
import org.apache.calcite.rel.core.Union;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.kylin.metadata.expression.ColumnTupleExpression;
+import org.apache.kylin.metadata.expression.RexCallTupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.model.TblColRef;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
/**
*/
@@ -96,8 +98,8 @@ public void implementOLAP(OLAPImplementor implementor) {
*/
private ColumnRowType buildColumnRowType() {
ColumnRowType inputColumnRowType = ((OLAPRel) getInput(0)).getColumnRowType();
- List<TblColRef> columns = new ArrayList<>();
- List<Set<TblColRef>> sourceColumns = new ArrayList<>();
+ List<TblColRef> columns = Lists.newArrayList();
+ List<TupleExpression> sourceColumns = Lists.newArrayList();
for (TblColRef tblColRef : inputColumnRowType.getAllColumns()) {
columns.add(TblColRef.newInnerColumn(tblColRef.getName(), TblColRef.InnerDataTypeEnum.LITERAL));
@@ -105,7 +107,12 @@ private ColumnRowType buildColumnRowType() {
for (RelNode child : getInputs()) {
OLAPRel olapChild = (OLAPRel) child;
- sourceColumns.add(new HashSet<>(olapChild.getColumnRowType().getAllColumns()));
+ List<TblColRef> innerCols = olapChild.getColumnRowType().getAllColumns();
+ List<TupleExpression> children = Lists.newArrayListWithExpectedSize(innerCols.size());
+ for (TblColRef innerCol : innerCols) {
+ children.add(new ColumnTupleExpression(innerCol));
+ }
+ sourceColumns.add(new RexCallTupleExpression(children));
}
ColumnRowType fackColumnRowType = new ColumnRowType(columns, sourceColumns);
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleExpressionVisitor.java b/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleExpressionVisitor.java
new file mode 100644
index 0000000000..4007754a67
--- /dev/null
+++ b/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleExpressionVisitor.java
@@ -0,0 +1,204 @@
+/*
+ * 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.relnode.visitor;
+
+import java.util.List;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.fun.SqlCastFunction;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
+import org.apache.calcite.util.NlsString;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.metadata.expression.BinaryTupleExpression;
+import org.apache.kylin.metadata.expression.CaseTupleExpression;
+import org.apache.kylin.metadata.expression.ColumnTupleExpression;
+import org.apache.kylin.metadata.expression.NumberTupleExpression;
+import org.apache.kylin.metadata.expression.RexCallTupleExpression;
+import org.apache.kylin.metadata.expression.StringTupleExpression;
+import org.apache.kylin.metadata.expression.TupleExpression;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+import org.apache.kylin.metadata.filter.FilterOptimizeTransformer;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.query.relnode.ColumnRowType;
+import org.apache.kylin.query.util.RexUtil;
+
+import com.google.common.collect.Lists;
+
+public class TupleExpressionVisitor extends RexVisitorImpl<TupleExpression> {
+ final ColumnRowType inputRowType;
+ final boolean afterAggregate;
+
+ public TupleExpressionVisitor(ColumnRowType inputRowType, boolean afterAggregate) {
+ super(true);
+ this.inputRowType = inputRowType;
+ this.afterAggregate = afterAggregate;
+ }
+
+ @Override
+ public TupleExpression visitCall(RexCall call) {
+ SqlOperator op = call.getOperator();
+ if (op == SqlStdOperatorTable.EXTRACT_DATE) {
+ return visitFirstRexInputRef(call);
+ } else if (op instanceof SqlCastFunction) {
+ return call.getOperands().get(0).accept(this);
+ } else if (op instanceof SqlUserDefinedFunction) {
+ if (op.getName().equals("QUARTER")) {
+ return visitFirstRexInputRef(call);
+ }
+ }
+
+ TupleExpression tupleExpression;
+ switch (op.getKind()) {
+ case PLUS:
+ tupleExpression = getBinaryTupleExpression(call, TupleExpression.ExpressionOperatorEnum.PLUS);
+ break;
+ case MINUS:
+ tupleExpression = getBinaryTupleExpression(call, TupleExpression.ExpressionOperatorEnum.MINUS);
+ break;
+ case TIMES:
+ tupleExpression = getBinaryTupleExpression(call, TupleExpression.ExpressionOperatorEnum.MULTIPLE);
+ break;
+ case DIVIDE:
+ tupleExpression = getBinaryTupleExpression(call, TupleExpression.ExpressionOperatorEnum.DIVIDE);
+ break;
+ case CASE:
+ tupleExpression = getCaseTupleExpression(call);
+ break;
+ default:
+ tupleExpression = getRexCallTupleExpression(call);
+ }
+ if (!afterAggregate) {
+ tupleExpression.verify();
+ }
+ return tupleExpression;
+ }
+
+ private BinaryTupleExpression getBinaryTupleExpression(RexCall call, TupleExpression.ExpressionOperatorEnum op) {
+ assert call.operands.size() == 2;
+ TupleExpression left = call.operands.get(0).accept(this);
+ TupleExpression right = call.operands.get(1).accept(this);
+ BinaryTupleExpression tuple = new BinaryTupleExpression(op, Lists.newArrayList(left, right));
+ tuple.setDigest(call.toString());
+ return tuple;
+ }
+
+ private CaseTupleExpression getCaseTupleExpression(RexCall call) {
+ List<Pair<TupleFilter, TupleExpression>> whenList = Lists
+ .newArrayListWithExpectedSize(call.operands.size() / 2);
+ TupleExpression elseExpr = null;
+
+ TupleFilterVisitor filterVistor = new TupleFilterVisitor(inputRowType);
+ for (int i = 0; i < call.operands.size() - 1; i += 2) {
+ if (call.operands.get(i) instanceof RexCall) {
+ RexCall whenCall = (RexCall) call.operands.get(i);
+ CompareTupleFilter.CompareResultType compareResultType = RexUtil.getCompareResultType(whenCall);
+ if (compareResultType == CompareTupleFilter.CompareResultType.AlwaysTrue) {
+ elseExpr = call.operands.get(i + 1).accept(this);
+ break;
+ } else if (compareResultType == CompareTupleFilter.CompareResultType.AlwaysFalse) {
+ continue;
+ }
+ TupleFilter whenFilter = whenCall.accept(filterVistor);
+ whenFilter = new FilterOptimizeTransformer().transform(whenFilter);
+
+ TupleExpression thenExpr = call.operands.get(i + 1).accept(this);
+ whenList.add(new Pair<>(whenFilter, thenExpr));
+ }
+ }
+ if (elseExpr == null && call.operands.size() % 2 == 1) {
+ RexNode elseNode = call.operands.get(call.operands.size() - 1);
+ if (!(elseNode instanceof RexLiteral && ((RexLiteral) elseNode).getValue() == null)) {
+ elseExpr = elseNode.accept(this);
+ }
+ }
+ CaseTupleExpression tuple = new CaseTupleExpression(whenList, elseExpr);
+ tuple.setDigest(call.toString());
+ return tuple;
+ }
+
+ private RexCallTupleExpression getRexCallTupleExpression(RexCall call) {
+ List<TupleExpression> children = Lists.newArrayListWithExpectedSize(call.getOperands().size());
+ for (RexNode rexNode : call.operands) {
+ children.add(rexNode.accept(this));
+ }
+ RexCallTupleExpression tuple = new RexCallTupleExpression(children);
+ tuple.setDigest(call.toString());
+ return tuple;
+ }
+
+ @Override
+ public TupleExpression visitLocalRef(RexLocalRef localRef) {
+ throw new UnsupportedOperationException("local ref:" + localRef);
+ }
+
+ @Override
+ public TupleExpression visitInputRef(RexInputRef inputRef) {
+ int index = inputRef.getIndex();
+ // check it for rewrite count
+ if (index < inputRowType.size()) {
+ TblColRef column = inputRowType.getColumnByIndex(index);
+ TupleExpression tuple = new ColumnTupleExpression(column);
+ tuple.setDigest(inputRef.toString());
+ return tuple;
+ } else {
+ throw new IllegalStateException("Can't find " + inputRef + " from child columnrowtype");
+ }
+ }
+
+ public TupleExpression visitFirstRexInputRef(RexCall call) {
+ for (RexNode operand : call.getOperands()) {
+ if (operand instanceof RexInputRef) {
+ return visitInputRef((RexInputRef) operand);
+ }
+ if (operand instanceof RexCall) {
+ TupleExpression r = visitFirstRexInputRef((RexCall) operand);
+ if (r != null)
+ return r;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public TupleExpression visitLiteral(RexLiteral literal) {
+ TupleExpression tuple;
+ Object value = literal.getValue();
+ if (value instanceof Number) {
+ tuple = new NumberTupleExpression(value);
+ } else {
+ if (value == null) {
+ tuple = new StringTupleExpression(null);
+ } else if (value instanceof NlsString) {
+ tuple = new StringTupleExpression(((NlsString) value).getValue());
+ } else {
+ tuple = new StringTupleExpression(value.toString());
+ }
+ }
+ tuple.setDigest(literal.toString());
+ return tuple;
+ }
+}
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleFilterVisitor.java b/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleFilterVisitor.java
new file mode 100644
index 0000000000..bf52f91117
--- /dev/null
+++ b/query/src/main/java/org/apache/kylin/query/relnode/visitor/TupleFilterVisitor.java
@@ -0,0 +1,304 @@
+/*
+ * 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.relnode.visitor;
+
+import java.math.BigDecimal;
+import java.util.GregorianCalendar;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.calcite.avatica.util.TimeUnitRange;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.util.NlsString;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.metadata.filter.CaseTupleFilter;
+import org.apache.kylin.metadata.filter.ColumnTupleFilter;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+import org.apache.kylin.metadata.filter.ConstantTupleFilter;
+import org.apache.kylin.metadata.filter.DynamicTupleFilter;
+import org.apache.kylin.metadata.filter.ExtractTupleFilter;
+import org.apache.kylin.metadata.filter.LogicalTupleFilter;
+import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.filter.UnsupportedTupleFilter;
+import org.apache.kylin.metadata.filter.function.Functions;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.query.relnode.ColumnRowType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+public class TupleFilterVisitor extends RexVisitorImpl<TupleFilter> {
+
+ final ColumnRowType inputRowType;
+
+ public TupleFilterVisitor(ColumnRowType inputRowType) {
+ super(true);
+ this.inputRowType = inputRowType;
+ }
+
+ @Override
+ public TupleFilter visitCall(RexCall call) {
+ TupleFilter filter = null;
+ SqlOperator op = call.getOperator();
+ switch (op.getKind()) {
+ case AND:
+ filter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.AND);
+ break;
+ case OR:
+ filter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.OR);
+ break;
+ case NOT:
+ filter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.NOT);
+ break;
+ case EQUALS:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.EQ);
+ break;
+ case GREATER_THAN:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GT);
+ break;
+ case LESS_THAN:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.LT);
+ break;
+ case GREATER_THAN_OR_EQUAL:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GTE);
+ break;
+ case LESS_THAN_OR_EQUAL:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.LTE);
+ break;
+ case NOT_EQUALS:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.NEQ);
+ break;
+ case IS_NULL:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.ISNULL);
+ break;
+ case IS_NOT_NULL:
+ filter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.ISNOTNULL);
+ break;
+ case CAST:
+ case REINTERPRET:
+ // NOTE: use child directly
+ break;
+ case CASE:
+ filter = new CaseTupleFilter();
+ break;
+ case OTHER:
+ if (op.getName().equalsIgnoreCase("extract_date")) {
+ filter = new ExtractTupleFilter(TupleFilter.FilterOperatorEnum.EXTRACT);
+ } else {
+ filter = Functions.getFunctionTupleFilter(op.getName());
+ }
+ break;
+ case LIKE:
+ case OTHER_FUNCTION:
+ filter = Functions.getFunctionTupleFilter(op.getName());
+ break;
+ case PLUS:
+ case MINUS:
+ case TIMES:
+ case DIVIDE:
+ TupleFilter f = dealWithTrivialExpr(call);
+ if (f != null) {
+ // is a trivial expr
+ return f;
+ }
+ //else go to default
+ default:
+ filter = new UnsupportedTupleFilter(TupleFilter.FilterOperatorEnum.UNSUPPORTED);
+ break;
+ }
+
+ for (RexNode operand : call.operands) {
+ TupleFilter childFilter = operand.accept(this);
+ if (filter == null) {
+ filter = cast(childFilter, call.type);
+ } else {
+ filter.addChild(childFilter);
+ }
+ }
+
+ if (op.getKind() == SqlKind.OR) {
+ CompareTupleFilter inFilter = mergeToInClause(filter);
+ if (inFilter != null) {
+ filter = inFilter;
+ }
+ } else if (op.getKind() == SqlKind.NOT) {
+ assert (filter.getChildren().size() == 1);
+ filter = filter.getChildren().get(0).reverse();
+ }
+ return filter;
+ }
+
+ //KYLIN-2597 - Deal with trivial expression in filters like x = 1 + 2
+ private TupleFilter dealWithTrivialExpr(RexCall call) {
+ ImmutableList<RexNode> operators = call.operands;
+ if (operators.size() != 2) {
+ return null;
+ }
+
+ BigDecimal left = null;
+ BigDecimal right = null;
+ for (RexNode rexNode : operators) {
+ if (!(rexNode instanceof RexLiteral)) {
+ return null;// only trivial expr with constants
+ }
+
+ RexLiteral temp = (RexLiteral) rexNode;
+ if (temp.getType().getFamily() != SqlTypeFamily.NUMERIC || !(temp.getValue() instanceof BigDecimal)) {
+ return null;// only numeric constants now
+ }
+
+ if (left == null) {
+ left = (BigDecimal) temp.getValue();
+ } else {
+ right = (BigDecimal) temp.getValue();
+ }
+ }
+
+ Preconditions.checkNotNull(left);
+ Preconditions.checkNotNull(right);
+
+ switch (call.op.getKind()) {
+ case PLUS:
+ return new ConstantTupleFilter(left.add(right).toString());
+ case MINUS:
+ return new ConstantTupleFilter(left.subtract(right).toString());
+ case TIMES:
+ return new ConstantTupleFilter(left.multiply(right).toString());
+ case DIVIDE:
+ return new ConstantTupleFilter(left.divide(right).toString());
+ default:
+ return null;
+ }
+ }
+
+ private TupleFilter cast(TupleFilter filter, RelDataType type) {
+ if ((filter instanceof ConstantTupleFilter) == false) {
+ return filter;
+ }
+
+ ConstantTupleFilter constFilter = (ConstantTupleFilter) filter;
+
+ if (type.getFamily() == SqlTypeFamily.DATE || type.getFamily() == SqlTypeFamily.DATETIME
+ || type.getFamily() == SqlTypeFamily.TIMESTAMP) {
+ List<String> newValues = Lists.newArrayList();
+ for (Object v : constFilter.getValues()) {
+ if (v == null)
+ newValues.add(null);
+ else
+ newValues.add(String.valueOf(DateFormat.stringToMillis(v.toString())));
+ }
+ constFilter = new ConstantTupleFilter(newValues);
+ }
+ return constFilter;
+ }
+
+ private CompareTupleFilter mergeToInClause(TupleFilter filter) {
+ List<? extends TupleFilter> children = filter.getChildren();
+ TblColRef inColumn = null;
+ List<Object> inValues = new LinkedList<Object>();
+ Map<String, Object> dynamicVariables = new HashMap<>();
+ for (TupleFilter child : children) {
+ if (child.getOperator() == TupleFilter.FilterOperatorEnum.EQ) {
+ CompareTupleFilter compFilter = (CompareTupleFilter) child;
+ TblColRef column = compFilter.getColumn();
+ if (inColumn == null) {
+ inColumn = column;
+ }
+
+ if (column == null || !column.equals(inColumn)) {
+ return null;
+ }
+ inValues.addAll(compFilter.getValues());
+ dynamicVariables.putAll(compFilter.getVariables());
+ } else {
+ return null;
+ }
+ }
+
+ children.clear();
+
+ CompareTupleFilter inFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.IN);
+ inFilter.addChild(new ColumnTupleFilter(inColumn));
+ inFilter.addChild(new ConstantTupleFilter(inValues));
+ inFilter.getVariables().putAll(dynamicVariables);
+ return inFilter;
+ }
+
+ @Override
+ public TupleFilter visitLocalRef(RexLocalRef localRef) {
+ throw new UnsupportedOperationException("local ref:" + localRef);
+ }
+
+ @Override
+ public TupleFilter visitInputRef(RexInputRef inputRef) {
+ TblColRef column = inputRowType.getColumnByIndex(inputRef.getIndex());
+ ColumnTupleFilter filter = new ColumnTupleFilter(column);
+ return filter;
+ }
+
+ @SuppressWarnings("unused")
+ private String normToTwoDigits(int i) {
+ if (i < 10)
+ return "0" + i;
+ else
+ return "" + i;
+ }
+
+ @Override
+ public TupleFilter visitLiteral(RexLiteral literal) {
+ String strValue = null;
+ Object literalValue = literal.getValue();
+ if (literalValue instanceof NlsString) {
+ strValue = ((NlsString) literalValue).getValue();
+ } else if (literalValue instanceof GregorianCalendar) {
+ GregorianCalendar g = (GregorianCalendar) literalValue;
+ //strValue = "" + g.get(Calendar.YEAR) + "-" + normToTwoDigits(g.get(Calendar.MONTH) + 1) + "-" + normToTwoDigits(g.get(Calendar.DAY_OF_MONTH));
+ strValue = Long.toString(g.getTimeInMillis());
+ } else if (literalValue instanceof TimeUnitRange) {
+ // Extract(x from y) in where clause
+ strValue = ((TimeUnitRange) literalValue).name();
+ } else if (literalValue == null) {
+ strValue = null;
+ } else {
+ strValue = literalValue.toString();
+ }
+ TupleFilter filter = new ConstantTupleFilter(strValue);
+ return filter;
+ }
+
+ @Override
+ public TupleFilter visitDynamicParam(RexDynamicParam dynamicParam) {
+ String name = dynamicParam.getName();
+ TupleFilter filter = new DynamicTupleFilter(name);
+ return filter;
+ }
+}
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 b5313ed738..216c6d4a7b 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
@@ -60,6 +60,7 @@
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
/**
*/
@@ -67,6 +68,26 @@
protected static final Logger logger = LoggerFactory.getLogger(OLAPTable.class);
+ public static final Map<SqlTypeName, String> DATATYPE_MAPPING = Maps.newHashMap();
+
+ static {
+ DATATYPE_MAPPING.put(SqlTypeName.CHAR, "char");
+ DATATYPE_MAPPING.put(SqlTypeName.VARCHAR, "varchar");
+ DATATYPE_MAPPING.put(SqlTypeName.BOOLEAN, "boolean");
+ DATATYPE_MAPPING.put(SqlTypeName.INTEGER, "integer");
+ DATATYPE_MAPPING.put(SqlTypeName.TINYINT, "tinyint");
+ DATATYPE_MAPPING.put(SqlTypeName.SMALLINT, "smallint");
+ DATATYPE_MAPPING.put(SqlTypeName.BIGINT, "bigint");
+ DATATYPE_MAPPING.put(SqlTypeName.DECIMAL, "decimal");
+ DATATYPE_MAPPING.put(SqlTypeName.FLOAT, "float");
+ DATATYPE_MAPPING.put(SqlTypeName.REAL, "real");
+ DATATYPE_MAPPING.put(SqlTypeName.DOUBLE, "double");
+ DATATYPE_MAPPING.put(SqlTypeName.DATE, "date");
+ DATATYPE_MAPPING.put(SqlTypeName.TIME, "time");
+ DATATYPE_MAPPING.put(SqlTypeName.TIMESTAMP, "timestamp");
+ DATATYPE_MAPPING.put(SqlTypeName.ANY, "any");
+ }
+
private static Map<String, SqlTypeName> SQLTYPE_MAPPING = new HashMap<String, SqlTypeName>();
private static Map<String, SqlTypeName> REGEX_SQLTYPE_MAPPING = new HashMap<String, SqlTypeName>();
diff --git a/query/src/main/java/org/apache/kylin/query/util/RexUtil.java b/query/src/main/java/org/apache/kylin/query/util/RexUtil.java
new file mode 100644
index 0000000000..27087ee6a0
--- /dev/null
+++ b/query/src/main/java/org/apache/kylin/query/util/RexUtil.java
@@ -0,0 +1,57 @@
+/*
+ * 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.util.List;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.kylin.metadata.filter.CompareTupleFilter;
+
+public class RexUtil {
+
+ public static CompareTupleFilter.CompareResultType getCompareResultType(RexCall whenCall) {
+ List<RexNode> operands = whenCall.getOperands();
+ if (SqlKind.EQUALS == whenCall.getKind() && operands != null && operands.size() == 2) {
+ if (operands.get(0).equals(operands.get(1))) {
+ return CompareTupleFilter.CompareResultType.AlwaysTrue;
+ }
+
+ if (isConstant(operands.get(0)) && isConstant(operands.get(1))) {
+ return CompareTupleFilter.CompareResultType.AlwaysFalse;
+ }
+ }
+ return CompareTupleFilter.CompareResultType.Unknown;
+ }
+
+ public static boolean isConstant(RexNode rexNode) {
+ if (rexNode instanceof RexLiteral) {
+ return true;
+ }
+
+ if (rexNode instanceof RexCall && SqlKind.CAST.equals(rexNode.getKind())
+ && ((RexCall) rexNode).getOperands().get(0) instanceof RexLiteral) {
+ return true;
+ }
+
+ return false;
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services