You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2017/09/15 14:42:53 UTC
[33/48] kylin git commit: KYLIN-2783 Refactor CuboidScheduler to be
extensible
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
index e160d27..417697d 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
@@ -81,7 +81,7 @@ public class CubeStatsReader {
public CubeStatsReader(CubeSegment cubeSegment, KylinConfig kylinConfig) throws IOException {
ResourceStore store = ResourceStore.getStore(kylinConfig);
- cuboidScheduler = cubeSegment.getCubeDesc().getCuboidScheduler();
+ cuboidScheduler = cubeSegment.getCuboidScheduler();
String statsKey = cubeSegment.getStatisticsResourcePath();
File tmpSeqFile = writeTmpSeqFile(store.getResource(statsKey).inputStream);
Reader reader = null;
@@ -174,8 +174,7 @@ public class CubeStatsReader {
public static Map<Long, Double> getCuboidSizeMapFromRowCount(CubeSegment cubeSegment, Map<Long, Long> rowCountMap) {
final CubeDesc cubeDesc = cubeSegment.getCubeDesc();
final List<Integer> rowkeyColumnSize = Lists.newArrayList();
- final long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- final Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
+ final Cuboid baseCuboid = Cuboid.getBaseCuboid(cubeDesc);
final List<TblColRef> columnList = baseCuboid.getColumns();
final CubeDimEncMap dimEncMap = cubeSegment.getDimensionEncodingMap();
@@ -185,7 +184,7 @@ public class CubeStatsReader {
Map<Long, Double> sizeMap = Maps.newHashMap();
for (Map.Entry<Long, Long> entry : rowCountMap.entrySet()) {
- sizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeSegment, entry.getKey(), entry.getValue(), baseCuboidId, rowkeyColumnSize));
+ sizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeSegment, entry.getKey(), entry.getValue(), baseCuboid.getId(), rowkeyColumnSize));
}
return sizeMap;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
index a9c7833..bcd6df0 100755
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
@@ -88,7 +88,7 @@ public class FactDistinctColumnsMapper<KEYIN> extends FactDistinctColumnsMapperB
collectStatistics = Boolean.parseBoolean(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_ENABLED));
if (collectStatistics) {
samplingPercentage = Integer.parseInt(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_SAMPLING_PERCENT));
- cuboidScheduler = cubeDesc.getCuboidScheduler();
+ cuboidScheduler = cubeDesc.getInitialCuboidScheduler();
nRowKey = cubeDesc.getRowkey().getRowKeyColumns().length;
List<Long> cuboidIdList = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
index eee189c..5511414 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
@@ -96,7 +96,8 @@ public class InMemCuboidMapper<KEYIN> extends KylinMapper<KEYIN, Object, ByteArr
}
int taskCount = config.getCubeAlgorithmInMemConcurrentThreads();
- DoggedCubeBuilder cubeBuilder = new DoggedCubeBuilder(cube.getDescriptor(), flatDesc, dictionaryMap);
+ DoggedCubeBuilder cubeBuilder = new DoggedCubeBuilder(cubeSegment.getCuboidScheduler(), flatDesc,
+ dictionaryMap);
cubeBuilder.setReserveMemoryMB(calculateReserveMB(context.getConfiguration()));
cubeBuilder.setConcurrentThreads(taskCount);
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
index aa323fd..2058bc9 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
@@ -95,7 +95,7 @@ public abstract class KVGTRecordWriter implements ICuboidWriter {
protected abstract void writeAsKeyValue(ByteArrayWritable key, ByteArrayWritable value) throws IOException;
private void initVariables(Long cuboidId) {
- rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeDesc, cuboidId));
+ rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeSegment, cuboidId));
keyBuf = rowKeyEncoder.createBuf();
dimensions = Long.bitCount(cuboidId);
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
index 4a579b8..97f5961 100755
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
@@ -147,7 +147,7 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
@Override
public void doMap(Text key, Text value, Context context) throws IOException, InterruptedException {
long cuboidID = rowKeySplitter.split(key.getBytes());
- Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidID);
+ Cuboid cuboid = Cuboid.findById(cube, cuboidID);
RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
SplittedBytes[] splittedByteses = rowKeySplitter.getSplitBuffers();
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/NDCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/NDCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/NDCuboidMapper.java
index 782ce72..6680fd7 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/NDCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/NDCuboidMapper.java
@@ -75,7 +75,7 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
cubeDesc = cube.getDescriptor();
ndCuboidBuilder = new NDCuboidBuilder(cubeSegment);
// initialize CubiodScheduler
- cuboidScheduler = cubeDesc.getCuboidScheduler();
+ cuboidScheduler = cubeSegment.getCuboidScheduler();
rowKeySplitter = new RowKeySplitter(cubeSegment, 65, 256);
}
@@ -84,7 +84,7 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
@Override
public void doMap(Text key, Text value, Context context) throws IOException, InterruptedException {
long cuboidId = rowKeySplitter.split(key.getBytes());
- Cuboid parentCuboid = Cuboid.findById(cubeDesc, cuboidId);
+ Cuboid parentCuboid = Cuboid.findById(cuboidScheduler, cuboidId);
Collection<Long> myChildren = cuboidScheduler.getSpanningCuboid(cuboidId);
@@ -104,7 +104,7 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
}
for (Long child : myChildren) {
- Cuboid childCuboid = Cuboid.findById(cubeDesc, child);
+ Cuboid childCuboid = Cuboid.findById(cuboidScheduler, child);
Pair<Integer, ByteArray> result = ndCuboidBuilder.buildKey(parentCuboid, childCuboid, rowKeySplitter.getSplitBuffers());
outputKey.set(result.getSecond().array(), 0, result.getFirst());
context.write(outputKey, value);
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
index a094cc2..fd53c5a 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
@@ -189,7 +189,7 @@ public class SparkCubing extends AbstractApplication {
final CubeDesc cubeDesc = cubeInstance.getDescriptor();
final HashMap<Integer, TblColRef> tblColRefMap = Maps.newHashMap();
final CubeJoinedFlatTableEnrich flatDesc = new CubeJoinedFlatTableEnrich(EngineFactory.getJoinedFlatTableDesc(seg), cubeDesc);
- final List<TblColRef> baseCuboidColumn = Cuboid.findById(cubeDesc, Cuboid.getBaseCuboidId(cubeDesc)).getColumns();
+ final List<TblColRef> baseCuboidColumn = Cuboid.findById(seg, Cuboid.getBaseCuboidId(cubeDesc)).getColumns();
final long start = System.currentTimeMillis();
final RowKeyDesc rowKey = cubeDesc.getRowkey();
for (int i = 0; i < baseCuboidColumn.size(); i++) {
@@ -253,7 +253,7 @@ public class SparkCubing extends AbstractApplication {
CubeInstance cubeInstance = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()).reloadCubeLocal(cubeName);
CubeSegment cubeSegment = cubeInstance.getSegmentById(segmentId);
CubeDesc cubeDesc = cubeInstance.getDescriptor();
- CuboidScheduler cuboidScheduler = cubeDesc.getCuboidScheduler();
+ CuboidScheduler cuboidScheduler = cubeInstance.getCuboidScheduler();
Set<Long> allCuboidIds = cuboidScheduler.getAllCuboidIds();
final HashMap<Long, HLLCounter> zeroValue = Maps.newHashMap();
for (Long id : allCuboidIds) {
@@ -335,7 +335,7 @@ public class SparkCubing extends AbstractApplication {
CubeInstance cubeInstance = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()).getCube(cubeName);
CubeDesc cubeDesc = cubeInstance.getDescriptor();
final CubeSegment cubeSegment = cubeInstance.getSegmentById(segmentId);
- List<TblColRef> baseCuboidColumn = Cuboid.findById(cubeDesc, Cuboid.getBaseCuboidId(cubeDesc)).getColumns();
+ List<TblColRef> baseCuboidColumn = Cuboid.findById(cubeSegment, Cuboid.getBaseCuboidId(cubeDesc)).getColumns();
final Map<TblColRef, Integer> columnLengthMap = Maps.newHashMap();
final CubeDimEncMap dimEncMap = cubeSegment.getDimensionEncodingMap();
for (TblColRef tblColRef : baseCuboidColumn) {
@@ -377,7 +377,8 @@ public class SparkCubing extends AbstractApplication {
LinkedBlockingQueue<List<String>> blockingQueue = new LinkedBlockingQueue();
System.out.println("load properties finished");
IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cubeSegment);
- AbstractInMemCubeBuilder inMemCubeBuilder = new DoggedCubeBuilder(cubeInstance.getDescriptor(), flatDesc, dictionaryMap);
+ AbstractInMemCubeBuilder inMemCubeBuilder = new DoggedCubeBuilder(
+ cubeSegment.getCuboidScheduler(), flatDesc, dictionaryMap);
final SparkCuboidWriter sparkCuboidWriter = new BufferedCuboidWriter(new DefaultTupleConverter(cubeInstance.getSegmentById(segmentId), columnLengthMap));
Executors.newCachedThreadPool().submit(inMemCubeBuilder.buildAsRunnable(blockingQueue, sparkCuboidWriter));
try {
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubingByLayer.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubingByLayer.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubingByLayer.java
index dab5fb7..15f6424 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubingByLayer.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubingByLayer.java
@@ -191,7 +191,7 @@ public class SparkCubingByLayer extends AbstractApplication implements Serializa
reducerFunction2 = new CuboidReducerFunction2(cubeName, metaUrl, needAggr);
}
- final int totalLevels = cubeDesc.getBuildLevel();
+ final int totalLevels = cubeSegment.getCuboidScheduler().getBuildLevel();
JavaPairRDD<ByteArray, Object[]>[] allRDDs = new JavaPairRDD[totalLevels + 1];
int level = 0;
long baseRDDSize = SizeEstimator.estimate(encodedBaseRDD) / (1024 * 1024);
@@ -298,7 +298,7 @@ public class SparkCubingByLayer extends AbstractApplication implements Serializa
CubeJoinedFlatTableEnrich interDesc = new CubeJoinedFlatTableEnrich(
EngineFactory.getJoinedFlatTableDesc(cubeSegment), cubeDesc);
long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
+ Cuboid baseCuboid = Cuboid.findById(cubeInstance, baseCuboidId);
baseCuboidBuilder = new BaseCuboidBuilder(kConfig, cubeDesc, cubeSegment, interDesc,
AbstractRowKeyEncoder.createInstance(cubeSegment, baseCuboid),
MeasureIngester.create(cubeDesc.getMeasures()), cubeSegment.buildDictionaryMap());
@@ -413,7 +413,7 @@ public class SparkCubingByLayer extends AbstractApplication implements Serializa
CubeInstance cubeInstance = CubeManager.getInstance(kConfig).getCube(cubeName);
this.cubeSegment = cubeInstance.getSegmentById(segmentId);
this.cubeDesc = cubeInstance.getDescriptor();
- this.cuboidScheduler = cubeDesc.getCuboidScheduler();
+ this.cuboidScheduler = cubeSegment.getCuboidScheduler();
this.ndCuboidBuilder = new NDCuboidBuilder(cubeSegment, new RowKeyEncoderProvider(cubeSegment));
this.rowKeySplitter = new RowKeySplitter(cubeSegment, 65, 256);
}
@@ -431,9 +431,9 @@ public class SparkCubingByLayer extends AbstractApplication implements Serializa
byte[] key = tuple2._1().array();
long cuboidId = rowKeySplitter.split(key);
- Cuboid parentCuboid = Cuboid.findById(cubeDesc, cuboidId);
+ Cuboid parentCuboid = Cuboid.findById(cubeSegment, cuboidId);
- Collection<Long> myChildren = cuboidScheduler.getSpanningCuboid(cuboidId);
+ Collection<Long> myChildren = cubeSegment.getCuboidScheduler().getSpanningCuboid(cuboidId);
// if still empty or null
if (myChildren == null || myChildren.size() == 0) {
@@ -442,7 +442,7 @@ public class SparkCubingByLayer extends AbstractApplication implements Serializa
List<Tuple2<ByteArray, Object[]>> tuples = new ArrayList(myChildren.size());
for (Long child : myChildren) {
- Cuboid childCuboid = Cuboid.findById(cubeDesc, child);
+ Cuboid childCuboid = Cuboid.findById(cubeSegment, child);
Pair<Integer, ByteArray> result = ndCuboidBuilder.buildKey(parentCuboid, childCuboid,
rowKeySplitter.getSplitBuffers());
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
index 793cd87..d7b20c8 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
@@ -65,7 +65,7 @@ public final class DefaultTupleConverter implements TupleConverter {
@Override
public final Tuple2<byte[], byte[]> convert(long cuboidId, GTRecord record) {
- Cuboid cuboid = Cuboid.findById(segment.getCubeDesc(), cuboidId);
+ Cuboid cuboid = Cuboid.findById(segment, cuboidId);
RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
final int dimensions = Long.bitCount(cuboidId);
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderStressTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderStressTest.java b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderStressTest.java
index a19b8c7..5e5b16a 100644
--- a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderStressTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderStressTest.java
@@ -84,7 +84,7 @@ public class ITDoggedCubeBuilderStressTest extends LocalFileMetadataTestCase {
long randSeed = System.currentTimeMillis();
IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cube.getDescriptor());
- DoggedCubeBuilder doggedBuilder = new DoggedCubeBuilder(cube.getDescriptor(), flatDesc, dictionaryMap);
+ DoggedCubeBuilder doggedBuilder = new DoggedCubeBuilder(cube.getCuboidScheduler(), flatDesc, dictionaryMap);
doggedBuilder.setConcurrentThreads(THREADS);
{
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
index dbd9ce2..1e10d79 100644
--- a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
@@ -88,7 +88,7 @@ public class ITDoggedCubeBuilderTest extends LocalFileMetadataTestCase {
long randSeed = System.currentTimeMillis();
IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cube.getDescriptor());
- DoggedCubeBuilder doggedBuilder = new DoggedCubeBuilder(cube.getDescriptor(), flatDesc, dictionaryMap);
+ DoggedCubeBuilder doggedBuilder = new DoggedCubeBuilder(cube.getCuboidScheduler(), flatDesc, dictionaryMap);
doggedBuilder.setConcurrentThreads(THREADS);
doggedBuilder.setSplitRowThreshold(SPLIT_ROWS);
FileRecordWriter doggedResult = new FileRecordWriter();
@@ -100,7 +100,7 @@ public class ITDoggedCubeBuilderTest extends LocalFileMetadataTestCase {
doggedResult.close();
}
- InMemCubeBuilder inmemBuilder = new InMemCubeBuilder(cube.getDescriptor(), flatDesc, dictionaryMap);
+ InMemCubeBuilder inmemBuilder = new InMemCubeBuilder(cube.getCuboidScheduler(), flatDesc, dictionaryMap);
inmemBuilder.setConcurrentThreads(THREADS);
FileRecordWriter inmemResult = new FileRecordWriter();
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
index ea66c73..49d267a 100644
--- a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
@@ -109,7 +109,7 @@ public class ITInMemCubeBuilderTest extends LocalFileMetadataTestCase {
private void testBuildInner() throws Exception {
IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cube.getDescriptor());
- InMemCubeBuilder cubeBuilder = new InMemCubeBuilder(cube.getDescriptor(), flatDesc, dictionaryMap);
+ InMemCubeBuilder cubeBuilder = new InMemCubeBuilder(cube.getCuboidScheduler(), flatDesc, dictionaryMap);
//DoggedCubeBuilder cubeBuilder = new DoggedCubeBuilder(cube.getDescriptor(), dictionaryMap);
cubeBuilder.setConcurrentThreads(nThreads);
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseCuboidWriter.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseCuboidWriter.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseCuboidWriter.java
index 6587d4e..59f1349 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseCuboidWriter.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseCuboidWriter.java
@@ -81,7 +81,7 @@ public class HBaseCuboidWriter implements ICuboidWriter {
//TODO:shardingonstreaming
private byte[] createKey(Long cuboidId, GTRecord record) {
if (rowKeyEncoder == null || rowKeyEncoder.getCuboidID() != cuboidId) {
- rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeDesc, cuboidId));
+ rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeSegment, cuboidId));
keybuf = rowKeyEncoder.createBuf();
}
rowKeyEncoder.encode(record, record.getInfo().getPrimaryKey(), keybuf);
@@ -92,7 +92,7 @@ public class HBaseCuboidWriter implements ICuboidWriter {
@Override
public void write(long cuboidId, GTRecord record) throws IOException {
byte[] key = createKey(cuboidId, record);
- final Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidId);
+ final Cuboid cuboid = Cuboid.findById(cubeSegment, cuboidId);
final int nDims = cuboid.getColumns().size();
final ImmutableBitSet bitSet = new ImmutableBitSet(nDims, nDims + cubeDesc.getMeasures().size());
http://git-wip-us.apache.org/repos/asf/kylin/blob/465c5070/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/MeasureTypeOnlyAggrInBaseTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/MeasureTypeOnlyAggrInBaseTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/MeasureTypeOnlyAggrInBaseTest.java
index f151876..7d089ac 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/MeasureTypeOnlyAggrInBaseTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/MeasureTypeOnlyAggrInBaseTest.java
@@ -91,10 +91,10 @@ public class MeasureTypeOnlyAggrInBaseTest extends LocalFileMetadataTestCase {
@Test
public void testIdentifyCuboidV2() throws InvocationTargetException, NoSuchMethodException, IllegalAccessException, NoSuchFieldException {
CubeDesc cubeDesc = cube.getDescriptor();
- Cuboid ret = Cuboid.identifyCuboid(cubeDesc, Sets.<TblColRef> newHashSet(), Lists.<FunctionDesc> newArrayList());
+ Cuboid ret = Cuboid.identifyCuboid(cube, Sets.<TblColRef> newHashSet(), Lists.<FunctionDesc> newArrayList());
long baseCuboidId = cubeDesc.getRowkey().getFullMask();
assertNotEquals(baseCuboidId, ret.getId());
- ret = Cuboid.identifyCuboid(cubeDesc, dimensions, metrics);
+ ret = Cuboid.identifyCuboid(cube, dimensions, metrics);
assertEquals(baseCuboidId, ret.getId());
}