You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by nj...@apache.org on 2017/09/08 16:10:56 UTC
[9/9] kylin git commit: APACHE-KYLIN-2861: For dictionary building of
lookup table columns, reduce the table scan chance
APACHE-KYLIN-2861: For dictionary building of lookup table columns, reduce the table scan chance
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/54484fe4
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/54484fe4
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/54484fe4
Branch: refs/heads/yaho-cube-planner
Commit: 54484fe422f2accb82e853a0f788bc4595bf05e4
Parents: 45eea45
Author: Zhong <nj...@apache.org>
Authored: Fri Sep 8 23:16:21 2017 +0800
Committer: Zhong <nj...@apache.org>
Committed: Sat Sep 9 00:09:54 2017 +0800
----------------------------------------------------------------------
.../java/org/apache/kylin/cube/CubeManager.java | 39 +++++++++
.../kylin/cube/cli/DictionaryGeneratorCLI.java | 78 +++++++++++-------
.../apache/kylin/dict/DictionaryGenerator.java | 87 +++++++++++++++++++-
.../apache/kylin/dict/DictionaryManager.java | 74 ++++++++++++++++-
.../kylin/dict/IDictionaryValueEnumerator.java | 4 +-
.../TableMultipleColumnValueEnumerator.java | 86 +++++++++++++++++++
.../kylin/engine/mr/SortedColumnReaderTest.java | 10 +--
7 files changed, 337 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
index b815e75..9c0760c 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
@@ -236,6 +236,29 @@ public class CubeManager implements IRealizationProvider {
return dictInfo;
}
+ /**
+ * This method is used to build multiple dictionaries from the same lookup table
+ *
+ * @param cubeSeg
+ * @param colList : pair of column from row key & source column. When storing the dictionary, the column should be from row key rather than source column
+ * @return
+ * @throws IOException
+ */
+ public List<DictionaryInfo> buildDictionaryList(CubeSegment cubeSeg, List<TblColRef> colList,
+ IReadableTable inpTable) throws IOException {
+ CubeDesc cubeDesc = cubeSeg.getCubeDesc();
+
+ List<String> builderClassList = Lists.newArrayListWithExpectedSize(colList.size());
+ for (TblColRef col : colList) {
+ builderClassList.add(cubeDesc.getDictionaryBuilderClass(col));
+ }
+ List<DictionaryInfo> dictInfoList = getDictionaryManager().buildDictionaryList(cubeDesc.getModel(), colList,
+ inpTable, builderClassList);
+
+ saveDictionaryInfoList(cubeSeg, colList, dictInfoList);
+ return dictInfoList;
+ }
+
public DictionaryInfo saveDictionary(CubeSegment cubeSeg, TblColRef col, IReadableTable inpTable,
Dictionary<String> dict) throws IOException {
CubeDesc cubeDesc = cubeSeg.getCubeDesc();
@@ -260,6 +283,22 @@ public class CubeManager implements IRealizationProvider {
}
}
+ private void saveDictionaryInfoList(CubeSegment cubeSeg, List<TblColRef> colList, List<DictionaryInfo> dictInfoList)
+ throws IOException {
+ if (dictInfoList != null && !dictInfoList.isEmpty()) {
+ for (int i = 0; i < dictInfoList.size(); i++) {
+ DictionaryInfo dictInfo = dictInfoList.get(i);
+ TblColRef col = colList.get(i);
+ Dictionary<?> dict = dictInfo.getDictionaryObject();
+ cubeSeg.putDictResPath(col, dictInfo.getResourcePath());
+ cubeSeg.getRowkeyStats().add(new Object[] { col.getName(), dict.getSize(), dict.getSizeOfId() });
+ }
+ CubeUpdate cubeBuilder = new CubeUpdate(cubeSeg.getCubeInstance());
+ cubeBuilder.setToUpdateSegs(cubeSeg);
+ updateCube(cubeBuilder);
+ }
+ }
+
/**
* return null if no dictionary for given column
*/
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java b/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
index 881d87d..726d493 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
@@ -19,10 +19,11 @@
package org.apache.kylin.cube.cli;
import java.io.IOException;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.Dictionary;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
@@ -31,7 +32,6 @@ import org.apache.kylin.dict.DictionaryManager;
import org.apache.kylin.dict.DictionaryProvider;
import org.apache.kylin.dict.DistinctColumnValuesProvider;
import org.apache.kylin.metadata.MetadataManager;
-import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.JoinDesc;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.model.TableRef;
@@ -41,6 +41,8 @@ import org.apache.kylin.source.SourceFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
public class DictionaryGeneratorCLI {
@@ -56,26 +58,57 @@ public class DictionaryGeneratorCLI {
private static void processSegment(KylinConfig config, CubeSegment cubeSeg, DistinctColumnValuesProvider factTableValueProvider, DictionaryProvider dictProvider) throws IOException {
CubeManager cubeMgr = CubeManager.getInstance(config);
+ DictionaryManager dictMgr = DictionaryManager.getInstance(config);
+ MetadataManager metadataManager = MetadataManager.getInstance(config);
+
+ Map<String, List<TblColRef>> lookUpTableColumnsMap = Maps.newHashMap();
+ Map<String, IReadableTable> lookUpTableMap = Maps.newHashMap();
// dictionary
for (TblColRef col : cubeSeg.getCubeDesc().getAllColumnsNeedDictionaryBuilt()) {
- logger.info("Building dictionary for " + col);
- IReadableTable inpTable = decideInputTable(cubeSeg.getModel(), col, factTableValueProvider);
- if (dictProvider != null) {
- Dictionary<String> dict = dictProvider.getDictionary(col);
- if (dict != null) {
- logger.debug("Dict for '" + col.getName() + "' has already been built, save it");
- cubeMgr.saveDictionary(cubeSeg, col, inpTable, dict);
- } else {
- logger.debug("Dict for '" + col.getName() + "' not pre-built, build it from " + inpTable.toString());
- cubeMgr.buildDictionary(cubeSeg, col, inpTable);
+ // decide input table
+ TblColRef srcCol = dictMgr.decideSourceData(cubeSeg.getModel(), col);
+ String srcTable = srcCol.getTable();
+ IReadableTable inpTable;
+ if (cubeSeg.getModel().isFactTable(srcTable)) {
+ inpTable = factTableValueProvider.getDistinctValuesFor(srcCol);
+ } else {
+ inpTable = lookUpTableMap.get(srcTable);
+ if (inpTable == null) {
+ TableDesc tableDesc = new TableDesc(metadataManager.getTableDesc(srcTable));
+ inpTable = SourceFactory.createReadableTable(tableDesc);
+ lookUpTableMap.put(srcTable, inpTable);
}
+ }
+
+ // # save built dictionaries
+ if (dictProvider != null && dictProvider.getDictionary(col) != null) {
+ logger.debug("Dict for '" + col.getName() + "' has already been built, save it");
+ cubeMgr.saveDictionary(cubeSeg, col, inpTable, dictProvider.getDictionary(col));
} else {
- logger.debug("Dict for '" + col.getName() + "' not pre-built, build it from " + inpTable.toString());
- cubeMgr.buildDictionary(cubeSeg, col, inpTable);
+ // # or build dictionaries for fact table columns
+ if (cubeSeg.getModel().isFactTable(srcTable)) {
+ logger.debug("Dict for fact table column '" + col.getName() + "' not pre-built, build it from "
+ + inpTable.toString());
+ cubeMgr.buildDictionary(cubeSeg, col, inpTable);
+ } else {
+ // # or put columns from lookup table to lookUpTableColumnsMap
+ List<TblColRef> lookUpDims = lookUpTableColumnsMap.get(srcTable);
+ if (lookUpDims == null) {
+ lookUpDims = Lists.newArrayList();
+ lookUpTableColumnsMap.put(srcTable, lookUpDims);
+ }
+ lookUpDims.add(col);
+ }
}
}
+ // build dictionaries for lookup table columns
+ for (Map.Entry<String, List<TblColRef>> lookUpEntry : lookUpTableColumnsMap.entrySet()) {
+ logger.info("Batch building dictionaries for lookup table:" + lookUpEntry.getKey());
+ cubeMgr.buildDictionaryList(cubeSeg, lookUpEntry.getValue(), lookUpTableMap.get(lookUpEntry.getKey()));
+ }
+
// snapshot
Set<String> toSnapshot = Sets.newHashSet();
Set<TableRef> toCheckLookup = Sets.newHashSet();
@@ -98,21 +131,4 @@ public class DictionaryGeneratorCLI {
cubeMgr.getLookupTable(cubeSeg, join);
}
}
-
- private static IReadableTable decideInputTable(DataModelDesc model, TblColRef col, DistinctColumnValuesProvider factTableValueProvider) {
- KylinConfig config = model.getConfig();
- DictionaryManager dictMgr = DictionaryManager.getInstance(config);
- TblColRef srcCol = dictMgr.decideSourceData(model, col);
- String srcTable = srcCol.getTable();
-
- IReadableTable inpTable;
- if (model.isFactTable(srcTable)) {
- inpTable = factTableValueProvider.getDistinctValuesFor(srcCol);
- } else {
- MetadataManager metadataManager = MetadataManager.getInstance(config);
- TableDesc tableDesc = new TableDesc(metadataManager.getTableDesc(srcTable));
- inpTable = SourceFactory.createReadableTable(tableDesc);
- }
- return inpTable;
- }
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
index 61a0664..9c87d98 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
@@ -21,6 +21,7 @@ package org.apache.kylin.dict;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.kylin.common.KylinConfig;
@@ -31,6 +32,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
/**
* @author yangli9
@@ -60,11 +63,13 @@ public class DictionaryGenerator {
return builder;
}
- public static Dictionary<String> buildDictionary(DataType dataType, IDictionaryValueEnumerator valueEnumerator) throws IOException {
+ public static Dictionary<String> buildDictionary(DataType dataType,
+ IDictionaryValueEnumerator<String> valueEnumerator) throws IOException {
return buildDictionary(newDictionaryBuilder(dataType), null, valueEnumerator);
}
- static Dictionary<String> buildDictionary(IDictionaryBuilder builder, DictionaryInfo dictInfo, IDictionaryValueEnumerator valueEnumerator) throws IOException {
+ static Dictionary<String> buildDictionary(IDictionaryBuilder builder, DictionaryInfo dictInfo,
+ IDictionaryValueEnumerator<String> valueEnumerator) throws IOException {
int baseId = 0; // always 0 for now
int nSamples = 5;
ArrayList<String> samples = new ArrayList<String>(nSamples);
@@ -100,6 +105,84 @@ public class DictionaryGenerator {
return dict;
}
+ /**
+ * Build multiple dictionaries from the same Lookup table
+ * @throws IOException
+ */
+ public static List<Dictionary<String>> buildDictionaryList(List<IDictionaryBuilder> builderList,
+ List<DictionaryInfo> dictInfoList, int[] colIndexArray,
+ IDictionaryValueEnumerator<Map<Integer, String>> valueEnumerator) throws IOException {
+ if (builderList.size() != dictInfoList.size()) {
+ throw new IllegalArgumentException(
+ "the builderList size does not match with the dictInfoList size, builderList: " + builderList.size()
+ + ", dictInfoList: " + dictInfoList.size());
+ }
+ if (builderList.size() != colIndexArray.length) {
+ throw new IllegalArgumentException(
+ "the builderList size does not match with the colIndexArray length, builderList: "
+ + builderList.size() + ", colIndexArray: " + colIndexArray.length);
+ }
+
+ int baseId = 0; // always 0 for now
+ int nSamples = 5;
+ Map<Integer, List<String>> samplesMap = Maps.newHashMapWithExpectedSize(builderList.size());
+
+ // init the builder map
+ Map<Integer, IDictionaryBuilder> builderMap = Maps.newHashMap();
+ for (int i = 0; i < colIndexArray.length; i++) {
+ int colIdx = colIndexArray[i];
+ IDictionaryBuilder builder = builderList.get(i);
+ builder.init(dictInfoList.get(i), baseId);
+ builderMap.put(colIdx, builder);
+ samplesMap.put(colIdx, Lists.<String> newArrayListWithExpectedSize(nSamples));
+ }
+
+ // add values
+ while (valueEnumerator.moveNext()) {
+ Map<Integer, String> valuesMap = valueEnumerator.current();
+ if (valuesMap == null) {
+ continue;
+ }
+ //Iterate the value map and get the value for each col
+ for (Map.Entry<Integer, String> entry : valuesMap.entrySet()) {
+ int colIdx = entry.getKey();
+ String value = entry.getValue();
+ if (value == null) {
+ continue;
+ }
+ builderMap.get(colIdx).addValue(value);
+
+ List<String> samples = samplesMap.get(colIdx);
+ if (samples.size() < nSamples && samples.contains(value) == false) {
+ samples.add(value);
+ }
+ }
+ }
+
+ List<Dictionary<String>> dictList = Lists.newArrayListWithExpectedSize(builderList.size());
+ for (int colIndex : colIndexArray) {
+ // build
+ IDictionaryBuilder builder = builderMap.get(colIndex);
+ Dictionary<String> dict = builder.build();
+ dictList.add(dict);
+
+ // log a few samples
+ List<String> samples = samplesMap.get(colIndex);
+ StringBuilder buf = new StringBuilder();
+ for (String s : samples) {
+ if (buf.length() > 0) {
+ buf.append(", ");
+ }
+ buf.append(s.toString()).append("=>").append(dict.getIdFromValue(s));
+ }
+ logger.debug("Dictionary value samples: " + buf.toString());
+ logger.debug("Dictionary cardinality: " + dict.getSize());
+ logger.debug("Dictionary builder class: " + builder.getClass().getName());
+ logger.debug("Dictionary class: " + dict.getClass().getName());
+ }
+ return dictList;
+ }
+
public static Dictionary mergeDictionaries(DataType dataType, List<DictionaryInfo> sourceDicts) throws IOException {
return buildDictionary(dataType, new MultipleDictionaryValueEnumerator(sourceDicts));
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
index 857ee30..fdf10d7 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
@@ -23,6 +23,7 @@ import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.NavigableSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -303,7 +304,7 @@ public class DictionaryManager {
private Dictionary<String> buildDictFromReadableTable(IReadableTable inpTable, DictionaryInfo dictInfo, String builderClass, TblColRef col) throws IOException {
Dictionary<String> dictionary;
- IDictionaryValueEnumerator columnValueEnumerator = null;
+ IDictionaryValueEnumerator<String> columnValueEnumerator = null;
try {
columnValueEnumerator = new TableColumnValueEnumerator(inpTable.getReader(), dictInfo.getSourceColumnIndex());
if (builderClass == null) {
@@ -321,6 +322,77 @@ public class DictionaryManager {
return dictionary;
}
+ public List<DictionaryInfo> buildDictionaryList(DataModelDesc model, List<TblColRef> colList,
+ IReadableTable inpTable, List<String> builderClassList) throws IOException {
+ if (inpTable.exists() == false)
+ return null;
+
+ logger.info("Batch building dictionaries for " + colList);
+
+ List<DictionaryInfo> resultList = Lists.newArrayListWithExpectedSize(colList.size());
+ List<DictionaryInfo> newDictList = Lists.newLinkedList();
+ List<IDictionaryBuilder> newDictBuilderList = Lists.newLinkedList();
+ List<Integer> newBuildIndex = Lists.newArrayList();
+
+ for (int i = 0; i < colList.size(); i++) {
+ DictionaryInfo dictInfo = createDictionaryInfo(model, colList.get(i), inpTable);
+
+ String dupInfo = checkDupByInfo(dictInfo);
+ if (dupInfo != null) {
+ logger.info("Identical dictionary input " + dictInfo.getInput() + ", reuse existing dictionary at "
+ + dupInfo);
+ resultList.add(getDictionaryInfo(dupInfo));
+ } else {
+ String builderClass = builderClassList.get(i);
+ if (builderClass == null) {
+ newDictBuilderList
+ .add(DictionaryGenerator.newDictionaryBuilder(DataType.getType(dictInfo.getDataType())));
+ } else {
+ newDictBuilderList.add((IDictionaryBuilder) ClassUtil.newInstance(builderClass));
+ }
+ resultList.add(dictInfo);
+ newDictList.add(dictInfo);
+ newBuildIndex.add(i);
+ }
+ }
+
+ if (newDictList.isEmpty()) {
+ logger.info("all of the columns will reuse existing dictionaries and no need to read table " + inpTable);
+ return resultList;
+ }
+
+ // currently all data types are casted to string to build dictionary
+ // String dataType = info.getDataType();
+ int[] colIndexArray = new int[newDictList.size()];
+ List<DataType> dataTypeList = Lists.newArrayList();
+ for (int i = 0; i < newDictList.size(); i++) {
+ DictionaryInfo info = newDictList.get(i);
+ colIndexArray[i] = info.getSourceColumnIndex();
+ dataTypeList.add(DataType.getType(info.getDataType()));
+ }
+
+ IDictionaryValueEnumerator<Map<Integer, String>> multipleColumnValueEnumerator = null;
+ try {
+ multipleColumnValueEnumerator = new TableMultipleColumnValueEnumerator(inpTable.getReader(), colIndexArray);
+ List<Dictionary<String>> dictionaryList = DictionaryGenerator.buildDictionaryList(newDictBuilderList,
+ newDictList, colIndexArray, multipleColumnValueEnumerator);
+ for (int i = 0; i < newDictList.size(); i++) {
+ DictionaryInfo savedDictInfo = trySaveNewDict(dictionaryList.get(i), newDictList.get(i));
+ resultList.set(newBuildIndex.get(i), savedDictInfo);
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException("Failed to create dictionaries on " + colList, ex);
+ } finally {
+ if (multipleColumnValueEnumerator != null)
+ multipleColumnValueEnumerator.close();
+ }
+
+ if (resultList.size() != colList.size()) {
+ throw new IllegalStateException("Illegal result size in buildDictionaryList.");
+ }
+ return resultList;
+ }
+
public DictionaryInfo saveDictionary(DataModelDesc model, TblColRef col, IReadableTable inpTable, Dictionary<String> dictionary) throws IOException {
DictionaryInfo dictInfo = createDictionaryInfo(model, col, inpTable);
String dupInfo = checkDupByInfo(dictInfo);
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-dictionary/src/main/java/org/apache/kylin/dict/IDictionaryValueEnumerator.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/IDictionaryValueEnumerator.java b/core-dictionary/src/main/java/org/apache/kylin/dict/IDictionaryValueEnumerator.java
index f193d4f..29425c5 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/IDictionaryValueEnumerator.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/IDictionaryValueEnumerator.java
@@ -23,9 +23,9 @@ import java.io.IOException;
/**
* Created by dongli on 10/28/15.
*/
-public interface IDictionaryValueEnumerator {
+public interface IDictionaryValueEnumerator<E> {
- String current() throws IOException;
+ E current() throws IOException;
boolean moveNext() throws IOException;
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/core-dictionary/src/main/java/org/apache/kylin/dict/TableMultipleColumnValueEnumerator.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/TableMultipleColumnValueEnumerator.java b/core-dictionary/src/main/java/org/apache/kylin/dict/TableMultipleColumnValueEnumerator.java
new file mode 100755
index 0000000..7338603
--- /dev/null
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/TableMultipleColumnValueEnumerator.java
@@ -0,0 +1,86 @@
+/*
+ * 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.dict;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+
+import org.apache.kylin.source.IReadableTable;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Created by mingmwang. optimize to read multiple cols.
+ */
+public class TableMultipleColumnValueEnumerator implements IDictionaryValueEnumerator<Map<Integer, String>> {
+
+ private IReadableTable.TableReader reader;
+ private int[] colIndexes;
+ private final int colMaxIndex;
+ private Map<Integer, String> colValueMap = Maps.newHashMap();
+
+ public TableMultipleColumnValueEnumerator(IReadableTable.TableReader reader, int[] colIndexes) {
+ this.reader = reader;
+ this.colIndexes = colIndexes;
+ int tmpColMaxIndex = -1;
+ for (int colIndex : colIndexes) {
+ if (colIndex > tmpColMaxIndex) {
+ tmpColMaxIndex = colIndex;
+ }
+ }
+ this.colMaxIndex = tmpColMaxIndex;
+ }
+
+ @Override
+ public boolean moveNext() throws IOException {
+ if (reader.next()) {
+ colValueMap.clear();
+ String[] split = reader.getRow();
+ if (split.length == 1 && colIndexes.length == 1) {
+ colValueMap.put(colIndexes[0], split[0]);
+ } else {
+ // normal case
+ if (split.length <= colMaxIndex) {
+ throw new ArrayIndexOutOfBoundsException("Column no. " + Arrays.asList(colIndexes)
+ + " not found, line split is " + Arrays.asList(split));
+ }
+ for (int i = 0; i < colIndexes.length; i++) {
+ colValueMap.put(colIndexes[i], split[colIndexes[i]]);
+ }
+ }
+ return true;
+
+ } else {
+ colValueMap = null;
+ return false;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null)
+ reader.close();
+ }
+
+ @Override
+ public Map<Integer, String> current() {
+ return colValueMap;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/54484fe4/engine-mr/src/test/java/org/apache/kylin/engine/mr/SortedColumnReaderTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/SortedColumnReaderTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/SortedColumnReaderTest.java
index 39c5bac..419563f 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/SortedColumnReaderTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/SortedColumnReaderTest.java
@@ -51,7 +51,7 @@ public class SortedColumnReaderTest {
ArrayList<String> correctAnswer = readAllFiles(dirPath);
Collections.sort(correctAnswer, new ByteComparator<String>(new StringBytesConverter()));
SortedColumnDFSFile column = new SortedColumnDFSFile(qualify(dirPath + "/"), DataType.getType("varchar"));
- IDictionaryValueEnumerator e = new TableColumnValueEnumerator(column.getReader(), -1);
+ IDictionaryValueEnumerator<String> e = new TableColumnValueEnumerator(column.getReader(), -1);
ArrayList<String> output = new ArrayList<>();
while (e.moveNext()) {
output.add(new String(e.current()));
@@ -124,7 +124,7 @@ public class SortedColumnReaderTest {
}
});
SortedColumnDFSFile column = new SortedColumnDFSFile(qualify(dirPath + "/"), DataType.getType("long"));
- IDictionaryValueEnumerator e = new TableColumnValueEnumerator(column.getReader(), -1);
+ IDictionaryValueEnumerator<String> e = new TableColumnValueEnumerator(column.getReader(), -1);
ArrayList<String> output = new ArrayList<>();
while (e.moveNext()) {
output.add(new String(e.current()));
@@ -141,7 +141,7 @@ public class SortedColumnReaderTest {
String dirPath = "src/test/resources/empty_dir";
new File(dirPath).mkdirs();
SortedColumnDFSFile column = new SortedColumnDFSFile(qualify(dirPath + "/"), DataType.getType("varchar"));
- IDictionaryValueEnumerator e = new TableColumnValueEnumerator(column.getReader(), -1);
+ IDictionaryValueEnumerator<String> e = new TableColumnValueEnumerator(column.getReader(), -1);
ArrayList<String> output = new ArrayList<>();
while (e.moveNext()) {
System.out.println(new String(e.current()));
@@ -158,7 +158,7 @@ public class SortedColumnReaderTest {
Collections.sort(correctAnswer, new ByteComparator<String>(new StringBytesConverter()));
System.out.println("correct answer:" + correctAnswer);
SortedColumnDFSFile column = new SortedColumnDFSFile(qualify(dirPath + "/"), DataType.getType("varchar"));
- IDictionaryValueEnumerator e = new TableColumnValueEnumerator(column.getReader(), -1);
+ IDictionaryValueEnumerator<String> e = new TableColumnValueEnumerator(column.getReader(), -1);
ArrayList<String> output = new ArrayList<>();
while (e.moveNext()) {
output.add(new String(e.current()));
@@ -229,7 +229,7 @@ public class SortedColumnReaderTest {
}
});
SortedColumnDFSFile column = new SortedColumnDFSFile(qualify(dirPath + "/"), DataType.getType("double"));
- IDictionaryValueEnumerator e = new TableColumnValueEnumerator(column.getReader(), -1);
+ IDictionaryValueEnumerator<String> e = new TableColumnValueEnumerator(column.getReader(), -1);
ArrayList<String> output = new ArrayList<>();
while (e.moveNext()) {
output.add(new String(e.current()));