You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2015/11/16 03:23:21 UTC
[1/5] incubator-kylin git commit: KYLIN-1126 pscan backward
compability with v1 storage
Repository: incubator-kylin
Updated Branches:
refs/heads/2.x-staging ae0f1a72e -> 134960c62
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/generated/CubeVisitProtos.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/generated/CubeVisitProtos.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/generated/CubeVisitProtos.java
index 0923484..7d9b313 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/generated/CubeVisitProtos.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/generated/CubeVisitProtos.java
@@ -46,27 +46,37 @@ public final class CubeVisitProtos {
*/
com.google.protobuf.ByteString getHbaseRawScan();
- // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;
+ // required int32 rowkeyPreambleSize = 4;
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ boolean hasRowkeyPreambleSize();
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ int getRowkeyPreambleSize();
+
+ // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;
+ /**
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList>
getHbaseColumnsToGTList();
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList getHbaseColumnsToGT(int index);
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
int getHbaseColumnsToGTCount();
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
java.util.List<? extends org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder>
getHbaseColumnsToGTOrBuilderList();
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder getHbaseColumnsToGTOrBuilder(
int index);
@@ -137,10 +147,15 @@ public final class CubeVisitProtos {
hbaseRawScan_ = input.readBytes();
break;
}
- case 34: {
- if (!((mutable_bitField0_ & 0x00000008) == 0x00000008)) {
+ case 32: {
+ bitField0_ |= 0x00000008;
+ rowkeyPreambleSize_ = input.readInt32();
+ break;
+ }
+ case 42: {
+ if (!((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
hbaseColumnsToGT_ = new java.util.ArrayList<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList>();
- mutable_bitField0_ |= 0x00000008;
+ mutable_bitField0_ |= 0x00000010;
}
hbaseColumnsToGT_.add(input.readMessage(org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.PARSER, extensionRegistry));
break;
@@ -153,7 +168,7 @@ public final class CubeVisitProtos {
throw new com.google.protobuf.InvalidProtocolBufferException(
e.getMessage()).setUnfinishedMessage(this);
} finally {
- if (((mutable_bitField0_ & 0x00000008) == 0x00000008)) {
+ if (((mutable_bitField0_ & 0x00000010) == 0x00000010)) {
hbaseColumnsToGT_ = java.util.Collections.unmodifiableList(hbaseColumnsToGT_);
}
this.unknownFields = unknownFields.build();
@@ -767,36 +782,52 @@ public final class CubeVisitProtos {
return hbaseRawScan_;
}
- // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;
- public static final int HBASECOLUMNSTOGT_FIELD_NUMBER = 4;
+ // required int32 rowkeyPreambleSize = 4;
+ public static final int ROWKEYPREAMBLESIZE_FIELD_NUMBER = 4;
+ private int rowkeyPreambleSize_;
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public boolean hasRowkeyPreambleSize() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public int getRowkeyPreambleSize() {
+ return rowkeyPreambleSize_;
+ }
+
+ // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;
+ public static final int HBASECOLUMNSTOGT_FIELD_NUMBER = 5;
private java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList> hbaseColumnsToGT_;
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList> getHbaseColumnsToGTList() {
return hbaseColumnsToGT_;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public java.util.List<? extends org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder>
getHbaseColumnsToGTOrBuilderList() {
return hbaseColumnsToGT_;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public int getHbaseColumnsToGTCount() {
return hbaseColumnsToGT_.size();
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList getHbaseColumnsToGT(int index) {
return hbaseColumnsToGT_.get(index);
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder getHbaseColumnsToGTOrBuilder(
int index) {
@@ -807,6 +838,7 @@ public final class CubeVisitProtos {
behavior_ = "";
gtScanRequest_ = com.google.protobuf.ByteString.EMPTY;
hbaseRawScan_ = com.google.protobuf.ByteString.EMPTY;
+ rowkeyPreambleSize_ = 0;
hbaseColumnsToGT_ = java.util.Collections.emptyList();
}
private byte memoizedIsInitialized = -1;
@@ -826,6 +858,10 @@ public final class CubeVisitProtos {
memoizedIsInitialized = 0;
return false;
}
+ if (!hasRowkeyPreambleSize()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
memoizedIsInitialized = 1;
return true;
}
@@ -842,8 +878,11 @@ public final class CubeVisitProtos {
if (((bitField0_ & 0x00000004) == 0x00000004)) {
output.writeBytes(3, hbaseRawScan_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ output.writeInt32(4, rowkeyPreambleSize_);
+ }
for (int i = 0; i < hbaseColumnsToGT_.size(); i++) {
- output.writeMessage(4, hbaseColumnsToGT_.get(i));
+ output.writeMessage(5, hbaseColumnsToGT_.get(i));
}
getUnknownFields().writeTo(output);
}
@@ -866,9 +905,13 @@ public final class CubeVisitProtos {
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(3, hbaseRawScan_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeInt32Size(4, rowkeyPreambleSize_);
+ }
for (int i = 0; i < hbaseColumnsToGT_.size(); i++) {
size += com.google.protobuf.CodedOutputStream
- .computeMessageSize(4, hbaseColumnsToGT_.get(i));
+ .computeMessageSize(5, hbaseColumnsToGT_.get(i));
}
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
@@ -908,6 +951,11 @@ public final class CubeVisitProtos {
result = result && getHbaseRawScan()
.equals(other.getHbaseRawScan());
}
+ result = result && (hasRowkeyPreambleSize() == other.hasRowkeyPreambleSize());
+ if (hasRowkeyPreambleSize()) {
+ result = result && (getRowkeyPreambleSize()
+ == other.getRowkeyPreambleSize());
+ }
result = result && getHbaseColumnsToGTList()
.equals(other.getHbaseColumnsToGTList());
result = result &&
@@ -935,6 +983,10 @@ public final class CubeVisitProtos {
hash = (37 * hash) + HBASERAWSCAN_FIELD_NUMBER;
hash = (53 * hash) + getHbaseRawScan().hashCode();
}
+ if (hasRowkeyPreambleSize()) {
+ hash = (37 * hash) + ROWKEYPREAMBLESIZE_FIELD_NUMBER;
+ hash = (53 * hash) + getRowkeyPreambleSize();
+ }
if (getHbaseColumnsToGTCount() > 0) {
hash = (37 * hash) + HBASECOLUMNSTOGT_FIELD_NUMBER;
hash = (53 * hash) + getHbaseColumnsToGTList().hashCode();
@@ -1055,9 +1107,11 @@ public final class CubeVisitProtos {
bitField0_ = (bitField0_ & ~0x00000002);
hbaseRawScan_ = com.google.protobuf.ByteString.EMPTY;
bitField0_ = (bitField0_ & ~0x00000004);
+ rowkeyPreambleSize_ = 0;
+ bitField0_ = (bitField0_ & ~0x00000008);
if (hbaseColumnsToGTBuilder_ == null) {
hbaseColumnsToGT_ = java.util.Collections.emptyList();
- bitField0_ = (bitField0_ & ~0x00000008);
+ bitField0_ = (bitField0_ & ~0x00000010);
} else {
hbaseColumnsToGTBuilder_.clear();
}
@@ -1101,10 +1155,14 @@ public final class CubeVisitProtos {
to_bitField0_ |= 0x00000004;
}
result.hbaseRawScan_ = hbaseRawScan_;
+ if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+ to_bitField0_ |= 0x00000008;
+ }
+ result.rowkeyPreambleSize_ = rowkeyPreambleSize_;
if (hbaseColumnsToGTBuilder_ == null) {
- if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
hbaseColumnsToGT_ = java.util.Collections.unmodifiableList(hbaseColumnsToGT_);
- bitField0_ = (bitField0_ & ~0x00000008);
+ bitField0_ = (bitField0_ & ~0x00000010);
}
result.hbaseColumnsToGT_ = hbaseColumnsToGT_;
} else {
@@ -1137,11 +1195,14 @@ public final class CubeVisitProtos {
if (other.hasHbaseRawScan()) {
setHbaseRawScan(other.getHbaseRawScan());
}
+ if (other.hasRowkeyPreambleSize()) {
+ setRowkeyPreambleSize(other.getRowkeyPreambleSize());
+ }
if (hbaseColumnsToGTBuilder_ == null) {
if (!other.hbaseColumnsToGT_.isEmpty()) {
if (hbaseColumnsToGT_.isEmpty()) {
hbaseColumnsToGT_ = other.hbaseColumnsToGT_;
- bitField0_ = (bitField0_ & ~0x00000008);
+ bitField0_ = (bitField0_ & ~0x00000010);
} else {
ensureHbaseColumnsToGTIsMutable();
hbaseColumnsToGT_.addAll(other.hbaseColumnsToGT_);
@@ -1154,7 +1215,7 @@ public final class CubeVisitProtos {
hbaseColumnsToGTBuilder_.dispose();
hbaseColumnsToGTBuilder_ = null;
hbaseColumnsToGT_ = other.hbaseColumnsToGT_;
- bitField0_ = (bitField0_ & ~0x00000008);
+ bitField0_ = (bitField0_ & ~0x00000010);
hbaseColumnsToGTBuilder_ =
com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ?
getHbaseColumnsToGTFieldBuilder() : null;
@@ -1180,6 +1241,10 @@ public final class CubeVisitProtos {
return false;
}
+ if (!hasRowkeyPreambleSize()) {
+
+ return false;
+ }
return true;
}
@@ -1348,13 +1413,46 @@ public final class CubeVisitProtos {
return this;
}
- // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;
+ // required int32 rowkeyPreambleSize = 4;
+ private int rowkeyPreambleSize_ ;
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public boolean hasRowkeyPreambleSize() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public int getRowkeyPreambleSize() {
+ return rowkeyPreambleSize_;
+ }
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public Builder setRowkeyPreambleSize(int value) {
+ bitField0_ |= 0x00000008;
+ rowkeyPreambleSize_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int32 rowkeyPreambleSize = 4;</code>
+ */
+ public Builder clearRowkeyPreambleSize() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ rowkeyPreambleSize_ = 0;
+ onChanged();
+ return this;
+ }
+
+ // repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;
private java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList> hbaseColumnsToGT_ =
java.util.Collections.emptyList();
private void ensureHbaseColumnsToGTIsMutable() {
- if (!((bitField0_ & 0x00000008) == 0x00000008)) {
+ if (!((bitField0_ & 0x00000010) == 0x00000010)) {
hbaseColumnsToGT_ = new java.util.ArrayList<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList>(hbaseColumnsToGT_);
- bitField0_ |= 0x00000008;
+ bitField0_ |= 0x00000010;
}
}
@@ -1362,7 +1460,7 @@ public final class CubeVisitProtos {
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder> hbaseColumnsToGTBuilder_;
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList> getHbaseColumnsToGTList() {
if (hbaseColumnsToGTBuilder_ == null) {
@@ -1372,7 +1470,7 @@ public final class CubeVisitProtos {
}
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public int getHbaseColumnsToGTCount() {
if (hbaseColumnsToGTBuilder_ == null) {
@@ -1382,7 +1480,7 @@ public final class CubeVisitProtos {
}
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList getHbaseColumnsToGT(int index) {
if (hbaseColumnsToGTBuilder_ == null) {
@@ -1392,7 +1490,7 @@ public final class CubeVisitProtos {
}
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder setHbaseColumnsToGT(
int index, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList value) {
@@ -1409,7 +1507,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder setHbaseColumnsToGT(
int index, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder builderForValue) {
@@ -1423,7 +1521,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder addHbaseColumnsToGT(org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList value) {
if (hbaseColumnsToGTBuilder_ == null) {
@@ -1439,7 +1537,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder addHbaseColumnsToGT(
int index, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList value) {
@@ -1456,7 +1554,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder addHbaseColumnsToGT(
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder builderForValue) {
@@ -1470,7 +1568,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder addHbaseColumnsToGT(
int index, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder builderForValue) {
@@ -1484,7 +1582,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder addAllHbaseColumnsToGT(
java.lang.Iterable<? extends org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList> values) {
@@ -1498,12 +1596,12 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder clearHbaseColumnsToGT() {
if (hbaseColumnsToGTBuilder_ == null) {
hbaseColumnsToGT_ = java.util.Collections.emptyList();
- bitField0_ = (bitField0_ & ~0x00000008);
+ bitField0_ = (bitField0_ & ~0x00000010);
onChanged();
} else {
hbaseColumnsToGTBuilder_.clear();
@@ -1511,7 +1609,7 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public Builder removeHbaseColumnsToGT(int index) {
if (hbaseColumnsToGTBuilder_ == null) {
@@ -1524,14 +1622,14 @@ public final class CubeVisitProtos {
return this;
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder getHbaseColumnsToGTBuilder(
int index) {
return getHbaseColumnsToGTFieldBuilder().getBuilder(index);
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder getHbaseColumnsToGTOrBuilder(
int index) {
@@ -1541,7 +1639,7 @@ public final class CubeVisitProtos {
}
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public java.util.List<? extends org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder>
getHbaseColumnsToGTOrBuilderList() {
@@ -1552,14 +1650,14 @@ public final class CubeVisitProtos {
}
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder addHbaseColumnsToGTBuilder() {
return getHbaseColumnsToGTFieldBuilder().addBuilder(
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.getDefaultInstance());
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder addHbaseColumnsToGTBuilder(
int index) {
@@ -1567,7 +1665,7 @@ public final class CubeVisitProtos {
index, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.getDefaultInstance());
}
/**
- * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 4;</code>
+ * <code>repeated .CubeVisitRequest.IntList hbaseColumnsToGT = 5;</code>
*/
public java.util.List<org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder>
getHbaseColumnsToGTBuilderList() {
@@ -1580,7 +1678,7 @@ public final class CubeVisitProtos {
hbaseColumnsToGTBuilder_ = new com.google.protobuf.RepeatedFieldBuilder<
org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList.Builder, org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntListOrBuilder>(
hbaseColumnsToGT_,
- ((bitField0_ & 0x00000008) == 0x00000008),
+ ((bitField0_ & 0x00000010) == 0x00000010),
getParentForChildren(),
isClean());
hbaseColumnsToGT_ = null;
@@ -1776,6 +1874,51 @@ public final class CubeVisitProtos {
* <code>optional int32 aggregatedRowCount = 4;</code>
*/
int getAggregatedRowCount();
+
+ // optional double systemCpuLoad = 5;
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ boolean hasSystemCpuLoad();
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ double getSystemCpuLoad();
+
+ // optional double freePhysicalMemorySize = 6;
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ boolean hasFreePhysicalMemorySize();
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ double getFreePhysicalMemorySize();
+
+ // optional double freeSwapSpaceSize = 7;
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ boolean hasFreeSwapSpaceSize();
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ double getFreeSwapSpaceSize();
+
+ // optional string hostname = 8;
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ boolean hasHostname();
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ java.lang.String getHostname();
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ com.google.protobuf.ByteString
+ getHostnameBytes();
}
/**
* Protobuf type {@code CubeVisitResponse.Stats}
@@ -1848,6 +1991,26 @@ public final class CubeVisitProtos {
aggregatedRowCount_ = input.readInt32();
break;
}
+ case 41: {
+ bitField0_ |= 0x00000010;
+ systemCpuLoad_ = input.readDouble();
+ break;
+ }
+ case 49: {
+ bitField0_ |= 0x00000020;
+ freePhysicalMemorySize_ = input.readDouble();
+ break;
+ }
+ case 57: {
+ bitField0_ |= 0x00000040;
+ freeSwapSpaceSize_ = input.readDouble();
+ break;
+ }
+ case 66: {
+ bitField0_ |= 0x00000080;
+ hostname_ = input.readBytes();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -1952,11 +2115,106 @@ public final class CubeVisitProtos {
return aggregatedRowCount_;
}
+ // optional double systemCpuLoad = 5;
+ public static final int SYSTEMCPULOAD_FIELD_NUMBER = 5;
+ private double systemCpuLoad_;
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public boolean hasSystemCpuLoad() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public double getSystemCpuLoad() {
+ return systemCpuLoad_;
+ }
+
+ // optional double freePhysicalMemorySize = 6;
+ public static final int FREEPHYSICALMEMORYSIZE_FIELD_NUMBER = 6;
+ private double freePhysicalMemorySize_;
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public boolean hasFreePhysicalMemorySize() {
+ return ((bitField0_ & 0x00000020) == 0x00000020);
+ }
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public double getFreePhysicalMemorySize() {
+ return freePhysicalMemorySize_;
+ }
+
+ // optional double freeSwapSpaceSize = 7;
+ public static final int FREESWAPSPACESIZE_FIELD_NUMBER = 7;
+ private double freeSwapSpaceSize_;
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public boolean hasFreeSwapSpaceSize() {
+ return ((bitField0_ & 0x00000040) == 0x00000040);
+ }
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public double getFreeSwapSpaceSize() {
+ return freeSwapSpaceSize_;
+ }
+
+ // optional string hostname = 8;
+ public static final int HOSTNAME_FIELD_NUMBER = 8;
+ private java.lang.Object hostname_;
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public boolean hasHostname() {
+ return ((bitField0_ & 0x00000080) == 0x00000080);
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public java.lang.String getHostname() {
+ java.lang.Object ref = hostname_;
+ if (ref instanceof java.lang.String) {
+ return (java.lang.String) ref;
+ } else {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ hostname_ = s;
+ }
+ return s;
+ }
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public com.google.protobuf.ByteString
+ getHostnameBytes() {
+ java.lang.Object ref = hostname_;
+ if (ref instanceof java.lang.String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ hostname_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
private void initFields() {
serviceStartTime_ = 0L;
serviceEndTime_ = 0L;
scannedRowCount_ = 0;
aggregatedRowCount_ = 0;
+ systemCpuLoad_ = 0D;
+ freePhysicalMemorySize_ = 0D;
+ freeSwapSpaceSize_ = 0D;
+ hostname_ = "";
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -1982,6 +2240,18 @@ public final class CubeVisitProtos {
if (((bitField0_ & 0x00000008) == 0x00000008)) {
output.writeInt32(4, aggregatedRowCount_);
}
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ output.writeDouble(5, systemCpuLoad_);
+ }
+ if (((bitField0_ & 0x00000020) == 0x00000020)) {
+ output.writeDouble(6, freePhysicalMemorySize_);
+ }
+ if (((bitField0_ & 0x00000040) == 0x00000040)) {
+ output.writeDouble(7, freeSwapSpaceSize_);
+ }
+ if (((bitField0_ & 0x00000080) == 0x00000080)) {
+ output.writeBytes(8, getHostnameBytes());
+ }
getUnknownFields().writeTo(output);
}
@@ -2007,6 +2277,22 @@ public final class CubeVisitProtos {
size += com.google.protobuf.CodedOutputStream
.computeInt32Size(4, aggregatedRowCount_);
}
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeDoubleSize(5, systemCpuLoad_);
+ }
+ if (((bitField0_ & 0x00000020) == 0x00000020)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeDoubleSize(6, freePhysicalMemorySize_);
+ }
+ if (((bitField0_ & 0x00000040) == 0x00000040)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeDoubleSize(7, freeSwapSpaceSize_);
+ }
+ if (((bitField0_ & 0x00000080) == 0x00000080)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(8, getHostnameBytes());
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -2050,6 +2336,23 @@ public final class CubeVisitProtos {
result = result && (getAggregatedRowCount()
== other.getAggregatedRowCount());
}
+ result = result && (hasSystemCpuLoad() == other.hasSystemCpuLoad());
+ if (hasSystemCpuLoad()) {
+ result = result && (Double.doubleToLongBits(getSystemCpuLoad()) == Double.doubleToLongBits(other.getSystemCpuLoad()));
+ }
+ result = result && (hasFreePhysicalMemorySize() == other.hasFreePhysicalMemorySize());
+ if (hasFreePhysicalMemorySize()) {
+ result = result && (Double.doubleToLongBits(getFreePhysicalMemorySize()) == Double.doubleToLongBits(other.getFreePhysicalMemorySize()));
+ }
+ result = result && (hasFreeSwapSpaceSize() == other.hasFreeSwapSpaceSize());
+ if (hasFreeSwapSpaceSize()) {
+ result = result && (Double.doubleToLongBits(getFreeSwapSpaceSize()) == Double.doubleToLongBits(other.getFreeSwapSpaceSize()));
+ }
+ result = result && (hasHostname() == other.hasHostname());
+ if (hasHostname()) {
+ result = result && getHostname()
+ .equals(other.getHostname());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -2079,6 +2382,25 @@ public final class CubeVisitProtos {
hash = (37 * hash) + AGGREGATEDROWCOUNT_FIELD_NUMBER;
hash = (53 * hash) + getAggregatedRowCount();
}
+ if (hasSystemCpuLoad()) {
+ hash = (37 * hash) + SYSTEMCPULOAD_FIELD_NUMBER;
+ hash = (53 * hash) + hashLong(
+ Double.doubleToLongBits(getSystemCpuLoad()));
+ }
+ if (hasFreePhysicalMemorySize()) {
+ hash = (37 * hash) + FREEPHYSICALMEMORYSIZE_FIELD_NUMBER;
+ hash = (53 * hash) + hashLong(
+ Double.doubleToLongBits(getFreePhysicalMemorySize()));
+ }
+ if (hasFreeSwapSpaceSize()) {
+ hash = (37 * hash) + FREESWAPSPACESIZE_FIELD_NUMBER;
+ hash = (53 * hash) + hashLong(
+ Double.doubleToLongBits(getFreeSwapSpaceSize()));
+ }
+ if (hasHostname()) {
+ hash = (37 * hash) + HOSTNAME_FIELD_NUMBER;
+ hash = (53 * hash) + getHostname().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -2196,6 +2518,14 @@ public final class CubeVisitProtos {
bitField0_ = (bitField0_ & ~0x00000004);
aggregatedRowCount_ = 0;
bitField0_ = (bitField0_ & ~0x00000008);
+ systemCpuLoad_ = 0D;
+ bitField0_ = (bitField0_ & ~0x00000010);
+ freePhysicalMemorySize_ = 0D;
+ bitField0_ = (bitField0_ & ~0x00000020);
+ freeSwapSpaceSize_ = 0D;
+ bitField0_ = (bitField0_ & ~0x00000040);
+ hostname_ = "";
+ bitField0_ = (bitField0_ & ~0x00000080);
return this;
}
@@ -2240,6 +2570,22 @@ public final class CubeVisitProtos {
to_bitField0_ |= 0x00000008;
}
result.aggregatedRowCount_ = aggregatedRowCount_;
+ if (((from_bitField0_ & 0x00000010) == 0x00000010)) {
+ to_bitField0_ |= 0x00000010;
+ }
+ result.systemCpuLoad_ = systemCpuLoad_;
+ if (((from_bitField0_ & 0x00000020) == 0x00000020)) {
+ to_bitField0_ |= 0x00000020;
+ }
+ result.freePhysicalMemorySize_ = freePhysicalMemorySize_;
+ if (((from_bitField0_ & 0x00000040) == 0x00000040)) {
+ to_bitField0_ |= 0x00000040;
+ }
+ result.freeSwapSpaceSize_ = freeSwapSpaceSize_;
+ if (((from_bitField0_ & 0x00000080) == 0x00000080)) {
+ to_bitField0_ |= 0x00000080;
+ }
+ result.hostname_ = hostname_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -2268,6 +2614,20 @@ public final class CubeVisitProtos {
if (other.hasAggregatedRowCount()) {
setAggregatedRowCount(other.getAggregatedRowCount());
}
+ if (other.hasSystemCpuLoad()) {
+ setSystemCpuLoad(other.getSystemCpuLoad());
+ }
+ if (other.hasFreePhysicalMemorySize()) {
+ setFreePhysicalMemorySize(other.getFreePhysicalMemorySize());
+ }
+ if (other.hasFreeSwapSpaceSize()) {
+ setFreeSwapSpaceSize(other.getFreeSwapSpaceSize());
+ }
+ if (other.hasHostname()) {
+ bitField0_ |= 0x00000080;
+ hostname_ = other.hostname_;
+ onChanged();
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -2427,6 +2787,179 @@ public final class CubeVisitProtos {
return this;
}
+ // optional double systemCpuLoad = 5;
+ private double systemCpuLoad_ ;
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public boolean hasSystemCpuLoad() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public double getSystemCpuLoad() {
+ return systemCpuLoad_;
+ }
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public Builder setSystemCpuLoad(double value) {
+ bitField0_ |= 0x00000010;
+ systemCpuLoad_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional double systemCpuLoad = 5;</code>
+ */
+ public Builder clearSystemCpuLoad() {
+ bitField0_ = (bitField0_ & ~0x00000010);
+ systemCpuLoad_ = 0D;
+ onChanged();
+ return this;
+ }
+
+ // optional double freePhysicalMemorySize = 6;
+ private double freePhysicalMemorySize_ ;
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public boolean hasFreePhysicalMemorySize() {
+ return ((bitField0_ & 0x00000020) == 0x00000020);
+ }
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public double getFreePhysicalMemorySize() {
+ return freePhysicalMemorySize_;
+ }
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public Builder setFreePhysicalMemorySize(double value) {
+ bitField0_ |= 0x00000020;
+ freePhysicalMemorySize_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional double freePhysicalMemorySize = 6;</code>
+ */
+ public Builder clearFreePhysicalMemorySize() {
+ bitField0_ = (bitField0_ & ~0x00000020);
+ freePhysicalMemorySize_ = 0D;
+ onChanged();
+ return this;
+ }
+
+ // optional double freeSwapSpaceSize = 7;
+ private double freeSwapSpaceSize_ ;
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public boolean hasFreeSwapSpaceSize() {
+ return ((bitField0_ & 0x00000040) == 0x00000040);
+ }
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public double getFreeSwapSpaceSize() {
+ return freeSwapSpaceSize_;
+ }
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public Builder setFreeSwapSpaceSize(double value) {
+ bitField0_ |= 0x00000040;
+ freeSwapSpaceSize_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional double freeSwapSpaceSize = 7;</code>
+ */
+ public Builder clearFreeSwapSpaceSize() {
+ bitField0_ = (bitField0_ & ~0x00000040);
+ freeSwapSpaceSize_ = 0D;
+ onChanged();
+ return this;
+ }
+
+ // optional string hostname = 8;
+ private java.lang.Object hostname_ = "";
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public boolean hasHostname() {
+ return ((bitField0_ & 0x00000080) == 0x00000080);
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public java.lang.String getHostname() {
+ java.lang.Object ref = hostname_;
+ if (!(ref instanceof java.lang.String)) {
+ java.lang.String s = ((com.google.protobuf.ByteString) ref)
+ .toStringUtf8();
+ hostname_ = s;
+ return s;
+ } else {
+ return (java.lang.String) ref;
+ }
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public com.google.protobuf.ByteString
+ getHostnameBytes() {
+ java.lang.Object ref = hostname_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ hostname_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public Builder setHostname(
+ java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000080;
+ hostname_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public Builder clearHostname() {
+ bitField0_ = (bitField0_ & ~0x00000080);
+ hostname_ = getDefaultInstance().getHostname();
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional string hostname = 8;</code>
+ */
+ public Builder setHostnameBytes(
+ com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000080;
+ hostname_ = value;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:CubeVisitResponse.Stats)
}
@@ -3220,21 +3753,24 @@ public final class CubeVisitProtos {
java.lang.String[] descriptorData = {
"\npstorage-hbase/src/main/java/org/apache" +
"/kylin/storage/hbase/cube/v2/coprocessor" +
- "/endpoint/protobuf/CubeVisit.proto\"\237\001\n\020C" +
+ "/endpoint/protobuf/CubeVisit.proto\"\273\001\n\020C" +
"ubeVisitRequest\022\020\n\010behavior\030\001 \002(\t\022\025\n\rgtS" +
- "canRequest\030\002 \002(\014\022\024\n\014hbaseRawScan\030\003 \002(\014\0223" +
- "\n\020hbaseColumnsToGT\030\004 \003(\0132\031.CubeVisitRequ" +
- "est.IntList\032\027\n\007IntList\022\014\n\004ints\030\001 \003(\005\"\304\001\n" +
- "\021CubeVisitResponse\022\026\n\016compressedRows\030\001 \002" +
- "(\014\022\'\n\005stats\030\002 \002(\0132\030.CubeVisitResponse.St" +
- "ats\032n\n\005Stats\022\030\n\020serviceStartTime\030\001 \001(\003\022\026",
- "\n\016serviceEndTime\030\002 \001(\003\022\027\n\017scannedRowCoun" +
- "t\030\003 \001(\005\022\032\n\022aggregatedRowCount\030\004 \001(\0052F\n\020C" +
- "ubeVisitService\0222\n\tvisitCube\022\021.CubeVisit" +
- "Request\032\022.CubeVisitResponseB`\nEorg.apach" +
- "e.kylin.storage.hbase.cube.v2.coprocesso" +
- "r.endpoint.generatedB\017CubeVisitProtosH\001\210" +
- "\001\001\240\001\001"
+ "canRequest\030\002 \002(\014\022\024\n\014hbaseRawScan\030\003 \002(\014\022\032" +
+ "\n\022rowkeyPreambleSize\030\004 \002(\005\0223\n\020hbaseColum" +
+ "nsToGT\030\005 \003(\0132\031.CubeVisitRequest.IntList\032" +
+ "\027\n\007IntList\022\014\n\004ints\030\001 \003(\005\"\251\002\n\021CubeVisitRe" +
+ "sponse\022\026\n\016compressedRows\030\001 \002(\014\022\'\n\005stats\030" +
+ "\002 \002(\0132\030.CubeVisitResponse.Stats\032\322\001\n\005Stat",
+ "s\022\030\n\020serviceStartTime\030\001 \001(\003\022\026\n\016serviceEn" +
+ "dTime\030\002 \001(\003\022\027\n\017scannedRowCount\030\003 \001(\005\022\032\n\022" +
+ "aggregatedRowCount\030\004 \001(\005\022\025\n\rsystemCpuLoa" +
+ "d\030\005 \001(\001\022\036\n\026freePhysicalMemorySize\030\006 \001(\001\022" +
+ "\031\n\021freeSwapSpaceSize\030\007 \001(\001\022\020\n\010hostname\030\010" +
+ " \001(\t2F\n\020CubeVisitService\0222\n\tvisitCube\022\021." +
+ "CubeVisitRequest\032\022.CubeVisitResponseB`\nE" +
+ "org.apache.kylin.storage.hbase.cube.v2.c" +
+ "oprocessor.endpoint.generatedB\017CubeVisit" +
+ "ProtosH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -3246,7 +3782,7 @@ public final class CubeVisitProtos {
internal_static_CubeVisitRequest_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_CubeVisitRequest_descriptor,
- new java.lang.String[] { "Behavior", "GtScanRequest", "HbaseRawScan", "HbaseColumnsToGT", });
+ new java.lang.String[] { "Behavior", "GtScanRequest", "HbaseRawScan", "RowkeyPreambleSize", "HbaseColumnsToGT", });
internal_static_CubeVisitRequest_IntList_descriptor =
internal_static_CubeVisitRequest_descriptor.getNestedTypes().get(0);
internal_static_CubeVisitRequest_IntList_fieldAccessorTable = new
@@ -3264,7 +3800,7 @@ public final class CubeVisitProtos {
internal_static_CubeVisitResponse_Stats_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_CubeVisitResponse_Stats_descriptor,
- new java.lang.String[] { "ServiceStartTime", "ServiceEndTime", "ScannedRowCount", "AggregatedRowCount", });
+ new java.lang.String[] { "ServiceStartTime", "ServiceEndTime", "ScannedRowCount", "AggregatedRowCount", "SystemCpuLoad", "FreePhysicalMemorySize", "FreeSwapSpaceSize", "Hostname", });
return null;
}
};
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/protobuf/CubeVisit.proto
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/protobuf/CubeVisit.proto b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/protobuf/CubeVisit.proto
index 4ac6414..aa05c0b 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/protobuf/CubeVisit.proto
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/protobuf/CubeVisit.proto
@@ -15,7 +15,8 @@ message CubeVisitRequest {
required string behavior = 1;
required bytes gtScanRequest = 2;
required bytes hbaseRawScan = 3;
- repeated IntList hbaseColumnsToGT = 4;
+ required int32 rowkeyPreambleSize = 4;
+ repeated IntList hbaseColumnsToGT = 5;
message IntList {
repeated int32 ints = 1;
}
@@ -27,6 +28,10 @@ message CubeVisitResponse {
optional int64 serviceEndTime = 2;
optional int32 scannedRowCount = 3;
optional int32 aggregatedRowCount = 4;
+ optional double systemCpuLoad = 5;
+ optional double freePhysicalMemorySize = 6;
+ optional double freeSwapSpaceSize = 7;
+ optional string hostname = 8;
}
required bytes compressedRows = 1;
required Stats stats = 2;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
index b29bd81..8bff4d1 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
@@ -55,7 +55,6 @@ import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
@@ -84,7 +83,6 @@ public class CreateHTableJob extends AbstractHadoopJob {
CubeDesc cubeDesc = null;
String segmentName = null;
KylinConfig kylinConfig;
- public static final boolean ENABLE_CUBOID_SHARDING = true;
@Override
public int run(String[] args) throws Exception {
@@ -179,7 +177,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
IOUtils.closeStream(tempFileStream);
}
Map<Long, HyperLogLogPlusCounter> counterMap = Maps.newHashMap();
-
+
FileSystem fs = HadoopUtil.getFileSystem("file:///" + tempFile.getAbsolutePath());
int samplingPercentage = 25;
SequenceFile.Reader reader = null;
@@ -257,7 +255,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
Map<Long, Double> cubeSizeMap = Maps.newHashMap();
for (Map.Entry<Long, Long> entry : cubeRowCountMap.entrySet()) {
- cubeSizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeDesc, entry.getKey(), entry.getValue(), baseCuboidId, rowkeyColumnSize));
+ cubeSizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeSegment, entry.getKey(), entry.getValue(), baseCuboidId, rowkeyColumnSize));
}
for (Double cuboidSize : cubeSizeMap.values()) {
@@ -268,7 +266,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
nRegion = Math.max(kylinConfig.getHBaseRegionCountMin(), nRegion);
nRegion = Math.min(kylinConfig.getHBaseRegionCountMax(), nRegion);
- if (ENABLE_CUBOID_SHARDING) {//&& (nRegion > 1)) {
+ if (cubeSegment.isEnableSharding()) {//&& (nRegion > 1)) {
//use prime nRegions to help random sharding
int original = nRegion;
nRegion = Primes.nextPrime(nRegion);//return 2 for input 1
@@ -290,7 +288,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
logger.info("Expecting " + nRegion + " regions.");
logger.info("Expecting " + mbPerRegion + " MB per region.");
- if (ENABLE_CUBOID_SHARDING) {
+ if (cubeSegment.isEnableSharding()) {
//each cuboid will be split into different number of shards
HashMap<Long, Short> cuboidShards = Maps.newHashMap();
double[] regionSizes = new double[nRegion];
@@ -357,14 +355,11 @@ public class CreateHTableJob extends AbstractHadoopJob {
/**
* Estimate the cuboid's size
*
- * @param cubeDesc
- * @param cuboidId
- * @param rowCount
* @return the cuboid size in M bytes
*/
- private static double estimateCuboidStorageSize(CubeDesc cubeDesc, long cuboidId, long rowCount, long baseCuboidId, List<Integer> rowKeyColumnLength) {
+ private static double estimateCuboidStorageSize(CubeSegment cubeSegment, long cuboidId, long rowCount, long baseCuboidId, List<Integer> rowKeyColumnLength) {
- int bytesLength = RowConstants.ROWKEY_HEADER_LEN;
+ int bytesLength = cubeSegment.getRowKeyPreambleSize();
long mask = Long.highestOneBit(baseCuboidId);
long parentCuboidIdActualLength = Long.SIZE - Long.numberOfLeadingZeros(baseCuboidId);
@@ -377,7 +372,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
// add the measure length
int space = 0;
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
+ for (MeasureDesc measureDesc : cubeSegment.getCubeDesc().getMeasures()) {
DataType returnType = measureDesc.getFunction().getReturnDataType();
if (returnType.isHLLC()) {
// for HLL, it will be compressed when export to bytes
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHTableUtil.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHTableUtil.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHTableUtil.java
index 221be8a..cdc259b 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHTableUtil.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHTableUtil.java
@@ -11,6 +11,7 @@ import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy;
+import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
import org.apache.hadoop.hbase.security.User;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.model.CubeDesc;
@@ -33,7 +34,7 @@ public class CubeHTableUtil {
HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
// https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/ConstantSizeRegionSplitPolicy.html
- tableDesc.setValue(HTableDescriptor.SPLIT_POLICY, ConstantSizeRegionSplitPolicy.class.getName());
+ tableDesc.setValue(HTableDescriptor.SPLIT_POLICY, DisabledRegionSplitPolicy.class.getName());
tableDesc.setValue(IRealizationConstants.HTableTag, kylinConfig.getMetadataUrlPrefix());
tableDesc.setValue(IRealizationConstants.HTableCreationTime, String.valueOf(System.currentTimeMillis()));
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/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 d857fb1..31cce7b 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
@@ -34,22 +34,16 @@
package org.apache.kylin.storage.hbase.steps;
import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.BitSet;
import java.util.List;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
-import org.apache.kylin.common.util.ByteArray;
-import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.ImmutableBitSet;
-import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.inmemcubing.ICuboidWriter;
-import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.HBaseColumnDesc;
import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
@@ -70,11 +64,13 @@ public final class HBaseCuboidWriter implements ICuboidWriter {
private final List<KeyValueCreator> keyValueCreators;
private final int nColumns;
private final HTableInterface hTable;
- private final ByteBuffer byteBuffer;
private final CubeDesc cubeDesc;
private final CubeSegment cubeSegment;
private final Object[] measureValues;
+
private List<Put> puts = Lists.newArrayList();
+ private AbstractRowKeyEncoder rowKeyEncoder;
+ private byte[] keybuf;
public HBaseCuboidWriter(CubeSegment segment, HTableInterface hTable) {
this.keyValueCreators = Lists.newArrayList();
@@ -87,7 +83,6 @@ public final class HBaseCuboidWriter implements ICuboidWriter {
}
this.nColumns = keyValueCreators.size();
this.hTable = hTable;
- this.byteBuffer = ByteBuffer.allocate(RowConstants.ROWKEY_BUFFER_SIZE);
this.measureValues = new Object[cubeDesc.getMeasures().size()];
}
@@ -97,38 +92,28 @@ public final class HBaseCuboidWriter implements ICuboidWriter {
return result;
}
- private ByteBuffer createKey(Long cuboidId, GTRecord record) {
- byteBuffer.clear();
- byteBuffer.put(Bytes.toBytes((short) 0), 0, RowConstants.ROWKEY_SHARDID_LEN);//occupy space first
- byteBuffer.put(Bytes.toBytes(cuboidId), 0, RowConstants.ROWKEY_CUBOIDID_LEN);
- final int cardinality = BitSet.valueOf(new long[] { cuboidId }).cardinality();
- for (int i = 0; i < cardinality; i++) {
- final ByteArray byteArray = record.get(i);
- byteBuffer.put(byteArray.array(), byteArray.offset(), byteArray.length());
+ //TODO:shardingonstreaming
+ private byte[] createKey(Long cuboidId, GTRecord record) {
+ if (rowKeyEncoder == null || rowKeyEncoder.getCuboidID() != cuboidId) {
+ rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeDesc, cuboidId));
+ keybuf = rowKeyEncoder.createBuf();
}
+ rowKeyEncoder.encode(record, record.getInfo().getPrimaryKey(), keybuf);
+ return keybuf;
- //fill shard
- short cuboidShardNum = cubeSegment.getCuboidShardNum(cuboidId);
- short shardOffset = ShardingHash.getShard(byteBuffer.array(), //
- RowConstants.ROWKEY_HEADER_LEN, byteBuffer.position() - RowConstants.ROWKEY_HEADER_LEN, cuboidShardNum);
- Short cuboidShardBase = cubeSegment.getCuboidBaseShard(cuboidId);
- short finalShard = ShardingHash.normalize(cuboidShardBase, shardOffset, cubeSegment.getTotalShards());
- BytesUtil.writeShort(finalShard, byteBuffer.array(), 0, RowConstants.ROWKEY_SHARDID_LEN);
-
- return byteBuffer;
}
@Override
public void write(long cuboidId, GTRecord record) throws IOException {
- final ByteBuffer key = createKey(cuboidId, record);
+ byte[] key = createKey(cuboidId, record);
final Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidId);
final int nDims = cuboid.getColumns().size();
final ImmutableBitSet bitSet = new ImmutableBitSet(nDims, nDims + cubeDesc.getMeasures().size());
for (int i = 0; i < nColumns; i++) {
final Object[] values = record.getValues(bitSet, measureValues);
- final KeyValue keyValue = keyValueCreators.get(i).create(key.array(), 0, key.position(), values);
- final Put put = new Put(copy(key.array(), 0, key.position()));
+ final KeyValue keyValue = keyValueCreators.get(i).create(key, 0, key.length, values);
+ final Put put = new Put(copy(key, 0, key.length));
byte[] family = copy(keyValue.getFamilyArray(), keyValue.getFamilyOffset(), keyValue.getFamilyLength());
byte[] qualifier = copy(keyValue.getQualifierArray(), keyValue.getQualifierOffset(), keyValue.getQualifierLength());
byte[] value = copy(keyValue.getValueArray(), keyValue.getValueOffset(), keyValue.getValueLength());
@@ -158,7 +143,7 @@ public final class HBaseCuboidWriter implements ICuboidWriter {
@Override
public void close() {
-
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseStreamingOutput.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseStreamingOutput.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseStreamingOutput.java
index 7b4e1a4..896bc72 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseStreamingOutput.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseStreamingOutput.java
@@ -51,6 +51,7 @@ public class HBaseStreamingOutput implements IStreamingOutput {
public ICuboidWriter getCuboidWriter(IBuildable buildable) {
try {
CubeSegment cubeSegment = (CubeSegment) buildable;
+
final HTableInterface hTable;
hTable = createHTable(cubeSegment);
return new HBaseCuboidWriter(cubeSegment, hTable);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
index a88e147..a5aba2c 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
@@ -206,7 +206,7 @@ public class AggregateRegionObserverTest {
t.setDatabase("DEFAULT");
TblColRef[] cols = new TblColRef[] { newCol(1, "A", t), newCol(2, "B", t), newCol(3, "C", t), newCol(4, "D", t) };
int[] sizes = new int[] { 1, 1, 1, 1 };
- return new CoprocessorRowType(cols, sizes);
+ return new CoprocessorRowType(cols, sizes,0);
}
private TblColRef newCol(int i, String name, TableDesc t) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJobTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJobTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJobTest.java
index 05d0b08..eebd9cd 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJobTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJobTest.java
@@ -32,10 +32,6 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-/**
- * @author ysong1
- *
- */
public class RangeKeyDistributionJobTest extends LocalFileMetadataTestCase {
private Configuration conf;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/webapp/app/js/model/cubeDescModel.js
----------------------------------------------------------------------
diff --git a/webapp/app/js/model/cubeDescModel.js b/webapp/app/js/model/cubeDescModel.js
index 3376bfc..c9dfe56 100644
--- a/webapp/app/js/model/cubeDescModel.js
+++ b/webapp/app/js/model/cubeDescModel.js
@@ -51,7 +51,8 @@ KylinApp.service('CubeDescModel', function () {
},
"retention_range": "0",
"auto_merge_time_ranges": [604800000, 2419200000],
- "engine_type": 2
+ "engine_type": 2,
+ "storage_type":2
};
return cubeMeta;
[5/5] incubator-kylin git commit: KYLIN-1137 TopN measure need
support dictionary merge
Posted by ma...@apache.org.
KYLIN-1137 TopN measure need support dictionary merge
KYLIN-1137 column name need be upper case in cube desc
KYLIN-1137 fix bug in MergeCuboidMapper
KYLIN-1137 add topN support in MergeCuboidFromStorageMapper for MR_V2
fix wrong fast mode message
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/134960c6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/134960c6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/134960c6
Branch: refs/heads/2.x-staging
Commit: 134960c62424c8f4bd9e4841e925f209caf3d8be
Parents: d256a7f
Author: shaofengshi <sh...@apache.org>
Authored: Wed Nov 11 19:58:02 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Mon Nov 16 10:28:25 2015 +0800
----------------------------------------------------------------------
.../kylin/job/BuildCubeWithEngineTest.java | 10 +--
.../org/apache/kylin/common/topn/Counter.java | 2 +-
.../org/apache/kylin/cube/model/CubeDesc.java | 10 +++
.../engine/mr/steps/InMemCuboidMapper.java | 26 ++----
.../mr/steps/MergeCuboidFromStorageMapper.java | 69 ++++++++++++++
.../engine/mr/steps/MergeCuboidMapper.java | 95 ++++++++++++++++++--
.../engine/mr/steps/MergeDictionaryStep.java | 17 ++--
.../cube_desc/test_kylin_cube_topn_desc.json | 2 +-
.../test_kylin_cube_topn_left_join_desc.json | 2 +-
.../test_kylin_cube_without_slr_desc.json | 2 +-
...t_kylin_cube_without_slr_left_join_desc.json | 2 +-
11 files changed, 193 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
index d2a101d..75a6c54 100644
--- a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
@@ -62,7 +62,7 @@ public class BuildCubeWithEngineTest {
private CubeManager cubeManager;
private DefaultScheduler scheduler;
protected ExecutableManager jobService;
- private static boolean fastBuildMode = true;
+ private static boolean fastBuildMode = false;
private static final Log logger = LogFactory.getLog(BuildCubeWithEngineTest.class);
@@ -87,11 +87,11 @@ public class BuildCubeWithEngineTest {
ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
String fastModeStr = System.getProperty("fastBuildMode");
- if (fastModeStr != null && fastModeStr.equalsIgnoreCase("false")) {
- fastBuildMode = false;
- logger.info("Will not use fast build mode");
- } else {
+ if (fastModeStr != null && fastModeStr.equalsIgnoreCase("true")) {
+ fastBuildMode = true;
logger.info("Will use fast build mode");
+ } else {
+ logger.info("Will not use fast build mode");
}
System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/core-common/src/main/java/org/apache/kylin/common/topn/Counter.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/topn/Counter.java b/core-common/src/main/java/org/apache/kylin/common/topn/Counter.java
index 2bca4df..31c5ed1 100644
--- a/core-common/src/main/java/org/apache/kylin/common/topn/Counter.java
+++ b/core-common/src/main/java/org/apache/kylin/common/topn/Counter.java
@@ -50,7 +50,7 @@ public class Counter<T> implements Externalizable {
public T getItem() {
return item;
}
-
+
public double getCount() {
return count;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index 95eaf6d..2250945 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -854,4 +854,14 @@ public class CubeDesc extends RootPersistentEntity {
public LinkedHashSet<TblColRef> getMeasureDisplayColumns() {
return measureDisplayColumns;
}
+
+
+ public boolean hasMeasureUsingDictionary() {
+ for (MeasureDesc measureDesc : this.getMeasures()) {
+ if (measureDesc.getFunction().isTopN())
+ return true;
+ }
+
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/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 2bf627b..d724c76 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
@@ -65,26 +65,14 @@ public class InMemCuboidMapper<KEYIN> extends KylinMapper<KEYIN, Object, ByteArr
Map<TblColRef, Dictionary<?>> dictionaryMap = Maps.newHashMap();
- for (DimensionDesc dim : cubeDesc.getDimensions()) {
- // dictionary
- for (TblColRef col : dim.getColumnRefs()) {
- if (cubeDesc.getRowkey().isUseDictionary(col)) {
- Dictionary<?> dict = cubeSegment.getDictionary(col);
- if (dict == null) {
- logger.warn("Dictionary for " + col + " was not found.");
- }
-
- dictionaryMap.put(col, cubeSegment.getDictionary(col));
- }
- }
- }
-
- for (MeasureDesc measureDesc : cubeDesc.getMeasures()) {
- if (measureDesc.getFunction().isTopN()) {
- List<TblColRef> colRefs = measureDesc.getFunction().getParameter().getColRefs();
- TblColRef col = colRefs.get(colRefs.size() - 1);
- dictionaryMap.put(col, cubeSegment.getDictionary(col));
+ // dictionary
+ for (TblColRef col : cubeDesc.getAllColumnsNeedDictionary()) {
+ Dictionary<?> dict = cubeSegment.getDictionary(col);
+ if (dict == null) {
+ logger.warn("Dictionary for " + col + " was not found.");
}
+
+ dictionaryMap.put(col, cubeSegment.getDictionary(col));
}
DoggedCubeBuilder cubeBuilder = new DoggedCubeBuilder(cube.getDescriptor(), dictionaryMap);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
index 50f3d4c..286ff02 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
@@ -21,8 +21,14 @@ package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
+import java.util.List;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.topn.Counter;
+import org.apache.kylin.common.topn.TopNCounter;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.Pair;
@@ -45,6 +51,8 @@ import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.metadata.measure.MeasureCodec;
+import org.apache.kylin.metadata.model.ColumnDesc;
+import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.TblColRef;
import org.slf4j.Logger;
@@ -81,6 +89,9 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
private MeasureCodec codec;
private ByteArrayWritable outputValue = new ByteArrayWritable();
+ private List<MeasureDesc> measuresDescs;
+ private Integer[] measureIdxUsingDict;
+
private Boolean checkNeedMerging(TblColRef col) throws IOException {
Boolean ret = dictsNeedMerging.get(col);
if (ret != null)
@@ -119,7 +130,18 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
+ measuresDescs = cubeDesc.getMeasures();
codec = new MeasureCodec(cubeDesc.getMeasures());
+ if (cubeDesc.hasMeasureUsingDictionary()) {
+ List<Integer> measuresUsingDict = Lists.newArrayList();
+ for (int i = 0; i < measuresDescs.size(); i++) {
+ if (measuresDescs.get(i).getFunction().isTopN()) {
+ // so far only TopN uses dic
+ measuresUsingDict.add(i);
+ }
+ }
+ measureIdxUsingDict = measuresUsingDict.toArray(new Integer[measuresUsingDict.size()]);
+ }
}
@Override
@@ -191,6 +213,11 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, bufOffset), newKeyBuf);
outputKey.set(newKeyBuf.array(), 0, fullKeySize);
+ // encode measure if it uses dictionary
+ if (cubeDesc.hasMeasureUsingDictionary()) {
+ reEncodeMeasure(value);
+ }
+
valueBuf.clear();
codec.encode(value, valueBuf);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
@@ -198,4 +225,46 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
context.write(outputKey, outputValue);
}
+ private void reEncodeMeasure(Object[] measureObjs) throws IOException, InterruptedException {
+ int bufOffset = 0;
+ for (int idx : measureIdxUsingDict) {
+ // only TopN measure uses dic
+ TopNCounter<ByteArray> topNCounters = (TopNCounter<ByteArray>) measureObjs[idx];
+
+ MeasureDesc measureDesc = measuresDescs.get(idx);
+ String displayCol = measureDesc.getFunction().getParameter().getDisplayColumn().toUpperCase();
+ if (StringUtils.isNotEmpty(displayCol)) {
+ ColumnDesc sourceColumn = cubeDesc.getFactTableDesc().findColumnByName(displayCol);
+ TblColRef colRef = new TblColRef(sourceColumn);
+ DictionaryManager dictMgr = DictionaryManager.getInstance(config);
+ Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(colRef));
+ Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(colRef));
+
+ int topNSize = topNCounters.size();
+ while (sourceDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfId() * topNSize > newKeyBodyBuf.length - bufOffset) {
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
+ }
+
+ for (Counter<ByteArray> c : topNCounters) {
+ int idInSourceDict = BytesUtil.readUnsigned(c.getItem().array(), c.getItem().offset(), c.getItem().length());
+ int idInMergedDict;
+ int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
+ if (size < 0) {
+ idInMergedDict = mergedDict.nullId();
+ } else {
+ idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
+ }
+
+ BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
+ c.getItem().set(newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
+ bufOffset += mergedDict.getSizeOfId();
+ }
+ }
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/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 0b68e59..68d1481 100644
--- 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
@@ -18,14 +18,13 @@
package org.apache.kylin.engine.mr.steps;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.topn.Counter;
+import org.apache.kylin.common.topn.TopNCounter;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.SplittedBytes;
@@ -43,9 +42,19 @@ import org.apache.kylin.dict.DictionaryManager;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.metadata.measure.MeasureCodec;
+import org.apache.kylin.metadata.model.ColumnDesc;
+import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.TblColRef;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
/**
* @author ysong1, honma
*/
@@ -68,6 +77,12 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
private RowKeyEncoderProvider rowKeyEncoderProvider;
private HashMap<TblColRef, Boolean> dictsNeedMerging = new HashMap<TblColRef, Boolean>();
+ private List<MeasureDesc> measuresDescs;
+ private MeasureCodec codec;
+ private Object[] measureObjs;
+ private Integer[] measureIdxUsingDict;
+ private ByteBuffer valueBuf;
+ private Text outputValue;
private Boolean checkNeedMerging(TblColRef col) throws IOException {
Boolean ret = dictsNeedMerging.get(col);
@@ -108,6 +123,22 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
+
+ if (cubeDesc.hasMeasureUsingDictionary()) {
+ measuresDescs = cubeDesc.getMeasures();
+ codec = new MeasureCodec(measuresDescs);
+ measureObjs = new Object[measuresDescs.size()];
+ List<Integer> measuresUsingDict = Lists.newArrayList();
+ for (int i = 0; i < measuresDescs.size(); i++) {
+ if (measuresDescs.get(i).getFunction().isTopN()) {
+ // so far only TopN uses dic
+ measuresUsingDict.add(i);
+ }
+ }
+ measureIdxUsingDict = measuresUsingDict.toArray(new Integer[measuresUsingDict.size()]);
+ valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ outputValue = new Text();
+ }
}
private static final Pattern JOB_NAME_PATTERN = Pattern.compile("kylin-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})");
@@ -200,6 +231,60 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, bufOffset), newKeyBuf);
outputKey.set(newKeyBuf.array(), 0, fullKeySize);
+ // encode measure if it uses dictionary
+ if (cubeDesc.hasMeasureUsingDictionary()) {
+ codec.decode(ByteBuffer.wrap(value.getBytes(), 0, value.getLength()), measureObjs);
+ reEncodeMeasure(measureObjs);
+ valueBuf.clear();
+ codec.encode(measureObjs, valueBuf);
+ outputValue.set(valueBuf.array(), 0, valueBuf.position());
+ value = outputValue;
+ }
+
context.write(outputKey, value);
}
+
+ private void reEncodeMeasure(Object[] measureObjs) throws IOException, InterruptedException {
+ int bufOffset = 0;
+ for (int idx : measureIdxUsingDict) {
+ // only TopN measure uses dic
+ TopNCounter<ByteArray> topNCounters = (TopNCounter<ByteArray>) measureObjs[idx];
+
+ MeasureDesc measureDesc = measuresDescs.get(idx);
+ String displayCol = measureDesc.getFunction().getParameter().getDisplayColumn().toUpperCase();
+ if (StringUtils.isNotEmpty(displayCol)) {
+ ColumnDesc sourceColumn = cubeDesc.getFactTableDesc().findColumnByName(displayCol);
+ TblColRef colRef = new TblColRef(sourceColumn);
+ DictionaryManager dictMgr = DictionaryManager.getInstance(config);
+ Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(colRef));
+ Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(colRef));
+
+ int topNSize = topNCounters.size();
+ while (sourceDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfId() * topNSize > newKeyBodyBuf.length - bufOffset) {
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
+ }
+
+ for (Counter<ByteArray> c : topNCounters) {
+ int idInSourceDict = BytesUtil.readUnsigned(c.getItem().array(), c.getItem().offset(), c.getItem().length());
+ int idInMergedDict;
+ int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
+ if (size < 0) {
+ idInMergedDict = mergedDict.nullId();
+ } else {
+ idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
+ }
+
+ BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
+ c.getItem().set(newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
+ bufOffset += mergedDict.getSizeOfId();
+ }
+ }
+ }
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeDictionaryStep.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeDictionaryStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeDictionaryStep.java
index 2db4ce7..b73fda4 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeDictionaryStep.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeDictionaryStep.java
@@ -110,16 +110,13 @@ public class MergeDictionaryStep extends AbstractExecutable {
DictionaryManager dictMgr = DictionaryManager.getInstance(conf);
CubeDesc cubeDesc = cube.getDescriptor();
- for (DimensionDesc dim : cubeDesc.getDimensions()) {
- for (TblColRef col : dim.getColumnRefs()) {
- if (newSeg.getCubeDesc().getRowkey().isUseDictionary(col)) {
- String dictTable = dictMgr.decideSourceData(cubeDesc.getModel(), cubeDesc.getRowkey().getDictionary(col), col).getTable();
- if (cubeDesc.getFactTable().equalsIgnoreCase(dictTable)) {
- colsNeedMeringDict.add(col);
- } else {
- colsNeedCopyDict.add(col);
- }
- }
+
+ for (TblColRef col : cubeDesc.getAllColumnsNeedDictionary()) {
+ String dictTable = dictMgr.decideSourceData(cubeDesc.getModel(), "true", col).getTable();
+ if (cubeDesc.getFactTable().equalsIgnoreCase(dictTable)) {
+ colsNeedMeringDict.add(col);
+ } else {
+ colsNeedCopyDict.add(col);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
index 84cdaf4..fddbb10 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
@@ -88,7 +88,7 @@
"parameter": {
"type": "column",
"value": "PRICE",
- "displaycolumn": "seller_id"
+ "displaycolumn": "SELLER_ID"
},
"returntype": "topn(100)"
},
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
index f7e700d..6aecaae 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
@@ -88,7 +88,7 @@
"parameter": {
"type": "column",
"value": "PRICE",
- "displaycolumn": "seller_id"
+ "displaycolumn": "SELLER_ID"
},
"returntype": "topn(100)"
},
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
index bd979e0..3f9957b 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
@@ -170,7 +170,7 @@
"parameter": {
"type": "column",
"value": "PRICE",
- "displaycolumn": "seller_id"
+ "displaycolumn": "SELLER_ID"
},
"returntype": "topn(100)"
},
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/134960c6/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
index dfa62f7..5835a41 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
@@ -170,7 +170,7 @@
"parameter": {
"type": "column",
"value": "PRICE",
- "displaycolumn": "seller_id"
+ "displaycolumn": "SELLER_ID"
},
"returntype": "topn(100)"
},
[3/5] incubator-kylin git commit: KYLIN-1126 pscan backward
compability with v1 storage
Posted by ma...@apache.org.
KYLIN-1126 pscan backward compability with v1 storage
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/fce575bc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/fce575bc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/fce575bc
Branch: refs/heads/2.x-staging
Commit: fce575bc78abc0426e65b67882fe1cba94ac7a15
Parents: ae0f1a7
Author: honma <ho...@ebay.com>
Authored: Wed Nov 4 16:51:02 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Mon Nov 16 10:27:54 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/common/util/ByteArray.java | 5 +
.../org/apache/kylin/common/util/BasicTest.java | 53 +-
.../java/org/apache/kylin/cube/CubeSegment.java | 21 +-
.../kylin/cube/common/RowKeySplitter.java | 32 +-
.../org/apache/kylin/cube/cuboid/Cuboid.java | 18 +-
.../kylin/cube/kv/AbstractRowKeyEncoder.java | 40 +-
.../apache/kylin/cube/kv/FuzzyKeyEncoder.java | 17 +-
.../apache/kylin/cube/kv/FuzzyMaskEncoder.java | 41 +-
.../apache/kylin/cube/kv/LazyRowKeyEncoder.java | 67 ++
.../org/apache/kylin/cube/kv/RowConstants.java | 6 +-
.../org/apache/kylin/cube/kv/RowKeyDecoder.java | 2 +-
.../org/apache/kylin/cube/kv/RowKeyEncoder.java | 100 ++-
.../kylin/cube/kv/RowKeyEncoderProvider.java | 46 ++
.../org/apache/kylin/cube/model/CubeDesc.java | 12 +-
.../org/apache/kylin/gridtable/GTRecord.java | 20 +
.../kylin/gridtable/GTScanRangePlanner.java | 2 +-
.../kylin/cube/common/RowKeySplitterTest.java | 6 +-
.../apache/kylin/cube/kv/RowKeyDecoderTest.java | 6 +-
.../apache/kylin/cube/kv/RowKeyEncoderTest.java | 30 +-
.../kylin/metadata/model/IStorageAware.java | 1 +
.../apache/kylin/storage/StorageFactory.java | 2 +
.../kylin/storage/translate/HBaseKeyRange.java | 8 +-
.../kylin/engine/mr/BatchCubingJobBuilder.java | 12 +-
.../kylin/engine/mr/BatchMergeJobBuilder.java | 9 +-
.../mr/steps/MapContextGTRecordWriter.java | 35 +-
.../mr/steps/MergeCuboidFromStorageMapper.java | 64 +-
.../engine/mr/steps/MergeCuboidMapper.java | 61 +-
.../kylin/engine/mr/steps/NDCuboidMapper.java | 43 +-
.../engine/mr/steps/MergeCuboidJobTest.java | 2 +
.../engine/mr/steps/NDCuboidMapperTest.java | 6 +-
.../spark/cube/DefaultTupleConverter.java | 30 +-
.../cube_desc/kylin_sales_cube_desc.json | 361 +++++-----
.../cube_desc/test_kylin_cube_topn_desc.json | 5 +-
.../test_kylin_cube_topn_left_join_desc.json | 8 +-
.../test_kylin_cube_with_slr_desc.json | 5 +-
...test_kylin_cube_with_slr_left_join_desc.json | 5 +-
.../test_kylin_cube_without_slr_desc.json | 5 +-
...t_kylin_cube_without_slr_left_join_desc.json | 5 +-
.../test_streaming_table_cube_desc.json | 5 +-
.../coprocessor/CoprocessorProjector.java | 5 +-
.../common/coprocessor/CoprocessorRowType.java | 17 +-
.../hbase/cube/v1/CubeSegmentTupleIterator.java | 1 -
.../storage/hbase/cube/v1/CubeStorageQuery.java | 11 +-
.../hbase/cube/v2/CubeHBaseEndpointRPC.java | 20 +-
.../storage/hbase/cube/v2/CubeHBaseRPC.java | 100 ++-
.../storage/hbase/cube/v2/CubeHBaseScanRPC.java | 2 +-
.../hbase/cube/v2/HBaseReadonlyStore.java | 7 +-
.../kylin/storage/hbase/cube/v2/HBaseScan.java | 88 ---
.../coprocessor/endpoint/CubeVisitService.java | 18 +-
.../endpoint/generated/CubeVisitProtos.java | 662 +++++++++++++++++--
.../endpoint/protobuf/CubeVisit.proto | 7 +-
.../storage/hbase/steps/CreateHTableJob.java | 19 +-
.../storage/hbase/steps/CubeHTableUtil.java | 3 +-
.../storage/hbase/steps/HBaseCuboidWriter.java | 45 +-
.../hbase/steps/HBaseStreamingOutput.java | 1 +
.../observer/AggregateRegionObserverTest.java | 2 +-
.../steps/RangeKeyDistributionJobTest.java | 4 -
webapp/app/js/model/cubeDescModel.js | 3 +-
58 files changed, 1510 insertions(+), 701 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java b/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
index a388dda..ccd5001 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
@@ -90,6 +90,11 @@ public class ByteArray implements Comparable<ByteArray>, Serializable {
set(o.data, o.offset, o.length);
}
+ public void set(int offset, int length) {
+ this.offset = offset;
+ this.length = length;
+ }
+
public void setLength(int length) {
this.length = length;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
index c60f007..2beb2c6 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
@@ -24,6 +24,7 @@ import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
+import java.util.HashMap;
import java.util.IdentityHashMap;
import org.apache.commons.configuration.ConfigurationException;
@@ -33,13 +34,14 @@ import org.junit.Test;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.TreeMultiset;
/**
-* <p/>
-* Keep this test case to test basic java functionality
-* development concept proving use
-*/
+ * <p/>
+ * Keep this test case to test basic java functionality
+ * development concept proving use
+ */
@Ignore("convenient trial tool for dev")
@SuppressWarnings("unused")
public class BasicTest {
@@ -71,11 +73,46 @@ public class BasicTest {
Count, DimensionAsMetric, DistinctCount, Normal
}
+ public static int counter = 1;
+
+ class X {
+ byte[] mm = new byte[100];
+
+ public X() {
+ counter++;
+ }
+ }
+
@Test
- public void testxx() {
- B b= new B();
- b.foo();;
-
+ public void testxx() throws InterruptedException {
+ byte[][] data = new byte[10000000][];
+ byte[] temp = new byte[100];
+ for (int i = 0; i < 100; i++) {
+ temp[i] = (byte) i;
+ }
+ for (int i = 0; i < 10000000; i++) {
+ data[i] = new byte[100];
+ }
+
+ long wallClock = System.currentTimeMillis();
+
+ for (int i = 0; i < 10000000; i++) {
+ System.arraycopy(temp, 0, data[i], 0, 100);
+ }
+ System.out.println("Time Consumed: " + (System.currentTimeMillis() - wallClock));
+ }
+
+ @Test
+ public void testyy() throws InterruptedException {
+ long wallClock = System.currentTimeMillis();
+
+ HashMap<Integer, byte[]> map = Maps.newHashMap();
+ for (int i = 0; i < 10000000; i++) {
+ byte[] a = new byte[100];
+ map.put(i, a);
+ }
+
+ System.out.println("Time Consumed: " + (System.currentTimeMillis() - wallClock));
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
index 7d17d30..076bd14 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
@@ -26,11 +26,12 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.util.ShardingHash;
+import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.dict.Dictionary;
import org.apache.kylin.dict.IDictionaryAware;
-import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
+import org.apache.kylin.metadata.model.IBuildable;
+import org.apache.kylin.metadata.model.IStorageAware;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.TblColRef;
@@ -373,6 +374,14 @@ public class CubeSegment implements Comparable<CubeSegment>, IDictionaryAware, I
return cubeInstance.getStorageType();
}
+ public boolean isEnableSharding() {
+ return getCubeDesc().isEnableSharding();
+ }
+
+ public int getRowKeyPreambleSize() {
+ return isEnableSharding() ? RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN : RowConstants.ROWKEY_CUBOIDID_LEN;
+ }
+
/**
* get the number of shards where each cuboid will distribute
* @return
@@ -386,14 +395,6 @@ public class CubeSegment implements Comparable<CubeSegment>, IDictionaryAware, I
}
}
- // /**
- // * get the number of shards where each cuboid will distribute
- // * @return
- // */
- // public Map<Long, Short> getCuboidShards() {
- // return this.cuboidShards;
- // }
-
public void setCuboidShardNums(Map<Long, Short> newCuboidShards) {
this.cuboidShardNums = newCuboidShards;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java b/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
index 0111cee..56247bc 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
@@ -36,25 +36,27 @@ public class RowKeySplitter {
private int bufferSize;
private long lastSplittedCuboidId;
- private short lastSplittedShard;
+ private boolean enableSharding;
public SplittedBytes[] getSplitBuffers() {
return splitBuffers;
}
- public int getBufferSize() {
- return bufferSize;
+ public int getBodySplitOffset() {
+ if (enableSharding) {
+ return 2;//shard+cuboid
+ } else {
+ return 1;//cuboid
+ }
}
- public long getLastSplittedCuboidId() {
- return lastSplittedCuboidId;
+ public int getBufferSize() {
+ return bufferSize;
}
- public short getLastSplittedShard() {
- return lastSplittedShard;
- }
public RowKeySplitter(CubeSegment cubeSeg, int splitLen, int bytesLen) {
+ this.enableSharding = cubeSeg.isEnableSharding();
this.cubeDesc = cubeSeg.getCubeDesc();
this.colIO = new RowKeyColumnIO(cubeSeg);
@@ -73,11 +75,14 @@ public class RowKeySplitter {
this.bufferSize = 0;
int offset = 0;
- // extract shard
- SplittedBytes shardSplit = this.splitBuffers[this.bufferSize++];
- shardSplit.length = RowConstants.ROWKEY_SHARDID_LEN;
- System.arraycopy(bytes, offset, shardSplit.value, 0, RowConstants.ROWKEY_SHARDID_LEN);
- offset += RowConstants.ROWKEY_SHARDID_LEN;
+ if (enableSharding) {
+ // extract shard
+ SplittedBytes shardSplit = this.splitBuffers[this.bufferSize++];
+ shardSplit.length = RowConstants.ROWKEY_SHARDID_LEN;
+ System.arraycopy(bytes, offset, shardSplit.value, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ offset += RowConstants.ROWKEY_SHARDID_LEN;
+ //lastSplittedShard = Bytes.toShort(shardSplit.value, 0, shardSplit.length);
+ }
// extract cuboid id
SplittedBytes cuboidIdSplit = this.splitBuffers[this.bufferSize++];
@@ -86,7 +91,6 @@ public class RowKeySplitter {
offset += RowConstants.ROWKEY_CUBOIDID_LEN;
lastSplittedCuboidId = Bytes.toLong(cuboidIdSplit.value, 0, cuboidIdSplit.length);
- lastSplittedShard = Bytes.toShort(shardSplit.value, 0, shardSplit.length);
Cuboid cuboid = Cuboid.findById(cubeDesc, lastSplittedCuboidId);
// rowkey columns
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
index 2c8680d..d7e7d9c 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
@@ -18,13 +18,23 @@
package org.apache.kylin.cube.cuboid;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
-import org.apache.kylin.cube.model.*;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.cube.model.DimensionDesc;
+import org.apache.kylin.cube.model.HierarchyDesc;
+import org.apache.kylin.cube.model.RowKeyColDesc;
+import org.apache.kylin.cube.model.RowKeyDesc;
import org.apache.kylin.cube.model.RowKeyDesc.AggrGroupMask;
import org.apache.kylin.cube.model.RowKeyDesc.HierarchyMask;
import org.apache.kylin.metadata.model.TblColRef;
@@ -147,7 +157,7 @@ public class Cuboid implements Comparable<Cuboid> {
return cuboidID;
} else {
// no column (except mandatory), add one column
- long toAddCol = (1 << (BitSet.valueOf(new long[]{rowkey.getTailMask()}).cardinality()));
+ long toAddCol = (1 << (BitSet.valueOf(new long[] { rowkey.getTailMask() }).cardinality()));
// check if the toAddCol belongs to any hierarchy
List<HierarchyMask> hierarchyMaskList = rowkey.getHierarchyMasks();
if (hierarchyMaskList != null && hierarchyMaskList.size() > 0) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
index 231f737..4316376 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
@@ -20,9 +20,12 @@ package org.apache.kylin.cube.kv;
import java.util.Map;
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.dict.Dictionary;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.metadata.model.TblColRef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,30 +37,51 @@ import org.slf4j.LoggerFactory;
*/
public abstract class AbstractRowKeyEncoder {
+ protected static final Logger logger = LoggerFactory.getLogger(AbstractRowKeyEncoder.class);
public static final byte DEFAULT_BLANK_BYTE = Dictionary.NULL;
- protected static final Logger logger = LoggerFactory.getLogger(AbstractRowKeyEncoder.class);
+ protected byte blankByte = DEFAULT_BLANK_BYTE;
+ protected final CubeSegment cubeSeg;
+ protected Cuboid cuboid;
public static AbstractRowKeyEncoder createInstance(CubeSegment cubeSeg, Cuboid cuboid) {
return new RowKeyEncoder(cubeSeg, cuboid);
}
- protected final Cuboid cuboid;
- protected byte blankByte = DEFAULT_BLANK_BYTE;
- protected boolean encodeShard = true;
-
- protected AbstractRowKeyEncoder(Cuboid cuboid) {
+ protected AbstractRowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
this.cuboid = cuboid;
+ this.cubeSeg = cubeSeg;
}
public void setBlankByte(byte blankByte) {
this.blankByte = blankByte;
}
- public void setEncodeShard(boolean encodeShard) {
- this.encodeShard = encodeShard;
+ public long getCuboidID() {
+ return cuboid.getId();
}
+ public void setCuboid(Cuboid cuboid) {
+ this.cuboid = cuboid;
+ }
+
+ abstract public byte[] createBuf();
+
+ /**
+ * encode a gtrecord into a given byte[] buffer
+ * @param record
+ * @param keyColumns
+ * @param buf
+ */
+ abstract public void encode(GTRecord record, ImmutableBitSet keyColumns, byte[] buf);
+
+ /**
+ * when a rowkey's body is provided, help to encode cuboid & shard (if apply)
+ * @param bodyBytes
+ * @param outputBuf
+ */
+ abstract public void encode(ByteArray bodyBytes, ByteArray outputBuf);
+
abstract public byte[] encode(Map<TblColRef, String> valueMap);
abstract public byte[] encode(byte[][] values);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyKeyEncoder.java
index 2185bc5..9da8ff5 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyKeyEncoder.java
@@ -18,8 +18,6 @@
package org.apache.kylin.cube.kv;
-import java.util.Arrays;
-
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
@@ -35,9 +33,16 @@ public class FuzzyKeyEncoder extends RowKeyEncoder {
}
@Override
- protected byte[] defaultValue(int length) {
- byte[] keyBytes = new byte[length];
- Arrays.fill(keyBytes, RowConstants.BYTE_ZERO);
- return keyBytes;
+ protected short calculateShard(byte[] key) {
+ if (enableSharding) {
+ return 0;
+ } else {
+ throw new RuntimeException("If enableSharding false, you should never calculate shard");
+ }
+ }
+
+ @Override
+ protected byte defaultValue() {
+ return RowConstants.BYTE_ZERO;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyMaskEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyMaskEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyMaskEncoder.java
index bf67538..94db94b 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyMaskEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/FuzzyMaskEncoder.java
@@ -20,8 +20,12 @@ package org.apache.kylin.cube.kv;
import java.util.Arrays;
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.metadata.model.TblColRef;
/**
@@ -36,11 +40,40 @@ public class FuzzyMaskEncoder extends RowKeyEncoder {
}
@Override
- protected int fillHeader(byte[] bytes) {
- Arrays.fill(bytes, 0, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.BYTE_ONE);
+ public void encode(GTRecord record, ImmutableBitSet keyColumns, byte[] buf) {
+ ByteArray byteArray = new ByteArray(buf, getHeaderLength(), 0);
+
+ GTInfo info = record.getInfo();
+ byte fill;
+ int pos = 0;
+ for (int i = 0; i < info.getPrimaryKey().trueBitCount(); i++) {
+ int c = info.getPrimaryKey().trueBitAt(i);
+ int colLength = info.getCodeSystem().maxCodeLength(c);
+
+ if (record.get(c).array() != null) {
+ fill = RowConstants.BYTE_ZERO;
+ } else {
+ fill = RowConstants.BYTE_ONE;
+ }
+ Arrays.fill(byteArray.array(), byteArray.offset() + pos, byteArray.offset() + pos + colLength, fill);
+ pos += colLength;
+ }
+ byteArray.setLength(pos);
+
+ //fill shard and cuboid
+ fillHeader(buf);
+ }
+
+ @Override
+ protected void fillHeader(byte[] bytes) {
+ int offset = 0;
+ if (enableSharding) {
+ Arrays.fill(bytes, 0, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.BYTE_ONE);
+ offset += RowConstants.ROWKEY_SHARDID_LEN;
+ }
// always fuzzy match cuboid ID to lock on the selected cuboid
- Arrays.fill(bytes, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_HEADER_LEN, RowConstants.BYTE_ZERO);
- return this.headerLength;
+ int headerLength = this.getHeaderLength();
+ Arrays.fill(bytes, offset, headerLength, RowConstants.BYTE_ZERO);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
new file mode 100644
index 0000000..7c70fff
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
@@ -0,0 +1,67 @@
+/*
+ * 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.kv;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.ShardingHash;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+
+import com.google.common.collect.Lists;
+
+/**
+ * A LazyRowKeyEncoder will not try to calculate shard
+ * It works for both enableSharding or non-enableSharding scenario
+ * Usually it's for sharded cube scanning, later all possible shard will be rewrite
+ */
+public class LazyRowKeyEncoder extends RowKeyEncoder {
+ public LazyRowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
+ super(cubeSeg, cuboid);
+ }
+
+ protected short calculateShard(byte[] key) {
+ if (enableSharding) {
+ return 0;
+ } else {
+ throw new RuntimeException("If enableSharding false, you should never calculate shard");
+ }
+ }
+
+ //for non-sharding cases it will only return one byte[] with not shard at beginning
+ public List<byte[]> getRowKeysDifferentShards(byte[] halfCookedKey) {
+ final short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
+
+ if (!enableSharding) {
+ return Lists.newArrayList(halfCookedKey);//not shard to append at head, so it is already well cooked
+ } else {
+ List<byte[]> ret = Lists.newArrayList();
+ for (short i = 0; i < cuboidShardNum; ++i) {
+ short shard = ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), i, cubeSeg.getTotalShards());
+ byte[] cookedKey = Arrays.copyOf(halfCookedKey, halfCookedKey.length);
+ BytesUtil.writeShort(shard, cookedKey, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ ret.add(cookedKey);
+ }
+ return ret;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
index 6a8eeb5..62dea02 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
@@ -32,8 +32,8 @@ public class RowConstants {
// row key shard length
public static final int ROWKEY_SHARDID_LEN = 2;
- public static final int ROWKEY_HEADER_LEN = ROWKEY_CUBOIDID_LEN + ROWKEY_SHARDID_LEN;
-
+ public static final int ROWKEY_SHARD_AND_CUBOID_LEN = ROWKEY_CUBOIDID_LEN + ROWKEY_SHARDID_LEN;
+
public static final byte BYTE_ZERO = 0;
public static final byte BYTE_ONE = 1;
@@ -42,7 +42,7 @@ public class RowConstants {
public static final String ROWVALUE_DELIMITER_STRING = String.valueOf((char) 7);
public static final byte[] ROWVALUE_DELIMITER_BYTES = { 7 };
- public static final int ROWKEY_BUFFER_SIZE = 1024 * 1024; // 1 MB
+ public static final int ROWKEY_BUFFER_SIZE = 65 * 256;// a little more than 64 dimensions * 256 bytes each
public static final int ROWVALUE_BUFFER_SIZE = 1024 * 1024; // 1 MB
// marker class
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyDecoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyDecoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyDecoder.java
index 3506845..e4a6a52 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyDecoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyDecoder.java
@@ -58,7 +58,7 @@ public class RowKeyDecoder {
SplittedBytes[] splits = rowKeySplitter.getSplitBuffers();
- int offset = 2; // skip shard and cuboid id part
+ int offset = rowKeySplitter.getBodySplitOffset(); // skip shard and cuboid id part
for (int i = 0; i < this.cuboid.getColumns().size(); i++) {
TblColRef col = this.cuboid.getColumns().get(i);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
index 0676df6..990cf06 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
@@ -23,27 +23,49 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.metadata.model.TblColRef;
+import com.google.common.base.Preconditions;
+
public class RowKeyEncoder extends AbstractRowKeyEncoder {
- private int bytesLength;
- protected int headerLength;
+ private int bodyLength = 0;
private RowKeyColumnIO colIO;
- CubeSegment cubeSeg;
+ protected boolean enableSharding;
- protected RowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
- super(cuboid);
- this.cubeSeg = cubeSeg;
+ public RowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
+ super(cubeSeg, cuboid);
+ enableSharding = cubeSeg.isEnableSharding();
colIO = new RowKeyColumnIO(cubeSeg);
- bytesLength = headerLength = RowConstants.ROWKEY_HEADER_LEN; // include shard and cuboidid
for (TblColRef column : cuboid.getColumns()) {
- bytesLength += colIO.getColumnLength(column);
+ bodyLength += colIO.getColumnLength(column);
+ }
+ }
+
+ public int getHeaderLength() {
+ return cubeSeg.getRowKeyPreambleSize();
+ }
+
+ public int getBytesLength() {
+ return getHeaderLength() + bodyLength;
+ }
+
+ protected short calculateShard(byte[] key) {
+ if (enableSharding) {
+ int bodyOffset = RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN;
+ short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
+ short shardOffset = ShardingHash.getShard(key, bodyOffset, bodyLength, cuboidShardNum);
+ return ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), shardOffset, cubeSeg.getTotalShards());
+ } else {
+ throw new RuntimeException("If enableSharding false, you should never calculate shard");
}
}
@@ -52,6 +74,31 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
}
@Override
+ public byte[] createBuf() {
+ return new byte[this.getBytesLength()];
+ }
+
+ @Override
+ public void encode(GTRecord record, ImmutableBitSet keyColumns, byte[] buf) {
+ ByteArray byteArray = new ByteArray(buf, getHeaderLength(), 0);
+ record.exportColumns(keyColumns, byteArray, defaultValue());
+
+ //fill shard and cuboid
+ fillHeader(buf);
+ }
+
+ @Override
+ public void encode(ByteArray bodyBytes, ByteArray outputBuf) {
+ Preconditions.checkState(bodyBytes.length() == bodyLength);
+ Preconditions.checkState(bodyBytes.length() + getHeaderLength() == outputBuf.length(),//
+ "bodybytes length: " + bodyBytes.length() + " outputBuf length: " + outputBuf.length() + " header length: " + getHeaderLength());
+ System.arraycopy(bodyBytes.array(), bodyBytes.offset(), outputBuf.array(), getHeaderLength(), bodyLength);
+
+ //fill shard and cuboid
+ fillHeader(outputBuf.array());
+ }
+
+ @Override
public byte[] encode(Map<TblColRef, String> valueMap) {
List<byte[]> valueList = new ArrayList<byte[]>();
for (TblColRef bdCol : cuboid.getColumns()) {
@@ -71,9 +118,8 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
@Override
public byte[] encode(byte[][] values) {
- byte[] bytes = new byte[this.bytesLength];
- int bodyOffset = RowConstants.ROWKEY_HEADER_LEN;
- int offset = bodyOffset;
+ byte[] bytes = new byte[this.getBytesLength()];
+ int offset = getHeaderLength();
for (int i = 0; i < cuboid.getColumns().size(); i++) {
TblColRef column = cuboid.getColumns().get(i);
@@ -93,44 +139,32 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
return bytes;
}
- protected int fillHeader(byte[] bytes) {
+ protected void fillHeader(byte[] bytes) {
int offset = 0;
- if (encodeShard) {
- short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
- short shardOffset = ShardingHash.getShard(bytes, RowConstants.ROWKEY_HEADER_LEN, bytes.length - RowConstants.ROWKEY_HEADER_LEN, cuboidShardNum);
- short finalShard = ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), shardOffset, cubeSeg.getTotalShards());
- BytesUtil.writeShort(finalShard, bytes, offset, RowConstants.ROWKEY_SHARDID_LEN);
- } else {
- BytesUtil.writeShort((short) 0, bytes, offset, RowConstants.ROWKEY_SHARDID_LEN);
+ if (enableSharding) {
+ short shard = calculateShard(bytes);
+ BytesUtil.writeShort(shard, bytes, offset, RowConstants.ROWKEY_SHARDID_LEN);
+ offset += RowConstants.ROWKEY_SHARDID_LEN;
}
- offset += RowConstants.ROWKEY_SHARDID_LEN;
System.arraycopy(cuboid.getBytes(), 0, bytes, offset, RowConstants.ROWKEY_CUBOIDID_LEN);
- offset += RowConstants.ROWKEY_CUBOIDID_LEN;
-
- if (this.headerLength != offset) {
- throw new IllegalStateException("Expected header length is " + headerLength + ". But the offset is " + offset);
- }
-
- return offset;
+ //offset += RowConstants.ROWKEY_CUBOIDID_LEN;
+ //return offset;
}
protected void fillColumnValue(TblColRef column, int columnLen, byte[] value, int valueLen, byte[] outputValue, int outputValueOffset) {
// special null value case
if (value == null) {
- byte[] valueBytes = defaultValue(columnLen);
- System.arraycopy(valueBytes, 0, outputValue, outputValueOffset, columnLen);
+ Arrays.fill(outputValue, outputValueOffset, outputValueOffset + columnLen, defaultValue());
return;
}
colIO.writeColumn(column, value, valueLen, this.blankByte, outputValue, outputValueOffset);
}
- protected byte[] defaultValue(int length) {
- byte[] values = new byte[length];
- Arrays.fill(values, this.blankByte);
- return values;
+ protected byte defaultValue() {
+ return this.blankByte;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoderProvider.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoderProvider.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoderProvider.java
new file mode 100644
index 0000000..2b1dea7
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoderProvider.java
@@ -0,0 +1,46 @@
+/*
+ * 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.kv;
+
+import java.util.HashMap;
+
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+
+import com.google.common.collect.Maps;
+
+public class RowKeyEncoderProvider {
+
+ private CubeSegment cubeSegment;
+ private HashMap<Long, RowKeyEncoder> rowKeyEncoders;
+
+ public RowKeyEncoderProvider(CubeSegment cubeSegment) {
+ this.cubeSegment = cubeSegment;
+ this.rowKeyEncoders = Maps.newHashMap();
+ }
+
+ public RowKeyEncoder getRowkeyEncoder(Cuboid cuboid) {
+ RowKeyEncoder rowKeyEncoder = rowKeyEncoders.get(cuboid.getId());
+ if (rowKeyEncoder == null) {
+ rowKeyEncoder = new RowKeyEncoder(cubeSegment, cuboid);
+ rowKeyEncoders.put(cuboid.getId(), rowKeyEncoder);
+ }
+ return rowKeyEncoder;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index a4968e0..95eaf6d 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -134,6 +134,11 @@ public class CubeDesc extends RootPersistentEntity {
private Map<TblColRef, DeriveInfo> derivedToHostMap = Maps.newHashMap();
private Map<Array<TblColRef>, List<DeriveInfo>> hostToDerivedMap = Maps.newHashMap();
+ public boolean isEnableSharding() {
+ //in the future may extend to other storage that is shard-able
+ return storageType == IStorageAware.ID_SHARDED_HBASE;
+ }
+
/**
* Error messages during resolving json metadata
*/
@@ -669,7 +674,7 @@ public class CubeDesc extends RootPersistentEntity {
if (colRefs.isEmpty() == false)
p.setColRefs(colRefs);
-
+
// verify holistic count distinct as a dependent measure
if (m.getFunction().isHolisticCountDistinct() && StringUtils.isBlank(m.getDependentMeasureRef())) {
throw new IllegalStateException(m + " is a holistic count distinct but it has no DependentMeasureRef defined!");
@@ -829,17 +834,16 @@ public class CubeDesc extends RootPersistentEntity {
this.engineType = engineType;
}
-
public List<TblColRef> getAllColumnsNeedDictionary() {
List<TblColRef> result = Lists.newArrayList();
-
+
for (RowKeyColDesc rowKeyColDesc : rowkey.getRowKeyColumns()) {
TblColRef colRef = rowKeyColDesc.getColRef();
if (rowkey.isUseDictionary(colRef)) {
result.add(colRef);
}
}
-
+
for (TblColRef colRef : measureDisplayColumns) {
if (!result.contains(colRef))
result.add(colRef);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
----------------------------------------------------------------------
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 0f4eb3d..98f6e2d 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
@@ -7,6 +7,8 @@ import java.util.List;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.ImmutableBitSet;
+import com.google.common.base.Preconditions;
+
public class GTRecord implements Comparable<GTRecord> {
final GTInfo info;
@@ -222,12 +224,30 @@ public class GTRecord implements Comparable<GTRecord> {
int pos = 0;
for (int i = 0; i < selectedCols.trueBitCount(); i++) {
int c = selectedCols.trueBitAt(i);
+ Preconditions.checkNotNull(cols[c].array());
System.arraycopy(cols[c].array(), cols[c].offset(), buf.array(), buf.offset() + pos, cols[c].length());
pos += cols[c].length();
}
buf.setLength(pos);
}
+ /** write data to given buffer, like serialize, use defaultValue when required column is not set*/
+ public void exportColumns(ImmutableBitSet selectedCols, ByteArray buf, byte defaultValue) {
+ int pos = 0;
+ for (int i = 0; i < selectedCols.trueBitCount(); i++) {
+ int c = selectedCols.trueBitAt(i);
+ if (cols[c].array() != null) {
+ System.arraycopy(cols[c].array(), cols[c].offset(), buf.array(), buf.offset() + pos, cols[c].length());
+ pos += cols[c].length();
+ } else {
+ int maxLength = info.codeSystem.maxCodeLength(c);
+ Arrays.fill(buf.array(), buf.offset() + pos, buf.offset() + pos + maxLength, defaultValue);
+ pos += maxLength;
+ }
+ }
+ buf.setLength(pos);
+ }
+
/** write data to given buffer, like serialize */
public void exportColumns(ImmutableBitSet selectedCols, ByteBuffer buf) {
for (int i = 0; i < selectedCols.trueBitCount(); i++) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRangePlanner.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRangePlanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRangePlanner.java
index d860090..3d07623 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRangePlanner.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRangePlanner.java
@@ -553,7 +553,7 @@ public class GTScanRangePlanner {
/**
* asymmetric means compare(a,b) > 0 does not cause compare(b,a) < 0
- * so min max functions will not bu supported
+ * so min max functions will not be supported
*/
private static class AsymmetricRecordComparator extends RecordComparator {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/test/java/org/apache/kylin/cube/common/RowKeySplitterTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/common/RowKeySplitterTest.java b/core-cube/src/test/java/org/apache/kylin/cube/common/RowKeySplitterTest.java
index 98f1eef..bfbfb01 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/common/RowKeySplitterTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/common/RowKeySplitterTest.java
@@ -43,6 +43,7 @@ public class RowKeySplitterTest extends LocalFileMetadataTestCase {
@Test
public void testWithSlr() throws Exception {
+ //has shard
CubeInstance cube = CubeManager.getInstance(getTestConfig()).getCube("TEST_KYLIN_CUBE_WITH_SLR_READY");
RowKeySplitter rowKeySplitter = new RowKeySplitter(cube.getFirstSegment(), 11, 20);
@@ -55,13 +56,14 @@ public class RowKeySplitterTest extends LocalFileMetadataTestCase {
@Test
public void testWithoutSlr() throws Exception {
+ //no shard
CubeInstance cube = CubeManager.getInstance(getTestConfig()).getCube("TEST_KYLIN_CUBE_WITHOUT_SLR_READY");
RowKeySplitter rowKeySplitter = new RowKeySplitter(cube.getFirstSegment(), 11, 20);
// base cuboid rowkey
- byte[] input = { 0, 0, 0, 0, 0, 0, 0, 0, 0, -1, 11, 55, -13, 13, 22, 34, 121, 70, 80, 45, 71, 84, 67, 9, 9, 9, 9, 9, 9, 0, 10, 5 };
+ byte[] input = { 0, 0, 0, 0, 0, 0, 0, -1, 11, 55, -13, 13, 22, 34, 121, 70, 80, 45, 71, 84, 67, 9, 9, 9, 9, 9, 9, 0, 10, 5 };
rowKeySplitter.split(input);
- assertEquals(10, rowKeySplitter.getBufferSize());
+ assertEquals(9, rowKeySplitter.getBufferSize());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyDecoderTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyDecoderTest.java b/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyDecoderTest.java
index d6b1718..ac20c04 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyDecoderTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyDecoderTest.java
@@ -53,7 +53,7 @@ public class RowKeyDecoderTest extends LocalFileMetadataTestCase {
RowKeyDecoder rowKeyDecoder = new RowKeyDecoder(cube.getFirstSegment());
- byte[] key = { 0, 0, 0, 0, 0, 0, 0, 0, 0, -1, 11, 55, -13, 13, 22, 34, 121, 70, 80, 45, 71, 84, 67, 9, 9, 9, 9, 9, 9, 0, 10, 5 };
+ byte[] key = { 0, 0, 0, 0, 0, 0, 0, -1, 11, 55, -13, 13, 22, 34, 121, 70, 80, 45, 71, 84, 67, 9, 9, 9, 9, 9, 9, 0, 10, 5 };
rowKeyDecoder.decode(key);
List<String> values = rowKeyDecoder.getValues();
@@ -90,10 +90,10 @@ public class RowKeyDecoderTest extends LocalFileMetadataTestCase {
long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- AbstractRowKeyEncoder rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cube.getFirstSegment(), baseCuboid);
+ RowKeyEncoder rowKeyEncoder = new RowKeyEncoder(cube.getFirstSegment(), baseCuboid);
byte[] encodedKey = rowKeyEncoder.encode(data);
- assertEquals(22 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
+ assertEquals(22 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
RowKeyDecoder rowKeyDecoder = new RowKeyDecoder(cube.getFirstSegment());
rowKeyDecoder.decode(encodedKey);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyEncoderTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyEncoderTest.java b/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyEncoderTest.java
index 45c8108..b29c0e0 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyEncoderTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/kv/RowKeyEncoderTest.java
@@ -67,14 +67,12 @@ public class RowKeyEncoderTest extends LocalFileMetadataTestCase {
long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- AbstractRowKeyEncoder rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cube.getFirstSegment(), baseCuboid);
+ RowKeyEncoder rowKeyEncoder = new RowKeyEncoder(cube.getFirstSegment(), baseCuboid);
byte[] encodedKey = rowKeyEncoder.encode(data);
- assertEquals(22 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
- byte[] shard = Arrays.copyOfRange(encodedKey, 0, RowConstants.ROWKEY_SHARDID_LEN);
- byte[] cuboidId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_HEADER_LEN);
- byte[] rest = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
- assertEquals(0, Bytes.toShort(shard));
+ assertEquals(22 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
+ byte[] cuboidId = Arrays.copyOfRange(encodedKey, 0, rowKeyEncoder.getHeaderLength());
+ byte[] rest = Arrays.copyOfRange(encodedKey, rowKeyEncoder.getHeaderLength(), encodedKey.length);
assertEquals(255, Bytes.toLong(cuboidId));
assertArrayEquals(new byte[] { 11, 55, -13, 13, 22, 34, 121, 70, 80, 45, 71, 84, 67, 9, 9, 9, 9, 9, 9, 0, 10, 5 }, rest);
}
@@ -99,14 +97,14 @@ public class RowKeyEncoderTest extends LocalFileMetadataTestCase {
long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- AbstractRowKeyEncoder rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cube.getFirstSegment(), baseCuboid);
+ RowKeyEncoder rowKeyEncoder = new RowKeyEncoder(cube.getFirstSegment(), baseCuboid);
byte[] encodedKey = rowKeyEncoder.encode(data);
- assertEquals(40 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
+ assertEquals(40 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
byte[] shard = Arrays.copyOfRange(encodedKey, 0, RowConstants.ROWKEY_SHARDID_LEN);
- byte[] sellerId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_HEADER_LEN, 18 + RowConstants.ROWKEY_HEADER_LEN);
- byte[] cuboidId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_HEADER_LEN);
- byte[] rest = Arrays.copyOfRange(encodedKey, 18 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
+ byte[] sellerId = Arrays.copyOfRange(encodedKey, rowKeyEncoder.getHeaderLength(), 18 + rowKeyEncoder.getHeaderLength());
+ byte[] cuboidId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_SHARDID_LEN, rowKeyEncoder.getHeaderLength());
+ byte[] rest = Arrays.copyOfRange(encodedKey, 18 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
assertEquals(0, Bytes.toShort(shard));
assertTrue(Bytes.toString(sellerId).startsWith("123456789"));
assertEquals(511, Bytes.toLong(cuboidId));
@@ -133,14 +131,14 @@ public class RowKeyEncoderTest extends LocalFileMetadataTestCase {
long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- AbstractRowKeyEncoder rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cube.getFirstSegment(), baseCuboid);
+ RowKeyEncoder rowKeyEncoder = new RowKeyEncoder(cube.getFirstSegment(), baseCuboid);
byte[] encodedKey = rowKeyEncoder.encode(data);
- assertEquals(40 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
+ assertEquals(40 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
byte[] shard = Arrays.copyOfRange(encodedKey, 0, RowConstants.ROWKEY_SHARDID_LEN);
- byte[] cuboidId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_HEADER_LEN);
- byte[] sellerId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_HEADER_LEN, 18 + RowConstants.ROWKEY_HEADER_LEN);
- byte[] rest = Arrays.copyOfRange(encodedKey, 18 + RowConstants.ROWKEY_HEADER_LEN, encodedKey.length);
+ byte[] cuboidId = Arrays.copyOfRange(encodedKey, RowConstants.ROWKEY_SHARDID_LEN, rowKeyEncoder.getHeaderLength());
+ byte[] sellerId = Arrays.copyOfRange(encodedKey, rowKeyEncoder.getHeaderLength(), 18 + rowKeyEncoder.getHeaderLength());
+ byte[] rest = Arrays.copyOfRange(encodedKey, 18 + rowKeyEncoder.getHeaderLength(), encodedKey.length);
assertEquals(0, Bytes.toShort(shard));
assertTrue(Bytes.toString(sellerId).startsWith("123456789"));
assertEquals(511, Bytes.toLong(cuboidId));
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-metadata/src/main/java/org/apache/kylin/metadata/model/IStorageAware.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/IStorageAware.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/IStorageAware.java
index ea1aae9..e552574 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/IStorageAware.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/IStorageAware.java
@@ -22,6 +22,7 @@ public interface IStorageAware {
public static final int ID_HBASE = 0;
public static final int ID_HYBRID = 1;
+ public static final int ID_SHARDED_HBASE = 2;
int getStorageType();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/StorageFactory.java b/core-storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
index 271583c..da2f69c 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
@@ -20,6 +20,7 @@ package org.apache.kylin.storage;
import static org.apache.kylin.metadata.model.IStorageAware.ID_HBASE;
import static org.apache.kylin.metadata.model.IStorageAware.ID_HYBRID;
+import static org.apache.kylin.metadata.model.IStorageAware.ID_SHARDED_HBASE;
import java.util.HashMap;
import java.util.Map;
@@ -36,6 +37,7 @@ public class StorageFactory {
static {
Map<Integer, String> impls = new HashMap<>();
impls.put(ID_HBASE, "org.apache.kylin.storage.hbase.HBaseStorage");
+ impls.put(ID_SHARDED_HBASE, "org.apache.kylin.storage.hbase.HBaseStorage");//ID_SHARDED_HBASE is a special HBaseStorage
impls.put(ID_HYBRID, "org.apache.kylin.storage.hybrid.HybridStorage");
storages = new ImplementationSwitch<IStorage>(impls, IStorage.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/core-storage/src/main/java/org/apache/kylin/storage/translate/HBaseKeyRange.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/translate/HBaseKeyRange.java b/core-storage/src/main/java/org/apache/kylin/storage/translate/HBaseKeyRange.java
index a6d78e7..fbb258f 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/translate/HBaseKeyRange.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/translate/HBaseKeyRange.java
@@ -34,6 +34,7 @@ import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
import org.apache.kylin.cube.kv.FuzzyKeyEncoder;
import org.apache.kylin.cube.kv.FuzzyMaskEncoder;
+import org.apache.kylin.cube.kv.LazyRowKeyEncoder;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.metadata.model.TblColRef;
@@ -118,15 +119,10 @@ public class HBaseKeyRange implements Comparable<HBaseKeyRange> {
}
}
- AbstractRowKeyEncoder encoder = AbstractRowKeyEncoder.createInstance(cubeSeg, cuboid);
- encoder.setEncodeShard(false);//will enumerate all possible shards when scanning
-
+ AbstractRowKeyEncoder encoder = new LazyRowKeyEncoder(cubeSeg, cuboid);
encoder.setBlankByte(RowConstants.ROWKEY_LOWER_BYTE);
-
this.startKey = encoder.encode(startValues);
-
encoder.setBlankByte(RowConstants.ROWKEY_UPPER_BYTE);
-
// In order to make stopRow inclusive add a trailing 0 byte. #See Scan.setStopRow(byte [] stopRow)
this.stopKey = Bytes.add(encoder.encode(stopValues), ZERO_TAIL_BYTES);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
index dcb887d..b5a7272 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
@@ -29,22 +29,28 @@ import org.apache.kylin.job.constant.ExecutableConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
+
public class BatchCubingJobBuilder extends JobBuilderSupport {
-
+
private static final Logger logger = LoggerFactory.getLogger(BatchCubingJobBuilder.class);
-
+
private final IMRBatchCubingInputSide inputSide;
private final IMRBatchCubingOutputSide outputSide;
public BatchCubingJobBuilder(CubeSegment newSegment, String submitter) {
super(newSegment, submitter);
+
+ Preconditions.checkArgument(!newSegment.isEnableSharding(), "V1 job engine does not support building sharded cubes");
+
this.inputSide = MRUtil.getBatchCubingInputSide(seg);
this.outputSide = MRUtil.getBatchCubingOutputSide((CubeSegment)seg);
}
public CubingJob build() {
logger.info("MR_V1 new job to BUILD segment " + seg);
- final CubingJob result = CubingJob.createBuildJob((CubeSegment)seg, submitter, config);
+
+ final CubingJob result = CubingJob.createBuildJob(seg, submitter, config);
final String jobId = result.getId();
final String cuboidRootPath = getCuboidRootPath(jobId);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
index 4b93b5d..1282e61 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
@@ -39,13 +39,16 @@ public class BatchMergeJobBuilder extends JobBuilderSupport {
public BatchMergeJobBuilder(CubeSegment mergeSegment, String submitter) {
super(mergeSegment, submitter);
- this.outputSide = MRUtil.getBatchMergeOutputSide((CubeSegment)seg);
+
+ Preconditions.checkArgument(!mergeSegment.isEnableSharding(), "V1 job engine does not support merging sharded cubes");
+
+ this.outputSide = MRUtil.getBatchMergeOutputSide(seg);
}
public CubingJob build() {
logger.info("MR_V1 new job to MERGE segment " + seg);
- final CubeSegment cubeSegment = (CubeSegment)seg;
- final CubingJob result = CubingJob.createMergeJob(cubeSegment, submitter, config);
+
+ final CubingJob result = CubingJob.createMergeJob(seg, submitter, config);
final String jobId = result.getId();
final String cuboidRootPath = getCuboidRootPath(jobId);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MapContextGTRecordWriter.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MapContextGTRecordWriter.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MapContextGTRecordWriter.java
index 4c743fb..6098381 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MapContextGTRecordWriter.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MapContextGTRecordWriter.java
@@ -7,16 +7,14 @@ import java.util.BitSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.MapContext;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.inmemcubing.ICuboidWriter;
+import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.ByteArrayWritable;
import org.apache.kylin.gridtable.GTRecord;
-import org.apache.kylin.metadata.model.TblColRef;
/**
*/
@@ -28,7 +26,7 @@ public class MapContextGTRecordWriter implements ICuboidWriter {
protected CubeSegment cubeSegment;
protected CubeDesc cubeDesc;
- private int bytesLength;
+ private AbstractRowKeyEncoder rowKeyEncoder;
private int dimensions;
private int measureCount;
private byte[] keyBuf;
@@ -61,25 +59,13 @@ public class MapContextGTRecordWriter implements ICuboidWriter {
}
cuboidRowCount++;
- int header = RowConstants.ROWKEY_HEADER_LEN;
- int offSet = header;
- for (int x = 0; x < dimensions; x++) {
- System.arraycopy(record.get(x).array(), record.get(x).offset(), keyBuf, offSet, record.get(x).length());
- offSet += record.get(x).length();
- }
-
- //fill shard
- short cuboidShardNum = cubeSegment.getCuboidShardNum(cuboidId);
- short shardOffset = ShardingHash.getShard(keyBuf, header, offSet - header, cuboidShardNum);
- short cuboidShardBase = cubeSegment.getCuboidBaseShard(cuboidId);
- short finalShard = ShardingHash.normalize(cuboidShardBase, shardOffset, cubeSegment.getTotalShards());
- BytesUtil.writeShort(finalShard, keyBuf, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ rowKeyEncoder.encode(record, record.getInfo().getPrimaryKey(), keyBuf);
//output measures
valueBuf.clear();
record.exportColumns(measureColumnsIndex, valueBuf);
- outputKey.set(keyBuf, 0, offSet);
+ outputKey.set(keyBuf, 0, keyBuf.length);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
try {
mapContext.write(outputKey, outputValue);
@@ -95,24 +81,17 @@ public class MapContextGTRecordWriter implements ICuboidWriter {
@Override
public void close() {
-
+
}
private void initVariables(Long cuboidId) {
- bytesLength = RowConstants.ROWKEY_HEADER_LEN;
- Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidId);
- for (TblColRef column : cuboid.getColumns()) {
- bytesLength += cubeSegment.getColumnLength(column);
- }
+ rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, Cuboid.findById(cubeDesc, cuboidId));
+ keyBuf = rowKeyEncoder.createBuf();
- keyBuf = new byte[bytesLength];
dimensions = BitSet.valueOf(new long[] { cuboidId }).cardinality();
measureColumnsIndex = new int[measureCount];
for (int i = 0; i < measureCount; i++) {
measureColumnsIndex[i] = dimensions + i;
}
-
- //write cuboid id first
- BytesUtil.writeLong(cuboidId, keyBuf, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_CUBOIDID_LEN);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
index 9b25c97..50f3d4c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
@@ -20,10 +20,10 @@ package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import java.util.HashMap;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.common.util.SplittedBytes;
@@ -33,6 +33,8 @@ import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.common.RowKeySplitter;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.RowKeyEncoder;
+import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.dict.Dictionary;
import org.apache.kylin.dict.DictionaryManager;
@@ -68,8 +70,10 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
private IMRStorageInputFormat storageInputFormat;
private ByteArrayWritable outputKey = new ByteArrayWritable();
- private byte[] newKeyBuf;
+ private byte[] newKeyBodyBuf;
+ private ByteArray newKeyBuf;
private RowKeySplitter rowKeySplitter;
+ private RowKeyEncoderProvider rowKeyEncoderProvider;
private HashMap<TblColRef, Boolean> dictsNeedMerging = new HashMap<TblColRef, Boolean>();
@@ -106,12 +110,14 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
mergedCubeSegment = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
storageInputFormat = MRUtil.getBatchMergeInputSide2(mergedCubeSegment).getStorageInputFormat();
- newKeyBuf = new byte[256]; // size will auto-grow
+ newKeyBodyBuf = new byte[RowConstants.ROWKEY_BUFFER_SIZE]; // size will auto-grow
+ newKeyBuf = ByteArray.allocate(RowConstants.ROWKEY_BUFFER_SIZE);
sourceCubeSegment = storageInputFormat.findSourceSegment(context);
logger.info("Source cube segment: " + sourceCubeSegment);
rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
+ rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
codec = new MeasureCodec(cubeDesc.getMeasures());
}
@@ -125,19 +131,15 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
Preconditions.checkState(key.offset() == 0);
long cuboidID = rowKeySplitter.split(key.array());
- short shard = rowKeySplitter.getLastSplittedShard();
Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidID);
+ RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
SplittedBytes[] splittedByteses = rowKeySplitter.getSplitBuffers();
int bufOffset = 0;
-
- BytesUtil.writeShort(shard, newKeyBuf, bufOffset, RowConstants.ROWKEY_SHARDID_LEN);
- bufOffset += RowConstants.ROWKEY_SHARDID_LEN;
-
- BytesUtil.writeLong(cuboidID, newKeyBuf, bufOffset, RowConstants.ROWKEY_CUBOIDID_LEN);
- bufOffset += RowConstants.ROWKEY_CUBOIDID_LEN;
+ int bodySplitOffset = rowKeySplitter.getBodySplitOffset();
for (int i = 0; i < cuboid.getColumns().size(); ++i) {
+ int useSplit = i + bodySplitOffset;
TblColRef col = cuboid.getColumns().get(i);
if (this.checkNeedMerging(col)) {
@@ -146,38 +148,48 @@ public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, By
Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(col));
Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(col));
- while (sourceDict.getSizeOfValue() > newKeyBuf.length - bufOffset || mergedDict.getSizeOfValue() > newKeyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBuf;
- newKeyBuf = new byte[2 * newKeyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBuf, 0, oldBuf.length);
+ while (sourceDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfId() > newKeyBodyBuf.length - bufOffset) {
+ //also use this buf to hold value before translating
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
}
- int idInSourceDict = BytesUtil.readUnsigned(splittedByteses[i + 1].value, 0, splittedByteses[i + 1].length);
+ int idInSourceDict = BytesUtil.readUnsigned(splittedByteses[useSplit].value, 0, splittedByteses[useSplit].length);
+ int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
- int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBuf, bufOffset);
int idInMergedDict;
if (size < 0) {
idInMergedDict = mergedDict.nullId();
} else {
- idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBuf, bufOffset, size);
+ idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
}
- BytesUtil.writeUnsigned(idInMergedDict, newKeyBuf, bufOffset, mergedDict.getSizeOfId());
+ BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
bufOffset += mergedDict.getSizeOfId();
} else {
// keep as it is
- while (splittedByteses[i + 1].length > newKeyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBuf;
- newKeyBuf = new byte[2 * newKeyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBuf, 0, oldBuf.length);
+ while (splittedByteses[useSplit].length > newKeyBodyBuf.length - bufOffset) {
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
}
- System.arraycopy(splittedByteses[i + 1].value, 0, newKeyBuf, bufOffset, splittedByteses[i + 1].length);
- bufOffset += splittedByteses[i + 1].length;
+ System.arraycopy(splittedByteses[useSplit].value, 0, newKeyBodyBuf, bufOffset, splittedByteses[useSplit].length);
+ bufOffset += splittedByteses[useSplit].length;
}
}
- byte[] newKey = Arrays.copyOf(newKeyBuf, bufOffset);
- outputKey.set(newKey, 0, newKey.length);
+
+ int fullKeySize = rowkeyEncoder.getBytesLength();
+ while (newKeyBuf.array().length < fullKeySize) {
+ newKeyBuf.set(new byte[newKeyBuf.length() * 2]);
+ }
+ newKeyBuf.set(0, fullKeySize);
+
+ rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, bufOffset), newKeyBuf);
+ outputKey.set(newKeyBuf.array(), 0, fullKeySize);
valueBuf.clear();
codec.encode(value, valueBuf);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/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 6301f3d..0b68e59 100644
--- 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
@@ -19,7 +19,6 @@
package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -27,6 +26,7 @@ import java.util.regex.Pattern;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.SplittedBytes;
import org.apache.kylin.cube.CubeInstance;
@@ -35,6 +35,8 @@ import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.common.RowKeySplitter;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.RowKeyEncoder;
+import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.dict.Dictionary;
import org.apache.kylin.dict.DictionaryManager;
@@ -60,8 +62,10 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
private Text outputKey = new Text();
- private byte[] newKeyBuf;
+ private byte[] newKeyBodyBuf;
+ private ByteArray newKeyBuf;
private RowKeySplitter rowKeySplitter;
+ private RowKeyEncoderProvider rowKeyEncoderProvider;
private HashMap<TblColRef, Boolean> dictsNeedMerging = new HashMap<TblColRef, Boolean>();
@@ -95,13 +99,15 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
mergedCubeSegment = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
// int colCount = cubeDesc.getRowkey().getRowKeyColumns().length;
- newKeyBuf = new byte[256];// size will auto-grow
+ newKeyBodyBuf = new byte[RowConstants.ROWKEY_BUFFER_SIZE];// size will auto-grow
+ newKeyBuf = ByteArray.allocate(RowConstants.ROWKEY_BUFFER_SIZE);
// decide which source segment
FileSplit fileSplit = (FileSplit) context.getInputSplit();
sourceCubeSegment = findSourceSegment(fileSplit, cube);
rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
+ rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
}
private static final Pattern JOB_NAME_PATTERN = Pattern.compile("kylin-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})");
@@ -135,17 +141,15 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
@Override
public void map(Text key, Text value, Context context) throws IOException, InterruptedException {
long cuboidID = rowKeySplitter.split(key.getBytes());
- short shard = rowKeySplitter.getLastSplittedShard();
Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidID);
+ RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
SplittedBytes[] splittedByteses = rowKeySplitter.getSplitBuffers();
int bufOffset = 0;
- BytesUtil.writeShort(shard, newKeyBuf, bufOffset, RowConstants.ROWKEY_SHARDID_LEN);
- bufOffset += RowConstants.ROWKEY_SHARDID_LEN;
- BytesUtil.writeLong(cuboidID, newKeyBuf, bufOffset, RowConstants.ROWKEY_CUBOIDID_LEN);
- bufOffset += RowConstants.ROWKEY_CUBOIDID_LEN;
+ int bodySplitOffset = rowKeySplitter.getBodySplitOffset();
for (int i = 0; i < cuboid.getColumns().size(); ++i) {
+ int useSplit = i + bodySplitOffset;
TblColRef col = cuboid.getColumns().get(i);
if (this.checkNeedMerging(col)) {
@@ -154,38 +158,47 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(col));
Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(col));
- while (sourceDict.getSizeOfValue() > newKeyBuf.length - bufOffset || mergedDict.getSizeOfValue() > newKeyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBuf;
- newKeyBuf = new byte[2 * newKeyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBuf, 0, oldBuf.length);
+ while (sourceDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
+ mergedDict.getSizeOfId() > newKeyBodyBuf.length - bufOffset) {
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
}
- int idInSourceDict = BytesUtil.readUnsigned(splittedByteses[i + 1].value, 0, splittedByteses[i + 1].length);
+ int idInSourceDict = BytesUtil.readUnsigned(splittedByteses[useSplit].value, 0, splittedByteses[useSplit].length);
int idInMergedDict;
- int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBuf, bufOffset);
+ int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
if (size < 0) {
idInMergedDict = mergedDict.nullId();
} else {
- idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBuf, bufOffset, size);
+ idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
}
- BytesUtil.writeUnsigned(idInMergedDict, newKeyBuf, bufOffset, mergedDict.getSizeOfId());
+ BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
bufOffset += mergedDict.getSizeOfId();
} else {
// keep as it is
- while (splittedByteses[i + 1].length > newKeyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBuf;
- newKeyBuf = new byte[2 * newKeyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBuf, 0, oldBuf.length);
+ while (splittedByteses[useSplit].length > newKeyBodyBuf.length - bufOffset) {
+ byte[] oldBuf = newKeyBodyBuf;
+ newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
+ System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
}
- System.arraycopy(splittedByteses[i + 1].value, 0, newKeyBuf, bufOffset, splittedByteses[i + 1].length);
- bufOffset += splittedByteses[i + 1].length;
+ System.arraycopy(splittedByteses[useSplit].value, 0, newKeyBodyBuf, bufOffset, splittedByteses[useSplit].length);
+ bufOffset += splittedByteses[useSplit].length;
}
}
- byte[] newKey = Arrays.copyOf(newKeyBuf, bufOffset);
- outputKey.set(newKey, 0, newKey.length);
+
+ int fullKeySize = rowkeyEncoder.getBytesLength();
+ while (newKeyBuf.array().length < fullKeySize) {
+ newKeyBuf.set(new byte[newKeyBuf.length() * 2]);
+ }
+ newKeyBuf.set(0, fullKeySize);
+
+ rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, bufOffset), newKeyBuf);
+ outputKey.set(newKeyBuf.array(), 0, fullKeySize);
context.write(outputKey, value);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/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 2180dd6..1dbce8e 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
@@ -23,8 +23,7 @@ import java.util.Collection;
import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.ShardingHash;
+import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.SplittedBytes;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
@@ -33,6 +32,8 @@ import org.apache.kylin.cube.common.RowKeySplitter;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.cuboid.CuboidScheduler;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.RowKeyEncoder;
+import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
@@ -59,8 +60,10 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
private int handleCounter;
private int skipCounter;
- private byte[] keyBuf = new byte[4096];
+ private byte[] newKeyBodyBuf = new byte[RowConstants.ROWKEY_BUFFER_SIZE];
+ private ByteArray newKeyBuf = ByteArray.allocate(RowConstants.ROWKEY_BUFFER_SIZE);
private RowKeySplitter rowKeySplitter;
+ private RowKeyEncoderProvider rowKeyEncoderProvider;
@Override
protected void setup(Context context) throws IOException {
@@ -79,32 +82,26 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
cuboidScheduler = new CuboidScheduler(cubeDesc);
rowKeySplitter = new RowKeySplitter(cubeSegment, 65, 256);
+ rowKeyEncoderProvider = new RowKeyEncoderProvider(cubeSegment);
}
private int buildKey(Cuboid parentCuboid, Cuboid childCuboid, SplittedBytes[] splitBuffers) {
- int offset = 0;
-
- //shard id will be filled after other contents
- offset += RowConstants.ROWKEY_SHARDID_LEN;
-
- // cuboid id
- System.arraycopy(childCuboid.getBytes(), 0, keyBuf, offset, childCuboid.getBytes().length);
- offset += RowConstants.ROWKEY_CUBOIDID_LEN;
+ RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(childCuboid);
- int bodyOffset = offset;
+ int offset = 0;
// rowkey columns
long mask = Long.highestOneBit(parentCuboid.getId());
long parentCuboidId = parentCuboid.getId();
long childCuboidId = childCuboid.getId();
long parentCuboidIdActualLength = Long.SIZE - Long.numberOfLeadingZeros(parentCuboid.getId());
- int index = 2; // skip shard and cuboidId
+ int index = rowKeySplitter.getBodySplitOffset(); // skip shard and cuboidId
for (int i = 0; i < parentCuboidIdActualLength; i++) {
if ((mask & parentCuboidId) > 0) {// if the this bit position equals
// 1
if ((mask & childCuboidId) > 0) {// if the child cuboid has this
// column
- System.arraycopy(splitBuffers[index].value, 0, keyBuf, offset, splitBuffers[index].length);
+ System.arraycopy(splitBuffers[index].value, 0, newKeyBodyBuf, offset, splitBuffers[index].length);
offset += splitBuffers[index].length;
}
index++;
@@ -112,13 +109,15 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
mask = mask >> 1;
}
- //fill shard
- short cuboidShardNum = cubeSegment.getCuboidShardNum(childCuboidId);
- short shardOffset = ShardingHash.getShard(keyBuf, bodyOffset, offset - bodyOffset, cuboidShardNum);
- short finalShard = ShardingHash.normalize(cubeSegment.getCuboidBaseShard(childCuboidId), shardOffset, cubeSegment.getTotalShards());
- BytesUtil.writeShort(finalShard, keyBuf, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ int fullKeySize = rowkeyEncoder.getBytesLength();
+ while (newKeyBuf.array().length < fullKeySize) {
+ newKeyBuf.set(new byte[newKeyBuf.length() * 2]);
+ }
+ newKeyBuf.set(0, fullKeySize);
+
+ rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, offset), newKeyBuf);
- return offset;
+ return fullKeySize;
}
@Override
@@ -147,8 +146,8 @@ public class NDCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
for (Long child : myChildren) {
Cuboid childCuboid = Cuboid.findById(cubeDesc, child);
- int keyLength = buildKey(parentCuboid, childCuboid, rowKeySplitter.getSplitBuffers());
- outputKey.set(keyBuf, 0, keyLength);
+ int fullKeySize = buildKey(parentCuboid, childCuboid, rowKeySplitter.getSplitBuffers());
+ outputKey.set(newKeyBuf.array(), 0, fullKeySize);
context.write(outputKey, value);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/MergeCuboidJobTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/MergeCuboidJobTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/MergeCuboidJobTest.java
index ccaa027..eacd37c 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/MergeCuboidJobTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/MergeCuboidJobTest.java
@@ -30,8 +30,10 @@ import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
+@Ignore("broken test, mergedCubeSegment in MergeCuboidMapper is not available. Besides, its input is difficult to maintain")
public class MergeCuboidJobTest extends LocalFileMetadataTestCase {
private Configuration conf;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
index 9e1fc2d..256f8a6 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
@@ -74,7 +74,7 @@ public class NDCuboidMapperTest extends LocalFileMetadataTestCase {
mapReduceDriver.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
mapReduceDriver.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_NAME, segmentName);
- byte[] key = { 0,0,0, 0, 0, 0, 0, 0, 1, -1, 49, 48, 48, 48, 48, 48, 48, 48, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 11, 54, -105, 55, 13, 71, 114, 65, 66, 73, 78, 9, 9, 9, 9, 9, 9, 9, 9, 0, 10, 0 };
+ byte[] key = { 0, 0, 0, 0, 0, 0, 0, 0, 1, -1, 49, 48, 48, 48, 48, 48, 48, 48, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 11, 54, -105, 55, 13, 71, 114, 65, 66, 73, 78, 9, 9, 9, 9, 9, 9, 9, 9, 0, 10, 0 };
byte[] value = { 14, 7, 23, -16, 56, 92, 114, -80, 118, 14, 7, 23, -16, 56, 92, 114, -80, 118, 14, 7, 23, -16, 56, 92, 114, -80, 118, 1, 1 };
Pair<Text, Text> input1 = new Pair<Text, Text>(new Text(key), new Text(value));
@@ -84,7 +84,7 @@ public class NDCuboidMapperTest extends LocalFileMetadataTestCase {
assertEquals(4, result.size());
- byte[] resultKey = { 0,0,0, 0, 0, 0, 0, 0, 1, 127, 49, 48, 48, 48, 48, 48, 48, 48, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 55, 13, 71, 114, 65, 66, 73, 78, 9, 9, 9, 9, 9, 9, 9, 9, 0, 10, 0 };
+ byte[] resultKey = { 0, 0, 0, 0, 0, 0, 0, 0, 1, 127, 49, 48, 48, 48, 48, 48, 48, 48, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 55, 13, 71, 114, 65, 66, 73, 78, 9, 9, 9, 9, 9, 9, 9, 9, 0, 10, 0 };
byte[] resultValue = { 14, 7, 23, -16, 56, 92, 114, -80, 118, 14, 7, 23, -16, 56, 92, 114, -80, 118, 14, 7, 23, -16, 56, 92, 114, -80, 118, 1, 1 };
Pair<Text, Text> output1 = new Pair<Text, Text>(new Text(resultKey), new Text(resultValue));
@@ -104,7 +104,7 @@ public class NDCuboidMapperTest extends LocalFileMetadataTestCase {
System.out.println(Bytes.toLong(new byte[] { 0, 0, 0, 0, 0, 0, 1, -1 }));
for (int i = 0; i < result.size(); i++) {
byte[] bytes = new byte[result.get(i).getFirst().getLength()];
- System.arraycopy(result.get(i).getFirst().getBytes(), RowConstants.ROWKEY_SHARDID_LEN, bytes, 0, result.get(i).getFirst().getLength()-RowConstants.ROWKEY_SHARDID_LEN);
+ System.arraycopy(result.get(i).getFirst().getBytes(), RowConstants.ROWKEY_SHARDID_LEN, bytes, 0, result.get(i).getFirst().getLength() - RowConstants.ROWKEY_SHARDID_LEN);
System.out.println(Bytes.toLong(bytes));
keySet[i] = Bytes.toLong(bytes);
}
[2/5] incubator-kylin git commit: KYLIN-1126 pscan backward
compability with v1 storage
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/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 986e45e..6b3a82c 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
@@ -22,12 +22,11 @@ import java.util.BitSet;
import java.util.Map;
import org.apache.kylin.common.util.ByteArray;
-import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.RowKeyEncoder;
+import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.metadata.model.TblColRef;
@@ -42,11 +41,14 @@ public final class DefaultTupleConverter implements TupleConverter {
private final CubeSegment segment;
private final int measureCount;
private final Map<TblColRef, Integer> columnLengthMap;
+ private RowKeyEncoderProvider rowKeyEncoderProvider;
+ private byte[] rowKeyBodyBuf = new byte[RowConstants.ROWKEY_BUFFER_SIZE];
public DefaultTupleConverter(CubeSegment segment, Map<TblColRef, Integer> columnLengthMap) {
this.segment = segment;
this.measureCount = segment.getCubeDesc().getMeasures().size();
this.columnLengthMap = columnLengthMap;
+ this.rowKeyEncoderProvider = new RowKeyEncoderProvider(this.segment);
}
private ByteBuffer getValueBuf() {
@@ -65,11 +67,8 @@ public final class DefaultTupleConverter implements TupleConverter {
@Override
public final Tuple2<byte[], byte[]> convert(long cuboidId, GTRecord record) {
- int bytesLength = RowConstants.ROWKEY_HEADER_LEN;
Cuboid cuboid = Cuboid.findById(segment.getCubeDesc(), cuboidId);
- for (TblColRef column : cuboid.getColumns()) {
- bytesLength += columnLengthMap.get(column);
- }
+ RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
final int dimensions = BitSet.valueOf(new long[] { cuboidId }).cardinality();
int[] measureColumnsIndex = getMeasureColumnsIndex();
@@ -77,22 +76,15 @@ public final class DefaultTupleConverter implements TupleConverter {
measureColumnsIndex[i] = dimensions + i;
}
- byte[] key = new byte[bytesLength];
- System.arraycopy(Bytes.toBytes(cuboidId), 0, key, 0, RowConstants.ROWKEY_CUBOIDID_LEN);
- int header = RowConstants.ROWKEY_HEADER_LEN;
- int offSet = header;
+ int offSet = 0;
for (int x = 0; x < dimensions; x++) {
final ByteArray byteArray = record.get(x);
- System.arraycopy(byteArray.array(), byteArray.offset(), key, offSet, byteArray.length());
+ System.arraycopy(byteArray.array(), byteArray.offset(), rowKeyBodyBuf, offSet, byteArray.length());
offSet += byteArray.length();
}
- //fill shard
- short cuboidShardNum = segment.getCuboidShardNum(cuboidId);
- short shardOffset = ShardingHash.getShard(key, header, offSet - header, cuboidShardNum);
- short cuboidShardBase = segment.getCuboidBaseShard(cuboidId);
- short finalShard = ShardingHash.normalize(cuboidShardBase, shardOffset, segment.getTotalShards());
- BytesUtil.writeShort(finalShard, key, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ byte[] rowKey = rowkeyEncoder.createBuf();
+ rowkeyEncoder.encode(new ByteArray(rowKeyBodyBuf), new ByteArray(rowKey));
ByteBuffer valueBuf = getValueBuf();
valueBuf.clear();
@@ -100,6 +92,6 @@ public final class DefaultTupleConverter implements TupleConverter {
byte[] value = new byte[valueBuf.position()];
System.arraycopy(valueBuf.array(), 0, value, 0, valueBuf.position());
- return new Tuple2<>(key, value);
+ return new Tuple2<>(rowKey, value);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/sample_cube/metadata/cube_desc/kylin_sales_cube_desc.json
----------------------------------------------------------------------
diff --git a/examples/sample_cube/metadata/cube_desc/kylin_sales_cube_desc.json b/examples/sample_cube/metadata/cube_desc/kylin_sales_cube_desc.json
index 26932b2..9320aaf 100644
--- a/examples/sample_cube/metadata/cube_desc/kylin_sales_cube_desc.json
+++ b/examples/sample_cube/metadata/cube_desc/kylin_sales_cube_desc.json
@@ -1,166 +1,225 @@
{
- "uuid" : "9ac9b7a8-3929-4dff-b59d-2100aadc8dbf",
- "name" : "kylin_sales_cube_desc",
- "description" : null,
- "engine_type": 2,
- "dimensions" : [ {
- "id" : 0,
- "name" : "CAL_DT",
- "table" : "DEFAULT.KYLIN_CAL_DT",
- "column" : null,
- "derived" : [ "WEEK_BEG_DT" ],
- "hierarchy" : false
- }, {
- "id" : 1,
- "name" : "CATEGORY",
- "table" : "DEFAULT.KYLIN_CATEGORY_GROUPINGS",
- "column" : null,
- "derived" : [ "USER_DEFINED_FIELD1", "USER_DEFINED_FIELD3", "UPD_DATE", "UPD_USER" ],
- "hierarchy" : false
- }, {
- "id" : 2,
- "name" : "CATEGORY_HIERARCHY",
- "table" : "DEFAULT.KYLIN_CATEGORY_GROUPINGS",
- "column" : [ "META_CATEG_NAME", "CATEG_LVL2_NAME", "CATEG_LVL3_NAME" ],
- "derived" : null,
- "hierarchy" : true
- }, {
- "id" : 3,
- "name" : "LSTG_FORMAT_NAME",
- "table" : "DEFAULT.KYLIN_SALES",
- "column" : [ "LSTG_FORMAT_NAME" ],
- "derived" : null,
- "hierarchy" : false
- } ],
- "measures" : [ {
- "id" : 1,
- "name" : "GMV_SUM",
- "function" : {
- "expression" : "SUM",
- "parameter" : {
- "type" : "column",
- "value" : "PRICE"
- },
- "returntype" : "decimal(19,4)"
+ "uuid": "9ac9b7a8-3929-4dff-b59d-2100aadc8dbf",
+ "name": "kylin_sales_cube_desc",
+ "description": null,
+ "dimensions": [
+ {
+ "id": 0,
+ "name": "CAL_DT",
+ "table": "DEFAULT.KYLIN_CAL_DT",
+ "column": null,
+ "derived": [
+ "WEEK_BEG_DT"
+ ],
+ "hierarchy": false
+ },
+ {
+ "id": 1,
+ "name": "CATEGORY",
+ "table": "DEFAULT.KYLIN_CATEGORY_GROUPINGS",
+ "column": null,
+ "derived": [
+ "USER_DEFINED_FIELD1",
+ "USER_DEFINED_FIELD3",
+ "UPD_DATE",
+ "UPD_USER"
+ ],
+ "hierarchy": false
+ },
+ {
+ "id": 2,
+ "name": "CATEGORY_HIERARCHY",
+ "table": "DEFAULT.KYLIN_CATEGORY_GROUPINGS",
+ "column": [
+ "META_CATEG_NAME",
+ "CATEG_LVL2_NAME",
+ "CATEG_LVL3_NAME"
+ ],
+ "derived": null,
+ "hierarchy": true
},
- "dependent_measure_ref" : null
- }, {
- "id" : 2,
- "name" : "GMV_MIN",
- "function" : {
- "expression" : "MIN",
- "parameter" : {
- "type" : "column",
- "value" : "PRICE"
+ {
+ "id": 3,
+ "name": "LSTG_FORMAT_NAME",
+ "table": "DEFAULT.KYLIN_SALES",
+ "column": [
+ "LSTG_FORMAT_NAME"
+ ],
+ "derived": null,
+ "hierarchy": false
+ }
+ ],
+ "measures": [
+ {
+ "id": 1,
+ "name": "GMV_SUM",
+ "function": {
+ "expression": "SUM",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
},
- "returntype" : "decimal(19,4)"
+ "dependent_measure_ref": null
},
- "dependent_measure_ref" : null
- }, {
- "id" : 3,
- "name" : "GMV_MAX",
- "function" : {
- "expression" : "MAX",
- "parameter" : {
- "type" : "column",
- "value" : "PRICE"
+ {
+ "id": 2,
+ "name": "GMV_MIN",
+ "function": {
+ "expression": "MIN",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
},
- "returntype" : "decimal(19,4)"
+ "dependent_measure_ref": null
},
- "dependent_measure_ref" : null
- }, {
- "id" : 4,
- "name" : "TRANS_CNT",
- "function" : {
- "expression" : "COUNT",
- "parameter" : {
- "type" : "constant",
- "value" : "1"
+ {
+ "id": 3,
+ "name": "GMV_MAX",
+ "function": {
+ "expression": "MAX",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
},
- "returntype" : "bigint"
+ "dependent_measure_ref": null
},
- "dependent_measure_ref" : null
- }, {
- "id" : 5,
- "name" : "SELLER_CNT_HLL",
- "function" : {
- "expression" : "COUNT_DISTINCT",
- "parameter" : {
- "type" : "column",
- "value" : "SELLER_ID"
+ {
+ "id": 4,
+ "name": "TRANS_CNT",
+ "function": {
+ "expression": "COUNT",
+ "parameter": {
+ "type": "constant",
+ "value": "1"
+ },
+ "returntype": "bigint"
},
- "returntype" : "hllc(10)"
+ "dependent_measure_ref": null
},
- "dependent_measure_ref" : null
- }, {
- "id" : 6,
- "name" : "SELLER_FORMAT_CNT",
- "function" : {
- "expression" : "COUNT_DISTINCT",
- "parameter" : {
- "type" : "column",
- "value" : "LSTG_FORMAT_NAME"
+ {
+ "id": 5,
+ "name": "SELLER_CNT_HLL",
+ "function": {
+ "expression": "COUNT_DISTINCT",
+ "parameter": {
+ "type": "column",
+ "value": "SELLER_ID"
+ },
+ "returntype": "hllc(10)"
},
- "returntype" : "hllc(10)"
+ "dependent_measure_ref": null
},
- "dependent_measure_ref" : null
- } ],
- "rowkey" : {
- "rowkey_columns" : [ {
- "column" : "part_dt",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- }, {
- "column" : "leaf_categ_id",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- }, {
- "column" : "meta_categ_name",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- }, {
- "column" : "categ_lvl2_name",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- }, {
- "column" : "categ_lvl3_name",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- }, {
- "column" : "lstg_format_name",
- "length" : 12,
- "dictionary" : null,
- "mandatory" : false
- }, {
- "column" : "lstg_site_id",
- "length" : 0,
- "dictionary" : "true",
- "mandatory" : false
- } ],
- "aggregation_groups" : [ [ "part_dt", "lstg_site_id", "leaf_categ_id", "meta_categ_name", "categ_lvl3_name", "categ_lvl2_name", "lstg_format_name" ] ]
+ {
+ "id": 6,
+ "name": "SELLER_FORMAT_CNT",
+ "function": {
+ "expression": "COUNT_DISTINCT",
+ "parameter": {
+ "type": "column",
+ "value": "LSTG_FORMAT_NAME"
+ },
+ "returntype": "hllc(10)"
+ },
+ "dependent_measure_ref": null
+ }
+ ],
+ "rowkey": {
+ "rowkey_columns": [
+ {
+ "column": "part_dt",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ },
+ {
+ "column": "leaf_categ_id",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ },
+ {
+ "column": "meta_categ_name",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ },
+ {
+ "column": "categ_lvl2_name",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ },
+ {
+ "column": "categ_lvl3_name",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ },
+ {
+ "column": "lstg_format_name",
+ "length": 12,
+ "dictionary": null,
+ "mandatory": false
+ },
+ {
+ "column": "lstg_site_id",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ }
+ ],
+ "aggregation_groups": [
+ [
+ "part_dt",
+ "lstg_site_id",
+ "leaf_categ_id",
+ "meta_categ_name",
+ "categ_lvl3_name",
+ "categ_lvl2_name",
+ "lstg_format_name"
+ ]
+ ]
},
- "last_modified" : 1426255280419,
- "model_name" : "kylin_sales_model",
- "null_string" : null,
- "hbase_mapping" : {
- "column_family" : [ {
- "name" : "f1",
- "columns" : [ {
- "qualifier" : "m",
- "measure_refs" : [ "gmv_sum", "gmv_min", "gmv_max", "trans_cnt" ]
- } ]
- }, {
- "name" : "f2",
- "columns" : [ {
- "qualifier" : "m",
- "measure_refs" : [ "seller_cnt_hll", "seller_format_cnt" ]
- } ]
- } ]
+ "last_modified": 1426255280419,
+ "model_name": "kylin_sales_model",
+ "null_string": null,
+ "hbase_mapping": {
+ "column_family": [
+ {
+ "name": "f1",
+ "columns": [
+ {
+ "qualifier": "m",
+ "measure_refs": [
+ "gmv_sum",
+ "gmv_min",
+ "gmv_max",
+ "trans_cnt"
+ ]
+ }
+ ]
+ },
+ {
+ "name": "f2",
+ "columns": [
+ {
+ "qualifier": "m",
+ "measure_refs": [
+ "seller_cnt_hll",
+ "seller_format_cnt"
+ ]
+ }
+ ]
+ }
+ ]
},
- "notify_list" : null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
index 17a3fdc..84cdaf4 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
@@ -2,7 +2,6 @@
"uuid": "4334a905-1fc6-4f67-985c-38fa5aeafd92",
"name": "test_kylin_cube_topn_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -143,5 +142,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
index 893ebcd..f7e700d 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_left_join_desc.json
@@ -2,7 +2,6 @@
"uuid": "5445a905-1fc6-4f67-985c-38fa5aeafd92",
"name": "test_kylin_cube_topn_left_join_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -130,7 +129,8 @@
]
}
]
- }, {
+ },
+ {
"name": "f2",
"columns": [
{
@@ -143,5 +143,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_desc.json
index 0b6c31a..0b99047 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_desc.json
@@ -2,7 +2,6 @@
"uuid": "a24ca905-1fc6-4f67-985c-38fa5aeafd92",
"name": "test_kylin_cube_with_slr_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -237,5 +236,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
index 1bd1ec5..8e22615 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
@@ -2,7 +2,6 @@
"uuid": "bbbba905-1fc6-4f67-985c-38fa5aeafd92",
"name": "test_kylin_cube_with_slr_left_join_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -237,5 +236,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
index c54b205..bd979e0 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_desc.json
@@ -2,7 +2,6 @@
"uuid": "9ac9b7a8-3929-4dff-b59d-2100aadc8dbf",
"name": "test_kylin_cube_without_slr_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -289,5 +288,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 2,
+ "storage_type": 0
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
index b4428cc..08a132e 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
@@ -2,7 +2,6 @@
"uuid": "9ac9b7a8-3929-4dff-b59d-2100aadc8dbf",
"name": "test_kylin_cube_without_slr_left_join_desc",
"description": null,
- "engine_type": 2,
"dimensions": [
{
"id": 0,
@@ -289,5 +288,7 @@
}
]
},
- "notify_list": null
+ "notify_list": null,
+ "engine_type": 0,
+ "storage_type": 0
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/examples/test_case_data/localmeta/cube_desc/test_streaming_table_cube_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_streaming_table_cube_desc.json b/examples/test_case_data/localmeta/cube_desc/test_streaming_table_cube_desc.json
index ebf656a..c9a6536 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_streaming_table_cube_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_streaming_table_cube_desc.json
@@ -2,7 +2,6 @@
"uuid": "901ed15e-7769-4c66-b7ae-fbdc971cd192",
"name": "test_streaming_table_cube_desc",
"description": "",
- "engine_type": 2,
"dimensions": [
{
"id": 1,
@@ -140,5 +139,7 @@
}
]
},
- "notify_list": []
+ "notify_list": [],
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorProjector.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorProjector.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorProjector.java
index 9b839c3..c37b2f4 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorProjector.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorProjector.java
@@ -41,9 +41,8 @@ public class CoprocessorProjector {
RowKeyEncoder rowKeyMaskEncoder = new RowKeyEncoder(cubeSegment, cuboid) {
@Override
- protected int fillHeader(byte[] bytes) {
- Arrays.fill(bytes, 0, this.headerLength, (byte) 0xff);
- return this.headerLength;
+ protected void fillHeader(byte[] bytes) {
+ Arrays.fill(bytes, 0, this.getHeaderLength(), (byte) 0xff);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorRowType.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorRowType.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorRowType.java
index 7ec97c0..35488d1 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorRowType.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/CoprocessorRowType.java
@@ -26,7 +26,6 @@ import org.apache.kylin.common.util.BytesSerializer;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.kv.RowKeyColumnIO;
import org.apache.kylin.invertedindex.index.TableRecordInfo;
import org.apache.kylin.metadata.model.ColumnDesc;
@@ -47,7 +46,9 @@ public class CoprocessorRowType {
for (int i = 0; i < cols.size(); i++) {
colSizes[i] = tableRecordInfo.getDigest().length(i);
}
- return new CoprocessorRowType(cols.toArray(new TblColRef[cols.size()]), colSizes);
+
+ //TODO:check0
+ return new CoprocessorRowType(cols.toArray(new TblColRef[cols.size()]), colSizes, 0);
}
//for observer
@@ -59,7 +60,7 @@ public class CoprocessorRowType {
for (int i = 0; i < cols.length; i++) {
colSizes[i] = colIO.getColumnLength(cols[i]);
}
- return new CoprocessorRowType(cols, colSizes);
+ return new CoprocessorRowType(cols, colSizes, seg.getRowKeyPreambleSize());
}
public static byte[] serialize(CoprocessorRowType o) {
@@ -82,6 +83,7 @@ public class CoprocessorRowType {
public void serialize(CoprocessorRowType o, ByteBuffer out) {
int n = o.columns.length;
BytesUtil.writeVInt(o.columns.length, out);
+ BytesUtil.writeVInt(o.bodyOffset, out);
for (int i = 0; i < n; i++) {
BytesUtil.writeAsciiString(o.columns[i].getTable(), out);
BytesUtil.writeAsciiString(o.columns[i].getName(), out);
@@ -92,6 +94,7 @@ public class CoprocessorRowType {
@Override
public CoprocessorRowType deserialize(ByteBuffer in) {
int n = BytesUtil.readVInt(in);
+ int bodyOffset = BytesUtil.readVInt(in);
TblColRef[] cols = new TblColRef[n];
int[] colSizes = new int[n];
for (int i = 0; i < n; i++) {
@@ -108,18 +111,20 @@ public class CoprocessorRowType {
int colSize = BytesUtil.readVInt(in);
colSizes[i] = colSize;
}
- return new CoprocessorRowType(cols, colSizes);
+ return new CoprocessorRowType(cols, colSizes, bodyOffset);
}
}
// ============================================================================
public TblColRef[] columns;
+ private int bodyOffset;
public int[] columnSizes;
public int[] columnOffsets;
public HashMap<TblColRef, Integer> columnIdxMap;
- public CoprocessorRowType(TblColRef[] columns, int[] columnSizes) {
+ public CoprocessorRowType(TblColRef[] columns, int[] columnSizes, int bodyOffset) {
+ this.bodyOffset = bodyOffset;
this.columns = columns;
this.columnSizes = columnSizes;
init();
@@ -131,7 +136,7 @@ public class CoprocessorRowType {
private void init() {
int[] offsets = new int[columns.length];
- int o = RowConstants.ROWKEY_HEADER_LEN;
+ int o = bodyOffset;
for (int i = 0; i < columns.length; i++) {
offsets[i] = o;
o += columnSizes[i];
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
index 034ffac..22f7017 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
@@ -254,7 +254,6 @@ public class CubeSegmentTupleIterator implements ITupleIterator {
flushScanCountDelta();
if (logger.isDebugEnabled() && scan != null) {
- logger.debug("Scan " + scan.toString());
byte[] metricsBytes = scan.getAttribute(Scan.SCAN_ATTRIBUTES_METRICS_DATA);
if (metricsBytes != null) {
ScanMetrics scanMetrics = ProtobufUtil.toScanMetrics(metricsBytes);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
index c62308e..f84e4e6 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
@@ -33,9 +33,6 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
-import com.google.common.collect.Sets;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesUtil;
@@ -75,6 +72,9 @@ 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.Range;
+import com.google.common.collect.Sets;
@SuppressWarnings("unused")
public class CubeStorageQuery implements ICachableStorageQuery {
@@ -482,7 +482,10 @@ public class CubeStorageQuery implements ICachableStorageQuery {
dropUnhitSegments(result);
logger.info("hbasekeyrange count after dropping unhit :" + result.size());
- result = duplicateRangeByShard(result);
+ //TODO: should use LazyRowKeyEncoder.getRowKeysDifferentShards like CubeHBaseRPC, not do so because v1 query engine is retiring. not worth changing it
+ if (cubeDesc.isEnableSharding()) {
+ result = duplicateRangeByShard(result);
+ }
logger.info("hbasekeyrange count after dropping duplicatebyshard :" + result.size());
return result;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
index 6ad30e2..b606d2e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
@@ -25,9 +25,12 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.DataFormatException;
@@ -163,12 +166,13 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
final AtomicInteger totalScannedCount = new AtomicInteger(0);
final String toggle = BackdoorToggles.getCoprocessorBehavior() == null ? CoprocessorBehavior.SCAN_FILTER_AGGR_CHECKMEM.toString() : BackdoorToggles.getCoprocessorBehavior();
logger.info("The execution of this query will use " + toggle + " as endpoint's behavior");
+ List<Future<?>> futures = Lists.newArrayList();
for (int i = 0; i < rawScans.size(); ++i) {
final int shardIndex = i;
final RawScan rawScan = rawScans.get(i);
- executorService.submit(new Runnable() {
+ Future<?> future = executorService.submit(new Runnable() {
@Override
public void run() {
final byte[] rawScanBytes = KryoUtils.serialize(rawScan);
@@ -177,7 +181,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
for (IntList intList : hbaseColumnsToGTIntList) {
builder.addHbaseColumnsToGT(intList);
}
-
+ builder.setRowkeyPreambleSize(cubeSeg.getRowKeyPreambleSize());
builder.setBehavior(toggle);
Collection<CubeVisitProtos.CubeVisitResponse> results;
@@ -211,14 +215,19 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
rowBlocks.addAll(part);
}
});
+ futures.add(future);
}
executorService.shutdown();
try {
- if (!executorService.awaitTermination(1, TimeUnit.HOURS)) {
- throw new RuntimeException("Visiting cube by endpoint timeout");
+ for (Future<?> future : futures) {
+ future.get(1, TimeUnit.HOURS);
}
} catch (InterruptedException e) {
throw new RuntimeException("Visiting cube by endpoint gets interrupted");
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Visiting cube throw exception", e);
+ } catch (TimeoutException e) {
+ throw new RuntimeException("Visiting cube by endpoint timeout");
}
return new EndpointResultsAsGTScanner(fullGTInfo, rowBlocks.iterator(), scanRequest.getColumns(), totalScannedCount.get());
@@ -227,10 +236,11 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
private String getStatsString(CubeVisitProtos.CubeVisitResponse result, int shardIndex) {
StringBuilder sb = new StringBuilder();
Stats stats = result.getStats();
- sb.append("Shard " + shardIndex + ": ");
+ sb.append("Shard " + shardIndex + " on host: " + stats.getHostname());
sb.append("Total scanned row: " + stats.getScannedRowCount() + ". ");
sb.append("Total filtered/aggred row: " + stats.getAggregatedRowCount() + ". ");
sb.append("Time elapsed in EP: " + (stats.getServiceEndTime() - stats.getServiceStartTime()) + "(ms). ");
+ sb.append("Server CPU usage: " + stats.getSystemCpuLoad() + ", server physical mem left: " + stats.getFreePhysicalMemorySize() + ", server swap mem left:" + stats.getFreeSwapSpaceSize());
return sb.toString();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
index 1d217ac..412e7602 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
@@ -5,20 +5,22 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import org.apache.hadoop.hbase.Cell;
+import javax.annotation.Nullable;
+
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FuzzyRowFilter;
-import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.kv.FuzzyKeyEncoder;
+import org.apache.kylin.cube.kv.FuzzyMaskEncoder;
+import org.apache.kylin.cube.kv.LazyRowKeyEncoder;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.kv.RowKeyEncoder;
import org.apache.kylin.cube.model.HBaseColumnDesc;
import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
import org.apache.kylin.cube.model.HBaseMappingDesc;
@@ -29,6 +31,7 @@ import org.apache.kylin.gridtable.IGTScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -41,11 +44,16 @@ public abstract class CubeHBaseRPC {
final protected CubeSegment cubeSeg;
final protected Cuboid cuboid;
final protected GTInfo fullGTInfo;
+ final private RowKeyEncoder fuzzyKeyEncoder;
+ final private RowKeyEncoder fuzzyMaskEncoder;
public CubeHBaseRPC(CubeSegment cubeSeg, Cuboid cuboid, GTInfo fullGTInfo) {
this.cubeSeg = cubeSeg;
this.cuboid = cuboid;
this.fullGTInfo = fullGTInfo;
+
+ this.fuzzyKeyEncoder = new FuzzyKeyEncoder(cubeSeg, cuboid);
+ this.fuzzyMaskEncoder = new FuzzyMaskEncoder(cubeSeg, cuboid);
}
abstract IGTScanner getGTScanner(GTScanRequest scanRequest) throws IOException;
@@ -76,23 +84,34 @@ public abstract class CubeHBaseRPC {
final List<Pair<byte[], byte[]>> selectedColumns = makeHBaseColumns(selectedColBlocks);
List<RawScan> ret = Lists.newArrayList();
- byte[] start = makeRowKeyToScan(pkStart, RowConstants.ROWKEY_LOWER_BYTE);
- byte[] end = makeRowKeyToScan(pkEnd, RowConstants.ROWKEY_UPPER_BYTE);
+ LazyRowKeyEncoder encoder = new LazyRowKeyEncoder(cubeSeg, cuboid);
+ byte[] start = encoder.createBuf();
+ byte[] end = encoder.createBuf();
+ List<byte[]> startKeys;
+ List<byte[]> endKeys;
+
+ encoder.setBlankByte(RowConstants.ROWKEY_LOWER_BYTE);
+ encoder.encode(pkStart, pkStart.getInfo().getPrimaryKey(), start);
+ startKeys = encoder.getRowKeysDifferentShards(start);
+
+ encoder.setBlankByte(RowConstants.ROWKEY_UPPER_BYTE);
+ encoder.encode(pkEnd, pkEnd.getInfo().getPrimaryKey(), end);
+ endKeys = encoder.getRowKeysDifferentShards(end);
+ endKeys = Lists.transform(endKeys, new Function<byte[], byte[]>() {
+ @Nullable
+ @Override
+ public byte[] apply(byte[] input) {
+ byte[] shardEnd = new byte[input.length + 1];//append extra 0 to the end key to make it inclusive while scanning
+ System.arraycopy(input, 0, shardEnd, 0, input.length);
+ return shardEnd;
+ }
+ });
+
+ Preconditions.checkState(startKeys.size() == endKeys.size());
List<Pair<byte[], byte[]>> hbaseFuzzyKeys = translateFuzzyKeys(fuzzyKeys);
- short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
-
- for (short i = 0; i < cuboidShardNum; ++i) {
- short shard = ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), i, cubeSeg.getTotalShards());
-
- byte[] shardStart = Arrays.copyOf(start, start.length);
- byte[] shardEnd = new byte[end.length + 1];//append extra 0 to the end key to make it inclusive while scanning
- System.arraycopy(end, 0, shardEnd, 0, end.length);
-
- BytesUtil.writeShort(shard, shardStart, 0, RowConstants.ROWKEY_SHARDID_LEN);
- BytesUtil.writeShort(shard, shardEnd, 0, RowConstants.ROWKEY_SHARDID_LEN);
-
- ret.add(new RawScan(shardStart, shardEnd, selectedColumns, hbaseFuzzyKeys));
+ for (short i = 0; i < startKeys.size(); ++i) {
+ ret.add(new RawScan(startKeys.get(i), endKeys.get(i), selectedColumns, hbaseFuzzyKeys));
}
return ret;
@@ -108,30 +127,12 @@ public abstract class CubeHBaseRPC {
}
List<Pair<byte[], byte[]>> ret = Lists.newArrayList();
- int coreLength = fullGTInfo.getMaxColumnLength(fullGTInfo.getPrimaryKey());
for (GTRecord gtRecordFuzzyKey : fuzzyKeys) {
- byte[] hbaseFuzzyKey = new byte[coreLength + RowConstants.ROWKEY_HEADER_LEN];
- byte[] hbaseFuzzyMask = new byte[coreLength + RowConstants.ROWKEY_HEADER_LEN];
-
- int pos = 0;
- //shard part
- Arrays.fill(hbaseFuzzyMask, pos, pos + RowConstants.ROWKEY_SHARDID_LEN, RowConstants.BYTE_ONE);//shard part should better be FIXED, for simplicity we make it non-fixed
- pos += RowConstants.ROWKEY_SHARDID_LEN;
-
- //cuboid part
- Arrays.fill(hbaseFuzzyMask, pos, pos + RowConstants.ROWKEY_CUBOIDID_LEN, RowConstants.BYTE_ZERO);
- System.arraycopy(cuboid.getBytes(), 0, hbaseFuzzyKey, pos, RowConstants.ROWKEY_CUBOIDID_LEN);
- pos += RowConstants.ROWKEY_CUBOIDID_LEN;
+ byte[] hbaseFuzzyKey = fuzzyKeyEncoder.createBuf();
+ byte[] hbaseFuzzyMask = fuzzyMaskEncoder.createBuf();
- //row key core part
- ByteArray coreKey = HBaseScan.exportScanKey(gtRecordFuzzyKey, RowConstants.BYTE_ZERO);
- System.arraycopy(coreKey.array(), coreKey.offset(), hbaseFuzzyKey, pos, coreKey.length());
- ByteArray coreMask = HBaseScan.exportScanMask(gtRecordFuzzyKey);
- System.arraycopy(coreMask.array(), coreMask.offset(), hbaseFuzzyMask, pos, coreMask.length());
-
- Preconditions.checkState(coreKey.length() == coreMask.length(), "corekey length not equal coremask length");
- pos += coreKey.length();
- Preconditions.checkState(hbaseFuzzyKey.length == pos, "HBase fuzzy key not completely populated");
+ fuzzyKeyEncoder.encode(gtRecordFuzzyKey, gtRecordFuzzyKey.getInfo().getPrimaryKey(), hbaseFuzzyKey);
+ fuzzyMaskEncoder.encode(gtRecordFuzzyKey, gtRecordFuzzyKey.getInfo().getPrimaryKey(), hbaseFuzzyMask);
ret.add(new Pair<byte[], byte[]>(hbaseFuzzyKey, hbaseFuzzyMask));
}
@@ -139,21 +140,6 @@ public abstract class CubeHBaseRPC {
return ret;
}
- private byte[] makeRowKeyToScan(GTRecord pkRec, byte fill) {
- ByteArray pk = HBaseScan.exportScanKey(pkRec, fill);
-
- byte[] buf = new byte[pk.length() + RowConstants.ROWKEY_HEADER_LEN];
- Arrays.fill(buf, fill);
-
- //for scanning/reading, later all possible shard will be applied
-
- System.arraycopy(cuboid.getBytes(), 0, buf, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_CUBOIDID_LEN);
- if (pk != null && pk.array() != null) {
- System.arraycopy(pk.array(), pk.offset(), buf, RowConstants.ROWKEY_HEADER_LEN, pk.length());
- }
- return buf;
- }
-
/**
* prune untouched hbase columns
*/
@@ -206,8 +192,6 @@ public abstract class CubeHBaseRPC {
return ret;
}
-
-
public static void applyHBaseColums(Scan scan, List<Pair<byte[], byte[]>> hbaseColumns) {
for (Pair<byte[], byte[]> hbaseColumn : hbaseColumns) {
byte[] byteFamily = hbaseColumn.getFirst();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
index fa5a844..69b95ca 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
@@ -118,7 +118,7 @@ public class CubeHBaseScanRPC extends CubeHBaseRPC {
}
};
- IGTStore store = new HBaseReadonlyStore(cellListIterator, scanRequest, rawScans.get(0).hbaseColumns, hbaseColumnsToGT);
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanRequest, rawScans.get(0).hbaseColumns, hbaseColumnsToGT,cubeSeg.getRowKeyPreambleSize());
IGTScanner rawScanner = store.scan(scanRequest);
final IGTScanner decorateScanner = scanRequest.decorateScanner(rawScanner);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
index 7731f19..303c360 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
@@ -43,13 +43,14 @@ public class HBaseReadonlyStore implements IGTStore {
private GTInfo info;
private List<Pair<byte[], byte[]>> hbaseColumns;
private List<List<Integer>> hbaseColumnsToGT;
+ private int rowkeyPreambleSize;
- public HBaseReadonlyStore(CellListIterator cellListIterator, GTScanRequest gtScanRequest, List<Pair<byte[], byte[]>> hbaseColumns, List<List<Integer>> hbaseColumnsToGT) {
+ public HBaseReadonlyStore(CellListIterator cellListIterator, GTScanRequest gtScanRequest, List<Pair<byte[], byte[]>> hbaseColumns, List<List<Integer>> hbaseColumnsToGT, int rowkeyPreambleSize) {
this.cellListIterator = cellListIterator;
-
this.info = gtScanRequest.getInfo();
this.hbaseColumns = hbaseColumns;
this.hbaseColumnsToGT = hbaseColumnsToGT;
+ this.rowkeyPreambleSize = rowkeyPreambleSize;
}
@Override
@@ -108,7 +109,7 @@ public class HBaseReadonlyStore implements IGTStore {
// dimensions, set to primary key, also the 0th column block
Cell firstCell = oneRow.get(0);
- ByteBuffer buf = byteBuffer(firstCell.getRowArray(), RowConstants.ROWKEY_HEADER_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_HEADER_LEN);
+ ByteBuffer buf = byteBuffer(firstCell.getRowArray(), rowkeyPreambleSize + firstCell.getRowOffset(), firstCell.getRowLength() - rowkeyPreambleSize);
oneRecord.loadCellBlock(0, buf);
// metrics
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseScan.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseScan.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseScan.java
deleted file mode 100644
index 65a963d..0000000
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseScan.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kylin.storage.hbase.cube.v2;
-
-import java.util.Arrays;
-
-import org.apache.kylin.common.util.ByteArray;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRecord;
-
-import com.google.common.base.Preconditions;
-
-public class HBaseScan {
-
- /**
- * every column in scan key is fixed length. for empty values, 0 zero will be populated
- */
- public static ByteArray exportScanKey(GTRecord rec, byte fill) {
-
- Preconditions.checkNotNull(rec);
-
- GTInfo info = rec.getInfo();
- int len = info.getMaxColumnLength(info.getPrimaryKey());
- ByteArray buf = ByteArray.allocate(len);
- int pos = 0;
- for (int i = 0; i < info.getPrimaryKey().trueBitCount(); i++) {
- int c = info.getPrimaryKey().trueBitAt(i);
- int colLength = info.getCodeSystem().maxCodeLength(c);
-
- if (rec.get(c).array() != null) {
- Preconditions.checkArgument(colLength == rec.get(c).length(), "ColLength :" + colLength + " != cols[c].length: " + rec.get(c).length() + ", c is " + c);
- System.arraycopy(rec.get(c).array(), rec.get(c).offset(), buf.array(), buf.offset() + pos, rec.get(c).length());
- } else {
- Arrays.fill(buf.array(), buf.offset() + pos, buf.offset() + pos + colLength, fill);
- }
- pos += colLength;
- }
- buf.setLength(pos);
-
- return buf;
- }
-
- /**
- * every column in scan key is fixed length. for fixed columns, 0 will be populated, for non-fixed columns, 1 will be populated
- */
- public static ByteArray exportScanMask(GTRecord rec) {
- Preconditions.checkNotNull(rec);
-
- GTInfo info = rec.getInfo();
- int len = info.getMaxColumnLength(info.getPrimaryKey());
- ByteArray buf = ByteArray.allocate(len);
- byte fill;
-
- int pos = 0;
- for (int i = 0; i < info.getPrimaryKey().trueBitCount(); i++) {
- int c = info.getPrimaryKey().trueBitAt(i);
- int colLength = info.getCodeSystem().maxCodeLength(c);
-
- if (rec.get(c).array() != null) {
- fill = RowConstants.BYTE_ZERO;
- } else {
- fill = RowConstants.BYTE_ONE;
- }
- Arrays.fill(buf.array(), buf.offset() + pos, buf.offset() + pos + colLength, fill);
- pos += colLength;
- }
- buf.setLength(pos);
-
- return buf;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fce575bc/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
index f474139..3759738 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
@@ -20,6 +20,8 @@ package org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.List;
@@ -56,6 +58,7 @@ import com.google.protobuf.HBaseZeroCopyByteString;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
+import com.sun.management.OperatingSystemMXBean;
@SuppressWarnings("unused")
//used in hbase endpoint
@@ -144,7 +147,7 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
innerScanner = region.getScanner(scan);
InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner);
- IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns, hbaseColumnsToGT);
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize());
IGTScanner rawScanner = store.scan(scanReq);
CoprocessorBehavior behavior = CoprocessorBehavior.valueOf(request.getBehavior());
@@ -165,6 +168,12 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
outputStream.write(buffer.array(), buffer.arrayOffset() - buffer.position(), buffer.remaining());
finalRowCount++;
}
+
+ OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
+ double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad();
+ double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize();
+ double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize();
+
//outputStream.close() is not necessary
byte[] allRows = outputStream.toByteArray();
CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder();
@@ -174,7 +183,12 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
setAggregatedRowCount(finalScanner.getScannedRowCount() - finalRowCount).//
setScannedRowCount(finalScanner.getScannedRowCount()).//
setServiceStartTime(serviceStartTime).//
- setServiceEndTime(System.currentTimeMillis()).build()).//
+ setServiceEndTime(System.currentTimeMillis()).//
+ setSystemCpuLoad(systemCpuLoad).//
+ setFreePhysicalMemorySize(freePhysicalMemorySize).//
+ setFreeSwapSpaceSize(freeSwapSpaceSize).//
+ setHostname(InetAddress.getLocalHost().getHostName()).//
+ build()).//
build());
} catch (IOException ioe) {
[4/5] incubator-kylin git commit: KYLIN-1136 Distinguish fast build
mode and complete build mode
Posted by ma...@apache.org.
KYLIN-1136 Distinguish fast build mode and complete build mode
bug fix
fix
temp
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/d256a7f3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/d256a7f3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/d256a7f3
Branch: refs/heads/2.x-staging
Commit: d256a7f30f1174aa0efa74ab58ca17a329e6c7f7
Parents: fce575b
Author: honma <ho...@ebay.com>
Authored: Tue Nov 10 16:34:39 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Mon Nov 16 10:28:24 2015 +0800
----------------------------------------------------------------------
.../kylin/job/BuildCubeWithEngineTest.java | 114 ++++++++++++-------
.../org/apache/kylin/cube/CubeManagerTest.java | 5 +
...test_kylin_cube_with_slr_left_join_desc.json | 4 +-
...t_kylin_cube_without_slr_left_join_desc.json | 6 +-
4 files changed, 86 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d256a7f3/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
index af3dc43..d2a101d 100644
--- a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
@@ -62,6 +62,7 @@ public class BuildCubeWithEngineTest {
private CubeManager cubeManager;
private DefaultScheduler scheduler;
protected ExecutableManager jobService;
+ private static boolean fastBuildMode = true;
private static final Log logger = LogFactory.getLog(BuildCubeWithEngineTest.class);
@@ -84,6 +85,15 @@ public class BuildCubeWithEngineTest {
public static void beforeClass() throws Exception {
logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+
+ String fastModeStr = System.getProperty("fastBuildMode");
+ if (fastModeStr != null && fastModeStr.equalsIgnoreCase("false")) {
+ fastBuildMode = false;
+ logger.info("Will not use fast build mode");
+ } else {
+ logger.info("Will use fast build mode");
+ }
+
System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
if (System.getProperty("hdp.version") == null) {
throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
@@ -93,6 +103,7 @@ public class BuildCubeWithEngineTest {
@Before
public void before() throws Exception {
+
HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
DeployUtil.initCliWorkDir();
@@ -128,12 +139,12 @@ public class BuildCubeWithEngineTest {
}
private void testInner() throws Exception {
- String[] testCase = new String[] { "testInnerJoinCube", "testInnerJoinCube2" };
+ String[] testCase = new String[] { "testInnerJoinCubeWithoutSlr", "testInnerJoinCubeWithSlr" };
runTestAndAssertSucceed(testCase);
}
private void testLeft() throws Exception {
- String[] testCase = new String[] { "testLeftJoinCube", "testLeftJoinCube2" };
+ String[] testCase = new String[] { "testLeftJoinCubeWithSlr", "testLeftJoinCubeWithoutSlr" };
runTestAndAssertSucceed(testCase);
}
@@ -180,6 +191,9 @@ public class BuildCubeWithEngineTest {
final Method method = BuildCubeWithEngineTest.class.getDeclaredMethod(methodName);
method.setAccessible(true);
return (List<String>) method.invoke(BuildCubeWithEngineTest.this);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ throw e;
} finally {
countDownLatch.countDown();
}
@@ -188,63 +202,71 @@ public class BuildCubeWithEngineTest {
@SuppressWarnings("unused")
// called by reflection
- private List<String> testInnerJoinCube2() throws Exception {
+ private List<String> testInnerJoinCubeWithSlr() throws Exception {
clearSegment("test_kylin_cube_with_slr_empty");
SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
f.setTimeZone(TimeZone.getTimeZone("GMT"));
long date1 = 0;
- long date2 = f.parse("2013-01-01").getTime();
+ long date2 = f.parse("2015-01-01").getTime();
long date3 = f.parse("2022-01-01").getTime();
List<String> result = Lists.newArrayList();
- result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date2));
- result.add(buildSegment("test_kylin_cube_with_slr_empty", date2, date3));
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date3));
+ } else {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date2, date3));//empty segment
+ }
return result;
}
@SuppressWarnings("unused")
// called by reflection
- private List<String> testInnerJoinCube() throws Exception {
- clearSegment("test_kylin_cube_without_slr_empty");
+ private List<String> testInnerJoinCubeWithoutSlr() throws Exception {
+ clearSegment("test_kylin_cube_without_slr_empty");
SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
f.setTimeZone(TimeZone.getTimeZone("GMT"));
-
- // this cube's start date is 0, end date is 20501112000000
long date1 = 0;
- long date2 = f.parse("2050-01-11").getTime();
-
- // this cube doesn't support incremental build, always do full build
-
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
List<String> result = Lists.newArrayList();
- result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date2));
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date3, date4));
+ result.add(mergeSegment("test_kylin_cube_without_slr_empty", date1, date3));//don't merge all segments
+ }
return result;
+
}
-
+
@SuppressWarnings("unused")
// called by reflection
- private List<String> testLeftJoinCube2() throws Exception {
+ private List<String> testLeftJoinCubeWithoutSlr() throws Exception {
SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
f.setTimeZone(TimeZone.getTimeZone("GMT"));
List<String> result = Lists.newArrayList();
final String cubeName = "test_kylin_cube_without_slr_left_join_empty";
- // this cube's start date is 0, end date is 20120601000000
- long dateStart = cubeManager.getCube(cubeName).getDescriptor().getModel().getPartitionDesc().getPartitionDateStart();
-
- long dateEnd = f.parse("2012-06-01").getTime();
-
clearSegment(cubeName);
- result.add(buildSegment(cubeName, dateStart, dateEnd));
- // then submit an append job, start date is 20120601000000, end
- // date is 20220101000000
- dateStart = dateEnd;
- dateEnd = f.parse("2022-01-01").getTime();
- result.add(buildSegment(cubeName, dateStart, dateEnd));
-
- // build an empty segment which doesn't have data
- dateStart = dateEnd;
- dateEnd = f.parse("2023-01-01").getTime();
- result.add(buildSegment(cubeName, dateStart, dateEnd));
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getModel().getPartitionDesc().getPartitionDateStart();
+ long date2 = f.parse("2012-06-01").getTime();
+ long date3 = f.parse("2022-01-01").getTime();
+ long date4 = f.parse("2023-01-01").getTime();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date3, date4));//empty segment
+ result.add(mergeSegment("test_kylin_cube_without_slr_left_join_empty", date1, date3));//don't merge all segments
+ }
return result;
@@ -252,18 +274,26 @@ public class BuildCubeWithEngineTest {
@SuppressWarnings("unused")
// called by reflection
- private List<String> testLeftJoinCube() throws Exception {
+ private List<String> testLeftJoinCubeWithSlr() throws Exception {
String cubeName = "test_kylin_cube_with_slr_left_join_empty";
clearSegment(cubeName);
SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
f.setTimeZone(TimeZone.getTimeZone("GMT"));
- long dateStart = cubeManager.getCube(cubeName).getDescriptor().getModel().getPartitionDesc().getPartitionDateStart();
- long dateEnd = f.parse("2050-11-12").getTime();
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getModel().getPartitionDesc().getPartitionDateStart();
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
- // this cube's start date is 0, end date is 20501112000000
List<String> result = Lists.newArrayList();
- result.add(buildSegment(cubeName, dateStart, dateEnd));
+ if (fastBuildMode) {
+ result.add(buildSegment(cubeName, date1, date4));
+ } else {
+ result.add(buildSegment(cubeName, date1, date2));
+ result.add(buildSegment(cubeName, date2, date3));
+ result.add(buildSegment(cubeName, date3, date4));
+ result.add(mergeSegment(cubeName, date1, date3));//don't merge all segments
+ }
return result;
}
@@ -276,6 +306,14 @@ public class BuildCubeWithEngineTest {
cubeManager.updateCube(cubeBuilder);
}
+ private String mergeSegment(String cubeName, long startDate, long endDate) throws Exception {
+ CubeSegment segment = cubeManager.mergeSegments(cubeManager.getCube(cubeName), startDate, endDate, true);
+ DefaultChainedExecutable job = EngineFactory.createBatchMergeJob(segment, "TEST");
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
private String buildSegment(String cubeName, long startDate, long endDate) throws Exception {
CubeSegment segment = cubeManager.appendSegments(cubeManager.getCube(cubeName), endDate);
DefaultChainedExecutable job = EngineFactory.createBatchCubingJob(segment, "TEST");
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d256a7f3/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java b/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
index 708bf32..fcfa67d 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
@@ -22,10 +22,13 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.TimeZone;
+import com.google.common.collect.Lists;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.util.JsonUtil;
@@ -55,6 +58,8 @@ public class CubeManagerTest extends LocalFileMetadataTestCase {
@Test
public void testBasics() throws Exception {
+
+
CubeInstance cube = CubeManager.getInstance(getTestConfig()).getCube("test_kylin_cube_without_slr_ready");
CubeDesc desc = cube.getDescriptor();
System.out.println(JsonUtil.writeValueAsIndentString(desc));
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d256a7f3/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
index 8e22615..32536e1 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_with_slr_left_join_desc.json
@@ -237,6 +237,6 @@
]
},
"notify_list": null,
- "engine_type": 2,
- "storage_type": 2
+ "engine_type": 0,
+ "storage_type": 0
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d256a7f3/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
index 08a132e..dfa62f7 100644
--- a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_without_slr_left_join_desc.json
@@ -15,7 +15,7 @@
},
{
"id": 1,
- "name": "CATEGORY",
+ "name": "CATEGORY"
"table": "DEFAULT.TEST_CATEGORY_GROUPINGS",
"column": null,
"derived": [
@@ -289,6 +289,6 @@
]
},
"notify_list": null,
- "engine_type": 0,
- "storage_type": 0
+ "engine_type": 2,
+ "storage_type": 2
}
\ No newline at end of file