KYLIN-2501 Stream Aggregate GTRecords at Query Server

Author: gaodayue

core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java

@@ -802,6 +802,10 @@ public boolean isSkippingEmptySegments() {
         return Boolean.valueOf(getOptional("kylin.query.skip-empty-segments", "true"));
     }
 
+    public boolean isStreamAggregateEnabled() {
+        return Boolean.parseBoolean(getOptional("kylin.query.stream-aggregate-enabled", "true"));
+    }
+
     @Deprecated //Limit is good even it's large. This config is meaning less since we already have scan threshold 
     public int getStoragePushDownLimitMax() {
         return Integer.parseInt(getOptional("kylin.query.max-limit-pushdown", "10000"));

core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java

@@ -19,8 +19,9 @@
 
 import java.nio.ByteBuffer;
 import java.util.BitSet;
+import java.util.Iterator;
 
-public class ImmutableBitSet {
+public class ImmutableBitSet implements Iterable<Integer> {
 
     public static final ImmutableBitSet EMPTY = new ImmutableBitSet(new BitSet());
 
@@ -168,4 +169,30 @@ public ImmutableBitSet deserialize(ByteBuffer in) {
             return new ImmutableBitSet(bitSet);
         }
     };
+
+    /**
+     * Iterate over the positions of true value.
+     * @return the iterator
+     */
+    @Override
+    public Iterator<Integer> iterator() {
+        return new Iterator<Integer>() {
+            int index = 0;
+
+            @Override
+            public boolean hasNext() {
+                return index < arr.length;
+            }
+
+            @Override
+            public Integer next() {
+                return arr[index++];
+            }
+
+            @Override
+            public void remove() {
+                throw new UnsupportedOperationException();
+            }
+        };
+    }
 }

core-cube/src/main/java/org/apache/kylin/GTForwardingScanner.java

@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.gridtable.IGTScanner;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * A {@link IGTScanner} which forwards all its method calls to another scanner.
+ *
+ * @see <a href="http://en.wikipedia.org/wiki/Decorator_pattern">decorator pattern</a>.
+ */
+public class GTForwardingScanner implements IGTScanner {
+    protected IGTScanner delegated;
+
+    protected GTForwardingScanner(IGTScanner delegated) {
+        this.delegated = checkNotNull(delegated, "delegated");
+    }
+
+    @Override
+    public GTInfo getInfo() {
+        return delegated.getInfo();
+    }
+
+    @Override
+    public void close() throws IOException {
+        delegated.close();
+    }
+
+    @Override
+    public Iterator<GTRecord> iterator() {
+        return delegated.iterator();
+    }
+}

core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeGridTable.java

@@ -18,29 +18,11 @@
 
 package org.apache.kylin.cube.gridtable;
 
-import java.util.Map;
-
-import org.apache.kylin.common.util.Dictionary;
-import org.apache.kylin.cube.CubeSegment;
 import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.CubeDimEncMap;
-import org.apache.kylin.cube.model.CubeDesc;
 import org.apache.kylin.dimension.IDimensionEncodingMap;
 import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.metadata.model.TblColRef;
 
 public class CubeGridTable {
-
-    public static GTInfo newGTInfo(CubeSegment cubeSeg, long cuboidId) {
-        Cuboid cuboid = Cuboid.findById(cubeSeg.getCubeDesc(), cuboidId);
-        return newGTInfo(cuboid, new CubeDimEncMap(cubeSeg));
-    }
-
-    public static GTInfo newGTInfo(CubeDesc cubeDesc, long cuboidId, Map<TblColRef, Dictionary<String>> dictionaryMap) {
-        Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidId);
-        return newGTInfo(cuboid, new CubeDimEncMap(cubeDesc, dictionaryMap));
-    }
-
     public static GTInfo newGTInfo(Cuboid cuboid, IDimensionEncodingMap dimEncMap) {
         CuboidToGridTableMapping mapping = new CuboidToGridTableMapping(cuboid);
 

core-cube/src/main/java/org/apache/kylin/cube/gridtable/CuboidToGridTableMapping.java

@@ -140,11 +140,29 @@ public int getIndexOf(TblColRef dimension) {
         return i == null ? -1 : i.intValue();
     }
 
+    public int[] getDimIndexes(Collection<TblColRef> dims) {
+        int[] result = new int[dims.size()];
+        int i = 0;
+        for (TblColRef dim : dims) {
+            result[i++] = getIndexOf(dim);
+        }
+        return result;
+    }
+
     public int getIndexOf(FunctionDesc metric) {
         Integer r = metrics2gt.get(metric);
         return r == null ? -1 : r;
     }
 
+    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;
+    }
+
     public List<TblColRef> getCuboidDimensionsInGTOrder() {
         return cuboid.getColumns();
     }

core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java

@@ -38,6 +38,7 @@
 import org.apache.kylin.cube.cuboid.Cuboid;
 import org.apache.kylin.cube.cuboid.CuboidScheduler;
 import org.apache.kylin.cube.gridtable.CubeGridTable;
+import org.apache.kylin.cube.kv.CubeDimEncMap;
 import org.apache.kylin.cube.model.CubeDesc;
 import org.apache.kylin.gridtable.GTAggregateScanner;
 import org.apache.kylin.gridtable.GTBuilder;
@@ -108,7 +109,10 @@ public InMemCubeBuilder(CubeDesc cubeDesc, IJoinedFlatTableDesc flatDesc, Map<Tb
     }
 
     private GridTable newGridTableByCuboidID(long cuboidID) throws IOException {
-        GTInfo info = CubeGridTable.newGTInfo(cubeDesc, cuboidID, dictionaryMap);
+        GTInfo info = CubeGridTable.newGTInfo(
+                Cuboid.findById(cubeDesc, cuboidID),
+                new CubeDimEncMap(cubeDesc, dictionaryMap)
+        );
 
         // Below several store implementation are very similar in performance. The ConcurrentDiskStore is the simplest.
         // MemDiskStore store = new MemDiskStore(info, memBudget == null ? MemoryBudgetController.ZERO_BUDGET : memBudget);

core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java

@@ -45,7 +45,6 @@
 import org.apache.kylin.measure.BufferedMeasureCodec;
 import org.apache.kylin.measure.MeasureAggregator;
 import org.apache.kylin.measure.MeasureAggregators;
-import org.apache.kylin.metadata.datatype.DataType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,6 +62,7 @@ public class GTAggregateScanner implements IGTScanner {
     final ImmutableBitSet metrics;
     final String[] metricsAggrFuncs;
     final IGTScanner inputScanner;
+    final BufferedMeasureCodec measureCodec;
     final AggregationCache aggrCache;
     final long spillThreshold; // 0 means no memory control && no spill
     final int storagePushDownLimit;//default to be Int.MAX
@@ -86,6 +86,7 @@ public GTAggregateScanner(IGTScanner inputScanner, GTScanRequest req, boolean sp
         this.metrics = req.getAggrMetrics();
         this.metricsAggrFuncs = req.getAggrMetricsFuncs();
         this.inputScanner = inputScanner;
+        this.measureCodec = req.createMeasureCodec();
         this.aggrCache = new AggregationCache();
         this.spillThreshold = (long) (req.getAggCacheMemThreshold() * MemoryBudgetController.ONE_GB);
         this.aggrMask = new boolean[metricsAggrFuncs.length];
@@ -175,7 +176,6 @@ class AggregationCache implements Closeable {
         final int keyLength;
         final boolean[] compareMask;
         boolean compareAll = true;
-        final BufferedMeasureCodec measureCodec;
 
         final Comparator<byte[]> bytesComparator = new Comparator<byte[]>() {
             @Override
@@ -213,18 +213,6 @@ public AggregationCache() {
             keyLength = compareMask.length;
             dumps = Lists.newArrayList();
             aggBufMap = createBuffMap();
-            measureCodec = createMeasureCodec();
-        }
-
-        private BufferedMeasureCodec createMeasureCodec() {
-            DataType[] types = new DataType[metrics.trueBitCount()];
-            for (int i = 0; i < types.length; i++) {
-                types[i] = info.getColumnType(metrics.trueBitAt(i));
-            }
-
-            BufferedMeasureCodec result = new BufferedMeasureCodec(types);
-            result.setBufferSize(info.getMaxColumnLength(metrics));
-            return result;
         }
 
         private boolean[] createCompareMask() {

core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java

@@ -25,6 +25,7 @@
 import java.util.NoSuchElementException;
 import java.util.Set;
 
+import org.apache.kylin.GTForwardingScanner;
 import org.apache.kylin.common.util.ByteArray;
 import org.apache.kylin.common.util.BytesUtil;
 import org.apache.kylin.common.util.ImmutableBitSet;
@@ -33,17 +34,16 @@
 import org.apache.kylin.metadata.model.TblColRef;
 import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
 
-public class GTFilterScanner implements IGTScanner {
+public class GTFilterScanner extends GTForwardingScanner {
 
-    final private IGTScanner inputScanner;
     final private TupleFilter filter;
     final private IFilterCodeSystem<ByteArray> filterCodeSystem;
     final private IEvaluatableTuple oneTuple; // avoid instance creation
 
     private GTRecord next = null;
 
-    public GTFilterScanner(IGTScanner inputScanner, GTScanRequest req) throws IOException {
-        this.inputScanner = inputScanner;
+    public GTFilterScanner(IGTScanner delegated, GTScanRequest req) throws IOException {
+        super(delegated);
         this.filter = req.getFilterPushDown();
         this.filterCodeSystem = GTUtil.wrap(getInfo().codeSystem.getComparator());
         this.oneTuple = new IEvaluatableTuple() {
@@ -53,25 +53,15 @@ public Object getValue(TblColRef col) {
             }
         };
 
-        if (TupleFilter.isEvaluableRecursively(filter) == false)
+        if (!TupleFilter.isEvaluableRecursively(filter))
             throw new IllegalArgumentException();
     }
 
-    @Override
-    public GTInfo getInfo() {
-        return inputScanner.getInfo();
-    }
-
-    @Override
-    public void close() throws IOException {
-        inputScanner.close();
-    }
-
     @Override
     public Iterator<GTRecord> iterator() {
         return new Iterator<GTRecord>() {
 
-            private Iterator<GTRecord> inputIterator = inputScanner.iterator();
+            private Iterator<GTRecord> inputIterator = delegated.iterator();
             private FilterResultCache resultCache = new FilterResultCache(getInfo(), filter);
 
             @Override