http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Md5.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Md5.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Md5.java index f4cb7cb..209b3cc 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Md5.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Md5.java @@ -52,14 +52,15 @@ public class Md5 extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0)) { + return NullDatum.get(); + } try { - MessageDigest md = MessageDigest.getInstance("MD5"); - return DatumFactory.createText(new String(Hex.encodeHex(md.digest(datum.asByteArray())))); + MessageDigest md = MessageDigest.getInstance("MD5"); + return DatumFactory.createText(new String(Hex.encodeHex(md.digest(params.getBytes(0))))); } catch (NoSuchAlgorithmException e){ - return NullDatum.get(); + return NullDatum.get(); } } }
http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/OctetLength.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/OctetLength.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/OctetLength.java index cb5af15..6d8bffb 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/OctetLength.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/OctetLength.java @@ -51,11 +51,10 @@ public class OctetLength extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) { + if (params.isBlankOrNull(0)) { return NullDatum.get(); } - return DatumFactory.createInt4(datum.asByteArray().length); + return DatumFactory.createInt4(params.getBytes(0).length); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/QuoteIdent.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/QuoteIdent.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/QuoteIdent.java index aaef10d..c7bd7b8 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/QuoteIdent.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/QuoteIdent.java @@ -56,12 +56,10 @@ public class QuoteIdent extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - - if(datum instanceof NullDatum) { + if (params.isBlankOrNull(0)) { return NullDatum.get(); } - return DatumFactory.createText("\"" + datum.asChars() + "\""); + return DatumFactory.createText("\"" + params.getText(0) + "\""); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RTrim.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RTrim.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RTrim.java index 03b9c25..ff4e4ba 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RTrim.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RTrim.java @@ -66,13 +66,15 @@ public class RTrim extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0)) { + return NullDatum.get(); + } + String value = params.getText(0); if (!hasTrimCharacters) { - return DatumFactory.createText(StringUtils.stripEnd(datum.asChars(), null)); + return DatumFactory.createText(StringUtils.stripEnd(value, null)); } else { - return DatumFactory.createText(StringUtils.stripEnd(datum.asChars(), params.get(1).asChars())); + return DatumFactory.createText(StringUtils.stripEnd(value, params.getText(1))); } } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java index e3abeb5..6257c42 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java @@ -22,7 +22,6 @@ import com.google.gson.annotations.Expose; import org.apache.tajo.OverridableConf; import org.apache.tajo.catalog.Column; import org.apache.tajo.common.TajoDataTypes; -import org.apache.tajo.datum.BooleanDatum; import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.DatumFactory; import org.apache.tajo.datum.NullDatum; @@ -80,39 +79,33 @@ public class RegexpReplace extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum thisValue = params.get(0); - Datum thisPattern = params.get(1); - Datum thisReplacement = params.get(2); - boolean nullResult = isAlwaysNull - || thisValue instanceof NullDatum - || thisReplacement instanceof NullDatum - || thisPattern instanceof NullDatum; + if (isAlwaysNull || params.isBlankOrNull(0) || params.isBlankOrNull(1) || params.isBlankOrNull(2)) { + return NullDatum.get(); + } - Pattern thisCompiled; - if (!nullResult) { - if (compiled != null) { - thisCompiled = compiled; - } else { - thisCompiled = Pattern.compile(thisPattern.asChars()); + String value = params.getText(0); + String replacement = params.getText(2); - // if a regular expression pattern is a constant, - // it will be reused in every call - if (isPatternConstant) { - compiled = thisCompiled; - } - } + Pattern thisCompiled; + if (compiled != null) { + thisCompiled = compiled; + } else { + thisCompiled = Pattern.compile(params.getText(1)); - Matcher matcher = thisCompiled.matcher(thisValue.asChars()); - String replacement = thisReplacement.asChars(); - StringBuffer sb = new StringBuffer(); - while (matcher.find()) { - matcher.appendReplacement(sb, replacement); + // if a regular expression pattern is a constant, + // it will be reused in every call + if (isPatternConstant) { + compiled = thisCompiled; } - matcher.appendTail(sb); + } - return DatumFactory.createText(sb.toString()); - } else { - return NullDatum.get(); + Matcher matcher = thisCompiled.matcher(value); + StringBuffer sb = new StringBuffer(); + while (matcher.find()) { + matcher.appendReplacement(sb, replacement); } + matcher.appendTail(sb); + + return DatumFactory.createText(sb.toString()); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Repeat.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Repeat.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Repeat.java index c02ef66..339c4ea 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Repeat.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Repeat.java @@ -61,12 +61,10 @@ public class Repeat extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { + return NullDatum.get(); + } - Datum countDatum = params.get(1); - if(countDatum instanceof NullDatum) return NullDatum.get(); - - return DatumFactory.createText(repeat(datum.asChars(), countDatum.asInt4())); + return DatumFactory.createText(repeat(params.getText(0), params.getInt4(1))); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Reverse.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Reverse.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Reverse.java index bf67294..44d2c49 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Reverse.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Reverse.java @@ -50,9 +50,10 @@ public class Reverse extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0)) { + return NullDatum.get(); + } - return DatumFactory.createText(new StringBuffer(datum.asChars()).reverse().toString()); + return DatumFactory.createText(new StringBuffer(params.getText(0)).reverse().toString()); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Right.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Right.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Right.java index 68af423..0d0138a 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Right.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Right.java @@ -64,15 +64,13 @@ public class Right extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); - - Datum sizeDatum = params.get(1); - if(sizeDatum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { + return NullDatum.get(); + } - String data = datum.asChars(); + String data = params.getText(0); int length = data.length(); - int size = sizeDatum.asInt4(); + int size = params.getInt4(1); size = getSize(length, size); if (size == 0) { http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Rpad.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Rpad.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Rpad.java index 1509118..e50575d 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Rpad.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Rpad.java @@ -68,25 +68,26 @@ public class Rpad extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - Datum lengthDatum = params.get(1); - - if(datum instanceof NullDatum) return NullDatum.get(); - if(lengthDatum instanceof NullDatum) return NullDatum.get(); - - Datum fillText=NullDatum.get(); - if(hasFillCharacters) { - fillText=params.get(2); + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { + return NullDatum.get(); } - else { - fillText=DatumFactory.createText(" "); + + String fillText; + if (hasFillCharacters) { + fillText = params.getText(2); + } else { + fillText = " "; } - int templen = lengthDatum.asInt4() - datum.asChars().length(); + String input = params.getText(0); + int expected = params.getInt4(1); - if(templen<=0) - return DatumFactory.createText(datum.asChars().substring(0,lengthDatum.asInt4())); + int templen = expected - params.size(0); + + if (templen <= 0) { + return DatumFactory.createText(input.substring(0, expected)); + } - return DatumFactory.createText(StringUtils.rightPad(datum.asChars(), lengthDatum.asInt4(), fillText.asChars())); + return DatumFactory.createText(StringUtils.rightPad(input, expected, fillText)); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/SplitPart.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/SplitPart.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/SplitPart.java index 1c3be00..9604d18 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/SplitPart.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/SplitPart.java @@ -54,15 +54,12 @@ public class SplitPart extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum text = params.get(0); - Datum part = params.get(2); - - if (text.isNull() || part.isNull()) { + if (params.isBlankOrNull(0) || params.isBlankOrNull(2)) { return NullDatum.get(); } - String [] split = StringUtils.splitByWholeSeparatorPreserveAllTokens(text.asChars(), params.get(1).asChars(), -1); - int idx = params.get(2).asInt4() - 1; + String [] split = StringUtils.splitByWholeSeparatorPreserveAllTokens(params.getText(0), params.getText(1), -1); + int idx = params.getInt4(2) - 1; if (split.length > idx) { return DatumFactory.createText(split[idx]); } else { http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPos.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPos.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPos.java index 9c3e4f1..3ff1a3b 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPos.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPos.java @@ -52,19 +52,12 @@ public class StrPos extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum valueDatum = params.get(0); - if(valueDatum instanceof NullDatum) { + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { return NullDatum.get(); } - - Datum substringDatum = params.get(1); - if(substringDatum instanceof NullDatum) { - return NullDatum.get(); - } - - String value = valueDatum.asChars(); - String substring = substringDatum.asChars(); - if(substring.length() == 0) { + String value = params.getText(0); + String substring = params.getText(1); + if (substring.length() == 0) { return DatumFactory.createInt4(1); } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPosb.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPosb.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPosb.java index c332006..5bfcc50 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPosb.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/StrPosb.java @@ -54,19 +54,13 @@ public class StrPosb extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum valueDatum = params.get(0); - if(valueDatum instanceof NullDatum) { + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { return NullDatum.get(); } - Datum substringDatum = params.get(1); - if(substringDatum instanceof NullDatum) { - return NullDatum.get(); - } - - String value = valueDatum.asChars(); - String substring = substringDatum.asChars(); - if(substring.length() == 0) { + String value = params.getText(0); + String substring = params.getText(1); + if (substring.length() == 0) { return DatumFactory.createInt4(1); } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Substr.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Substr.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Substr.java index 7e4ddd6..4319853 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Substr.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Substr.java @@ -56,39 +56,28 @@ public class Substr extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum valueDatum = params.get(0); - Datum fromDatum = params.get(1); - Datum countDatum = params.size() > 2 ? params.get(2) : null; - - if(valueDatum instanceof NullDatum || fromDatum instanceof NullDatum || countDatum instanceof NullDatum) { + if (params.isBlankOrNull(0) || params.isBlankOrNull(1)) { return NullDatum.get(); } - - String value = valueDatum.asChars(); - int from = fromDatum.asInt4(); - int strLength = value.length(); - int count; - - if (countDatum == null) { - count = strLength; - } else { - count = (countDatum.asInt4() + from) - 1; + if (params.size() > 2 && params.isBlankOrNull(2)) { + return NullDatum.get(); } - if (count > strLength) { - count = strLength; - } + String value = params.getText(0); + int start = params.getInt4(1) - 1; + + int from = Math.max(0, start); + int length = params.size() > 2 ? params.getInt4(2) : -1; - if (from < 1) { - from = 0; - } else { - from --; + int to = value.length(); + if (length >= 0) { + to = Math.min(start + length, to); } - if (from >= count) { + if (from >= to) { return DatumFactory.createText(""); } - return DatumFactory.createText(value.substring(from, count)); + return DatumFactory.createText(value.substring(from, to)); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToBin.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToBin.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToBin.java index a9f85f3..a768e7f 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToBin.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToBin.java @@ -54,12 +54,11 @@ public class ToBin extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) { + if (params.isBlankOrNull(0)) { return NullDatum.get(); } - return DatumFactory.createText(Long.toBinaryString(datum.asInt8())); + return DatumFactory.createText(Long.toBinaryString(params.getInt8(0))); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToCharLong.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToCharLong.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToCharLong.java index 5fed940..7104835 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToCharLong.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToCharLong.java @@ -48,8 +48,8 @@ public class ToCharLong extends GeneralFunction { @Override public Datum eval(Tuple params) { if (df == null) { - df = new DecimalFormat(params.get(1).asChars()); + df = new DecimalFormat(params.getText(1)); } - return new TextDatum(df.format(params.get(0).asInt8())); + return new TextDatum(df.format(params.getInt8(0))); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToHex.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToHex.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToHex.java index 2e20008..d410b9b 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToHex.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/ToHex.java @@ -65,12 +65,11 @@ public class ToHex extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) { + if (params.isBlankOrNull(0)) { return NullDatum.get(); } - String ret = new String(Hex.encodeHex(datum.asByteArray())); + String ret = new String(Hex.encodeHex(params.getBytes(0))); return DatumFactory.createText(trimZero(ret)); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Upper.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Upper.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Upper.java index fc6ff3a..7525d95 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Upper.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/Upper.java @@ -50,9 +50,10 @@ public class Upper extends GeneralFunction { @Override public Datum eval(Tuple params) { - Datum datum = params.get(0); - if(datum instanceof NullDatum) return NullDatum.get(); + if (params.isBlankOrNull(0)) { + return NullDatum.get(); + } - return DatumFactory.createText(datum.asChars().toUpperCase()); + return DatumFactory.createText(params.getText(0).toUpperCase()); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java index ba3b3b6..6a667ec 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java @@ -38,10 +38,10 @@ public abstract class FirstValue extends WindowAggFunc<Datum> { @Override public void eval(FunctionContext ctx, Tuple params) { FirstValueContext firstValueCtx = (FirstValueContext)ctx; - if(firstValueCtx.isSet == false) { + if(!firstValueCtx.isSet) { firstValueCtx.isSet = true; - if (params.get(0).isNotNull()) { - firstValueCtx.first = params.get(0); + if (!params.isBlankOrNull(0)) { + firstValueCtx.first = params.asDatum(0); } } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java index 5107297..eb7e8b7 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java @@ -45,20 +45,20 @@ public abstract class Lag extends WindowAggFunc<Datum> { if (params.size() == 1) { lagNum = 1; } else { - lagNum = params.get(1).asInt4(); + lagNum = params.getInt4(1); } lagCtx.lagBuffer = new CircularFifoBuffer(lagNum+1); } - if (params.get(0).isNotNull()) { - lagCtx.lagBuffer.add(params.get(0)); + if (!params.isBlankOrNull(0)) { + lagCtx.lagBuffer.add(params.asDatum(0)); } else { lagCtx.lagBuffer.add(NullDatum.get()); } if (lagCtx.defaultDatum == null) { if (params.size() == 3) { - lagCtx.defaultDatum = params.get(2); + lagCtx.defaultDatum = params.asDatum(2); } else { lagCtx.defaultDatum = NullDatum.get(); } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java index e469c83..9cb95f7 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java @@ -48,7 +48,7 @@ public final class Rank extends WindowAggFunc { public static boolean checkIfDistinctValue(RankContext context, Tuple params) { for (int i = 0; i < context.latest.length; i++) { - if (!context.latest[i].equalsTo(params.get(i)).isTrue()) { + if (!context.latest[i].equalsTo(params.asDatum(i)).isTrue()) { return true; } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java index 38aa928..c505b77 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java @@ -57,81 +57,83 @@ public abstract class RangePartitionAlgorithm { * It computes the value cardinality of a tuple range. * * @param dataType - * @param start - * @param end + * @param range + * @param i * @return */ - public static BigInteger computeCardinality(DataType dataType, Datum start, Datum end, + public static BigInteger computeCardinality(DataType dataType, TupleRange range, int i, boolean inclusive, boolean isAscending) { BigInteger columnCard; + Tuple start = range.getStart(); + Tuple end = range.getEnd(); switch (dataType.getType()) { case BOOLEAN: columnCard = BigInteger.valueOf(2); break; case CHAR: if (isAscending) { - columnCard = BigInteger.valueOf((int)end.asChar() - (int)start.asChar()); + columnCard = BigInteger.valueOf((int)end.getChar(i) - (int)start.getChar(i)); } else { - columnCard = BigInteger.valueOf(start.asChar() - end.asChar()); + columnCard = BigInteger.valueOf(start.getChar(i) - end.getChar(i)); } break; case BIT: if (isAscending) { - columnCard = BigInteger.valueOf(end.asByte() - start.asByte()); + columnCard = BigInteger.valueOf(end.getByte(i) - start.getByte(i)); } else { - columnCard = BigInteger.valueOf(start.asByte() - end.asByte()); + columnCard = BigInteger.valueOf(start.getByte(i) - end.getByte(i)); } break; case INT2: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt2() - start.asInt2()); + columnCard = BigInteger.valueOf(end.getInt2(i) - start.getInt2(i)); } else { - columnCard = BigInteger.valueOf(start.asInt2() - end.asInt2()); + columnCard = BigInteger.valueOf(start.getInt2(i) - end.getInt2(i)); } break; case INT4: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt4() - start.asInt4()); + columnCard = BigInteger.valueOf(end.getInt4(i) - start.getInt4(i)); } else { - columnCard = BigInteger.valueOf(start.asInt4() - end.asInt4()); + columnCard = BigInteger.valueOf(start.getInt4(i) - end.getInt4(i)); } break; case INT8: case TIME: case TIMESTAMP: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt8() - start.asInt8()); + columnCard = BigInteger.valueOf(end.getInt8(i) - start.getInt8(i)); } else { - columnCard = BigInteger.valueOf(start.asInt8() - end.asInt8()); + columnCard = BigInteger.valueOf(start.getInt8(i) - end.getInt8(i)); } break; case FLOAT4: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt4() - start.asInt4()); + columnCard = BigInteger.valueOf(end.getInt4(i) - start.getInt4(i)); } else { - columnCard = BigInteger.valueOf(start.asInt4() - end.asInt4()); + columnCard = BigInteger.valueOf(start.getInt4(i) - end.getInt4(i)); } break; case FLOAT8: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt8() - start.asInt8()); + columnCard = BigInteger.valueOf(end.getInt8(i) - start.getInt8(i)); } else { - columnCard = BigInteger.valueOf(start.asInt8() - end.asInt8()); + columnCard = BigInteger.valueOf(start.getInt8(i) - end.getInt8(i)); } break; case TEXT: { - boolean isPureAscii = StringUtils.isPureAscii(start.asChars()) && StringUtils.isPureAscii(end.asChars()); + boolean isPureAscii = StringUtils.isPureAscii(start.getText(i)) && StringUtils.isPureAscii(end.getText(i)); if (isPureAscii) { byte[] a; byte[] b; if (isAscending) { - a = start.asByteArray(); - b = end.asByteArray(); + a = start.getBytes(i); + b = end.getBytes(i); } else { - b = start.asByteArray(); - a = end.asByteArray(); + b = start.getBytes(i); + a = end.getBytes(i); } byte [][] padded = BytesUtils.padBytes(a, b); @@ -148,11 +150,11 @@ public abstract class RangePartitionAlgorithm { char [] b; if (isAscending) { - a = start.asUnicodeChars(); - b = end.asUnicodeChars(); + a = start.getUnicodeChars(i); + b = end.getUnicodeChars(i); } else { - b = start.asUnicodeChars(); - a = end.asUnicodeChars(); + b = start.getUnicodeChars(i); + a = end.getUnicodeChars(i); } BigInteger startBI = UniformRangePartition.charsToBigInteger(a); @@ -165,16 +167,16 @@ public abstract class RangePartitionAlgorithm { } case DATE: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt4() - start.asInt4()); + columnCard = BigInteger.valueOf(end.getInt4(i) - start.getInt4(i)); } else { - columnCard = BigInteger.valueOf(start.asInt4() - end.asInt4()); + columnCard = BigInteger.valueOf(start.getInt4(i) - end.getInt4(i)); } break; case INET4: if (isAscending) { - columnCard = BigInteger.valueOf(end.asInt4() - start.asInt4()); + columnCard = BigInteger.valueOf(end.getInt4(i) - start.getInt4(i)); } else { - columnCard = BigInteger.valueOf(start.asInt4() - end.asInt4()); + columnCard = BigInteger.valueOf(start.getInt4(i) - end.getInt4(i)); } break; default: @@ -189,13 +191,11 @@ public abstract class RangePartitionAlgorithm { * @return */ public static BigInteger computeCardinalityForAllColumns(SortSpec[] sortSpecs, TupleRange range, boolean inclusive) { - Tuple start = range.getStart(); - Tuple end = range.getEnd(); BigInteger cardinality = BigInteger.ONE; BigInteger columnCard; for (int i = 0; i < sortSpecs.length; i++) { - columnCard = computeCardinality(sortSpecs[i].getSortKey().getDataType(), start.get(i), end.get(i), inclusive, + columnCard = computeCardinality(sortSpecs[i].getSortKey().getDataType(), range, i, inclusive, sortSpecs[i].isAscending()); if (BigInteger.ZERO.compareTo(columnCard) < 0) { http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/UniformRangePartition.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/UniformRangePartition.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/UniformRangePartition.java index 551a9d0..7c26857 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/UniformRangePartition.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/UniformRangePartition.java @@ -24,7 +24,6 @@ import com.sun.tools.javac.util.Convert; import org.apache.tajo.catalog.Column; import org.apache.tajo.catalog.SortSpec; import org.apache.tajo.common.TajoDataTypes; -import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.DatumFactory; import org.apache.tajo.datum.NullDatum; import org.apache.tajo.datum.TextDatum; @@ -70,21 +69,18 @@ public class UniformRangePartition extends RangePartitionAlgorithm { // filling pure ascii flags isPureAscii = new boolean[sortSpecs.length]; for (int i = 0; i < sortSpecs.length; i++) { - Datum startValue = entireRange.getStart().get(i); - Datum endValue = entireRange.getEnd().get(i); - isPureAscii[i] = StringUtils.isPureAscii(startValue.asChars()) && StringUtils.isPureAscii(endValue.asChars()); - beginNulls[i] = startValue.isNull(); - endNulls[i] = endValue.isNull(); + String startValue = entireRange.getStart().getText(i); + String endValue = entireRange.getEnd().getText(i); + isPureAscii[i] = StringUtils.isPureAscii(startValue) && StringUtils.isPureAscii(endValue); + beginNulls[i] = entireRange.getStart().isBlankOrNull(i); + endNulls[i] = entireRange.getEnd().isBlankOrNull(i); } colCards = new BigInteger[sortSpecs.length]; normalize(sortSpecs, this.mergedRange); for (int i = 0; i < sortSpecs.length; i++) { - Datum startValue = entireRange.getStart().get(i); - Datum endValue = entireRange.getEnd().get(i); - - colCards[i] = computeCardinality(sortSpecs[i].getSortKey().getDataType(), startValue, endValue, + colCards[i] = computeCardinality(sortSpecs[i].getSortKey().getDataType(), entireRange, i, inclusive, sortSpecs[i].isAscending()); } @@ -177,13 +173,13 @@ public class UniformRangePartition extends RangePartitionAlgorithm { if (isPureAscii[i]) { byte[] startBytes; byte[] endBytes; - if (range.getStart().isNull(i)) { + if (range.getStart().isBlankOrNull(i)) { startBytes = BigInteger.ZERO.toByteArray(); } else { startBytes = range.getStart().getBytes(i); } - if (range.getEnd().isNull(i)) { + if (range.getEnd().isBlankOrNull(i)) { endBytes = BigInteger.ZERO.toByteArray(); } else { endBytes = range.getEnd().getBytes(i); @@ -196,13 +192,13 @@ public class UniformRangePartition extends RangePartitionAlgorithm { } else { char[] startChars; char[] endChars; - if (range.getStart().isNull(i)) { + if (range.getStart().isBlankOrNull(i)) { startChars = new char[] {0}; } else { startChars = range.getStart().getUnicodeChars(i); } - if (range.getEnd().isNull(i)) { + if (range.getEnd().isBlankOrNull(i)) { endChars = new char[] {0}; } else { endChars = range.getEnd().getUnicodeChars(i); @@ -220,12 +216,12 @@ public class UniformRangePartition extends RangePartitionAlgorithm { * Check whether an overflow occurs or not. * * @param colId The column id to be checked - * @param last + * @param tuple * @param inc * @param sortSpecs * @return */ - public boolean isOverflow(int colId, Datum last, BigInteger inc, SortSpec [] sortSpecs) { + public boolean isOverflow(int colId, Tuple tuple, int i, BigInteger inc, SortSpec [] sortSpecs) { Column column = sortSpecs[colId].getSortKey(); BigDecimal incDecimal = new BigDecimal(inc); BigDecimal candidate; @@ -234,74 +230,74 @@ public class UniformRangePartition extends RangePartitionAlgorithm { switch (column.getDataType().getType()) { case BIT: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asByte())); - return new BigDecimal(mergedRange.getEnd().get(colId).asByte()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getByte(i))); + return new BigDecimal(mergedRange.getEnd().getByte(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asByte()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asByte())) < 0; + candidate = new BigDecimal(tuple.getByte(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getByte(colId))) < 0; } } case CHAR: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal((int)last.asChar())); - return new BigDecimal((int) mergedRange.getEnd().get(colId).asChar()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal((int)tuple.getChar(i))); + return new BigDecimal((int) mergedRange.getEnd().getChar(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal((int)last.asChar()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal((int) mergedRange.getEnd().get(colId).asChar())) < 0; + candidate = new BigDecimal((int)tuple.getChar(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal((int) mergedRange.getEnd().getChar(colId))) < 0; } } case INT2: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asInt2())); - return new BigDecimal(mergedRange.getEnd().get(colId).asInt2()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getInt2(i))); + return new BigDecimal(mergedRange.getEnd().getInt2(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asInt2()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asInt2())) < 0; + candidate = new BigDecimal(tuple.getInt2(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getInt2(colId))) < 0; } } case DATE: case INT4: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asInt4())); - return new BigDecimal(mergedRange.getEnd().get(colId).asInt4()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getInt4(i))); + return new BigDecimal(mergedRange.getEnd().getInt4(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asInt4()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asInt4())) < 0; + candidate = new BigDecimal(tuple.getInt4(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getInt4(colId))) < 0; } } case TIME: case TIMESTAMP: case INT8: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asInt8())); - return new BigDecimal(mergedRange.getEnd().get(colId).asInt8()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getInt8(i))); + return new BigDecimal(mergedRange.getEnd().getInt8(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asInt8()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asInt8())) < 0; + candidate = new BigDecimal(tuple.getInt8(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getInt8(colId))) < 0; } } case FLOAT4: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asFloat4())); - return new BigDecimal(mergedRange.getEnd().get(colId).asFloat4()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getFloat4(i))); + return new BigDecimal(mergedRange.getEnd().getFloat4(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asFloat4()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asFloat4())) < 0; + candidate = new BigDecimal(tuple.getFloat4(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getFloat4(colId))) < 0; } } case FLOAT8: { if (sortSpecs[colId].isAscending()) { - candidate = incDecimal.add(new BigDecimal(last.asFloat8())); - return new BigDecimal(mergedRange.getEnd().get(colId).asFloat8()).compareTo(candidate) < 0; + candidate = incDecimal.add(new BigDecimal(tuple.getFloat8(i))); + return new BigDecimal(mergedRange.getEnd().getFloat8(colId)).compareTo(candidate) < 0; } else { - candidate = new BigDecimal(last.asFloat8()).subtract(incDecimal); - return candidate.compareTo(new BigDecimal(mergedRange.getEnd().get(colId).asFloat8())) < 0; + candidate = new BigDecimal(tuple.getFloat8(i)).subtract(incDecimal); + return candidate.compareTo(new BigDecimal(mergedRange.getEnd().getFloat8(colId))) < 0; } } case TEXT: { if (isPureAscii[colId]) { - byte[] lastBytes = last.asByteArray(); + byte[] lastBytes = tuple.getBytes(i); byte[] endBytes = mergedRange.getEnd().getBytes(colId); Preconditions.checkState(lastBytes.length == endBytes.length); @@ -314,7 +310,7 @@ public class UniformRangePartition extends RangePartitionAlgorithm { return candidate.compareTo(new BigDecimal(new BigInteger(endBytes))) < 0; } } else { - char[] lastChars = last.asUnicodeChars(); + char[] lastChars = tuple.getUnicodeChars(i); char[] endChars = mergedRange.getEnd().getUnicodeChars(colId); Preconditions.checkState(lastChars.length == endChars.length); @@ -335,45 +331,45 @@ public class UniformRangePartition extends RangePartitionAlgorithm { int candidateIntVal; byte[] candidateBytesVal = new byte[4]; if (sortSpecs[colId].isAscending()) { - candidateIntVal = incDecimal.intValue() + last.asInt4(); - if (candidateIntVal - incDecimal.intValue() != last.asInt4()) { + candidateIntVal = incDecimal.intValue() + tuple.getInt4(i); + if (candidateIntVal - incDecimal.intValue() != tuple.getInt4(i)) { return true; } Bytes.putInt(candidateBytesVal, 0, candidateIntVal); - return Bytes.compareTo(mergedRange.getEnd().get(colId).asByteArray(), candidateBytesVal) < 0; + return Bytes.compareTo(mergedRange.getEnd().getBytes(colId), candidateBytesVal) < 0; } else { - candidateIntVal = last.asInt4() - incDecimal.intValue(); - if (candidateIntVal + incDecimal.intValue() != last.asInt4()) { + candidateIntVal = tuple.getInt4(i) - incDecimal.intValue(); + if (candidateIntVal + incDecimal.intValue() != tuple.getInt4(i)) { return true; } Bytes.putInt(candidateBytesVal, 0, candidateIntVal); - return Bytes.compareTo(candidateBytesVal, mergedRange.getEnd().get(colId).asByteArray()) < 0; + return Bytes.compareTo(candidateBytesVal, mergedRange.getEnd().getBytes(colId)) < 0; } } } return overflow; } - public long incrementAndGetReminder(int colId, Datum last, long inc) { + private long incrementAndGetReminder(int colId, Tuple last, long inc) { Column column = sortSpecs[colId].getSortKey(); long reminder = 0; switch (column.getDataType().getType()) { case BIT: { - long candidate = last.asByte() + inc; - byte end = mergedRange.getEnd().get(colId).asByte(); + long candidate = last.getByte(colId) + inc; + byte end = mergedRange.getEnd().getByte(colId); reminder = candidate - end; break; } case CHAR: { - long candidate = last.asChar() + inc; - char end = mergedRange.getEnd().get(colId).asChar(); + long candidate = last.getChar(colId) + inc; + char end = mergedRange.getEnd().getChar(colId); reminder = candidate - end; break; } case DATE: case INT4: { - int candidate = (int) (last.asInt4() + inc); - int end = mergedRange.getEnd().get(colId).asInt4(); + int candidate = (int) (last.getInt4(colId) + inc); + int end = mergedRange.getEnd().getInt4(colId); reminder = candidate - end; break; } @@ -381,26 +377,26 @@ public class UniformRangePartition extends RangePartitionAlgorithm { case TIMESTAMP: case INT8: case INET4: { - long candidate = last.asInt8() + inc; - long end = mergedRange.getEnd().get(colId).asInt8(); + long candidate = last.getInt8(colId) + inc; + long end = mergedRange.getEnd().getInt8(colId); reminder = candidate - end; break; } case FLOAT4: { - float candidate = last.asFloat4() + inc; - float end = mergedRange.getEnd().get(colId).asFloat4(); + float candidate = last.getFloat4(colId) + inc; + float end = mergedRange.getEnd().getFloat4(colId); reminder = (long) (candidate - end); break; } case FLOAT8: { - double candidate = last.asFloat8() + inc; - double end = mergedRange.getEnd().get(colId).asFloat8(); + double candidate = last.getFloat8(colId) + inc; + double end = mergedRange.getEnd().getFloat8(colId); reminder = (long) Math.ceil(candidate - end); break; } case TEXT: { - byte [] lastBytes = last.asByteArray(); - byte [] endBytes = mergedRange.getEnd().get(colId).asByteArray(); + byte [] lastBytes = last.getBytes(colId); + byte [] endBytes = mergedRange.getEnd().getBytes(colId); Preconditions.checkState(lastBytes.length == endBytes.length); @@ -447,12 +443,12 @@ public class UniformRangePartition extends RangePartitionAlgorithm { int finalId = baseDigit; incs[finalId] = value; for (int i = finalId; i >= 0; i--) { - if (isOverflow(i, last.get(i), incs[i], sortSpecs)) { + if (isOverflow(i, last, i, incs[i], sortSpecs)) { if (i == 0) { throw new RangeOverflowException(mergedRange, last, incs[i].longValue(), sortSpecs[i].isAscending()); } // increment some volume of the serialized one-dimension key space - long rem = incrementAndGetReminder(i, last.get(i), value.longValue()); + long rem = incrementAndGetReminder(i, last, value.longValue()); incs[i] = BigInteger.valueOf(rem); incs[i - 1] = incs[i-1].add(BigInteger.ONE); overflowFlag[i] = true; @@ -470,107 +466,107 @@ public class UniformRangePartition extends RangePartitionAlgorithm { } } - Tuple end = new VTuple(sortSpecs.length); + VTuple end = new VTuple(sortSpecs.length); Column column; for (int i = 0; i < last.size(); i++) { column = sortSpecs[i].getSortKey(); switch (column.getDataType().getType()) { case CHAR: if (overflowFlag[i]) { - end.put(i, DatumFactory.createChar((char) (mergedRange.getStart().get(i).asChar() + incs[i].longValue()))); + end.put(i, DatumFactory.createChar((char) (mergedRange.getStart().getChar(i) + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createChar((char) (last.get(i).asChar() + incs[i].longValue()))); + end.put(i, DatumFactory.createChar((char) (last.getChar(i) + incs[i].longValue()))); } else { - end.put(i, DatumFactory.createChar((char) (last.get(i).asChar() - incs[i].longValue()))); + end.put(i, DatumFactory.createChar((char) (last.getChar(i) - incs[i].longValue()))); } } break; case BIT: if (overflowFlag[i]) { end.put(i, DatumFactory.createBit( - (byte) (mergedRange.getStart().get(i).asByte() + incs[i].longValue()))); + (byte) (mergedRange.getStart().getByte(i) + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createBit((byte) (last.get(i).asByte() + incs[i].longValue()))); + end.put(i, DatumFactory.createBit((byte) (last.getByte(i) + incs[i].longValue()))); } else { - end.put(i, DatumFactory.createBit((byte) (last.get(i).asByte() - incs[i].longValue()))); + end.put(i, DatumFactory.createBit((byte) (last.getByte(i) - incs[i].longValue()))); } } break; case INT2: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt2( - (short) (mergedRange.getStart().get(i).asInt2() + incs[i].longValue()))); + (short) (mergedRange.getStart().getInt2(i) + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createInt2((short) (last.get(i).asInt2() + incs[i].longValue()))); + end.put(i, DatumFactory.createInt2((short) (last.getInt2(i) + incs[i].longValue()))); } else { - end.put(i, DatumFactory.createInt2((short) (last.get(i).asInt2() - incs[i].longValue()))); + end.put(i, DatumFactory.createInt2((short) (last.getInt2(i) - incs[i].longValue()))); } } break; case INT4: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt4( - (int) (mergedRange.getStart().get(i).asInt4() + incs[i].longValue()))); + (int) (mergedRange.getStart().getInt4(i) + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() + incs[i].longValue()))); + end.put(i, DatumFactory.createInt4((int) (last.getInt4(i) + incs[i].longValue()))); } else { - end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() - incs[i].longValue()))); + end.put(i, DatumFactory.createInt4((int) (last.getInt4(i) - incs[i].longValue()))); } } break; case INT8: if (overflowFlag[i]) { end.put(i, DatumFactory.createInt8( - mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); + mergedRange.getStart().getInt8(i) + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createInt8(last.get(i).asInt8() + incs[i].longValue())); + end.put(i, DatumFactory.createInt8(last.getInt8(i) + incs[i].longValue())); } else { - end.put(i, DatumFactory.createInt8(last.get(i).asInt8() - incs[i].longValue())); + end.put(i, DatumFactory.createInt8(last.getInt8(i) - incs[i].longValue())); } } break; case FLOAT4: if (overflowFlag[i]) { end.put(i, DatumFactory.createFloat4( - mergedRange.getStart().get(i).asFloat4() + incs[i].longValue())); + mergedRange.getStart().getFloat4(i) + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createFloat4(last.get(i).asFloat4() + incs[i].longValue())); + end.put(i, DatumFactory.createFloat4(last.getFloat4(i) + incs[i].longValue())); } else { - end.put(i, DatumFactory.createFloat4(last.get(i).asFloat4() - incs[i].longValue())); + end.put(i, DatumFactory.createFloat4(last.getFloat4(i) - incs[i].longValue())); } } break; case FLOAT8: if (overflowFlag[i]) { end.put(i, DatumFactory.createFloat8( - mergedRange.getStart().get(i).asFloat8() + incs[i].longValue())); + mergedRange.getStart().getFloat8(i) + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createFloat8(last.get(i).asFloat8() + incs[i].longValue())); + end.put(i, DatumFactory.createFloat8(last.getFloat8(i) + incs[i].longValue())); } else { - end.put(i, DatumFactory.createFloat8(last.get(i).asFloat8() - incs[i].longValue())); + end.put(i, DatumFactory.createFloat8(last.getFloat8(i) - incs[i].longValue())); } } break; case TEXT: if (overflowFlag[i]) { - end.put(i, DatumFactory.createText(((char) (mergedRange.getStart().get(i).asChars().charAt(0) + end.put(i, DatumFactory.createText(((char) (mergedRange.getStart().getText(i).charAt(0) + incs[i].longValue())) + "")); } else { BigInteger lastBigInt; - if (last.isNull(i)) { + if (last.isBlankOrNull(i)) { lastBigInt = BigInteger.valueOf(0); end.put(i, DatumFactory.createText(lastBigInt.add(incs[i]).toByteArray())); } else { if (isPureAscii[i]) { - lastBigInt = new BigInteger(last.get(i).asByteArray()); + lastBigInt = new BigInteger(last.getBytes(i)); if (sortSpecs[i].isAscending()) { end.put(i, DatumFactory.createText(lastBigInt.add(incs[i]).toByteArray())); } else { @@ -633,52 +629,52 @@ public class UniformRangePartition extends RangePartitionAlgorithm { break; case DATE: if (overflowFlag[i]) { - end.put(i, DatumFactory.createDate((int) (mergedRange.getStart().get(i).asInt4() + incs[i].longValue()))); + end.put(i, DatumFactory.createDate((int) (mergedRange.getStart().getInt4(i) + incs[i].longValue()))); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createDate((int) (last.get(i).asInt4() + incs[i].longValue()))); + end.put(i, DatumFactory.createDate((int) (last.getInt4(i) + incs[i].longValue()))); } else { - end.put(i, DatumFactory.createDate((int) (last.get(i).asInt4() - incs[i].longValue()))); + end.put(i, DatumFactory.createDate((int) (last.getInt4(i) - incs[i].longValue()))); } } break; case TIME: if (overflowFlag[i]) { - end.put(i, DatumFactory.createTime(mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); + end.put(i, DatumFactory.createTime(mergedRange.getStart().getInt8(i) + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createTime(last.get(i).asInt8() + incs[i].longValue())); + end.put(i, DatumFactory.createTime(last.getInt8(i) + incs[i].longValue())); } else { - end.put(i, DatumFactory.createTime(last.get(i).asInt8() - incs[i].longValue())); + end.put(i, DatumFactory.createTime(last.getInt8(i) - incs[i].longValue())); } } break; case TIMESTAMP: if (overflowFlag[i]) { end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis( - mergedRange.getStart().get(i).asInt8() + incs[i].longValue())); + mergedRange.getStart().getInt8(i) + incs[i].longValue())); } else { if (sortSpecs[i].isAscending()) { - end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.get(i).asInt8() + incs[i].longValue())); + end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.getInt8(i) + incs[i].longValue())); } else { - end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.get(i).asInt8() - incs[i].longValue())); + end.put(i, DatumFactory.createTimestmpDatumWithJavaMillis(last.getInt8(i) - incs[i].longValue())); } } break; case INET4: byte[] ipBytes; if (overflowFlag[i]) { - ipBytes = mergedRange.getStart().get(i).asByteArray(); + ipBytes = mergedRange.getStart().getBytes(i); assert ipBytes.length == 4; end.put(i, DatumFactory.createInet4(ipBytes)); } else { if (sortSpecs[i].isAscending()) { - int lastVal = last.get(i).asInt4() + incs[i].intValue(); + int lastVal = last.getInt4(i) + incs[i].intValue(); ipBytes = new byte[4]; Bytes.putInt(ipBytes, 0, lastVal); end.put(i, DatumFactory.createInet4(ipBytes)); } else { - int lastVal = last.get(i).asInt4() - incs[i].intValue(); + int lastVal = last.getInt4(i) - incs[i].intValue(); ipBytes = new byte[4]; Bytes.putInt(ipBytes, 0, lastVal); end.put(i, DatumFactory.createInet4(ipBytes)); http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java index 4612d45..bc6975a 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java @@ -80,8 +80,7 @@ public class BSTIndexScanExec extends PhysicalExec { public Tuple next() throws IOException { if(initialize) { //TODO : more complicated condition - Tuple key = new VTuple(datum.length); - key.put(datum); + Tuple key = new VTuple(datum); long offset = reader.find(key); if (offset == -1) { reader.close(); http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonHashJoinExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonHashJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonHashJoinExec.java index 588f0fc..a018fe1 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonHashJoinExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonHashJoinExec.java @@ -122,7 +122,7 @@ public abstract class CommonHashJoinExec<T> extends CommonJoinExec { while (!context.isStopped() && (tuple = rightChild.next()) != null) { Tuple keyTuple = new VTuple(joinKeyPairs.size()); for (int i = 0; i < rightKeyList.length; i++) { - keyTuple.put(i, tuple.get(rightKeyList[i])); + keyTuple.put(i, tuple.asDatum(rightKeyList[i])); } /* @@ -154,7 +154,7 @@ public abstract class CommonHashJoinExec<T> extends CommonJoinExec { protected Tuple toKey(final Tuple outerTuple) { for (int i = 0; i < leftKeyList.length; i++) { - keyTuple.put(i, outerTuple.get(leftKeyList[i])); + keyTuple.put(i, outerTuple.asDatum(leftKeyList[i])); } return keyTuple; } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ComparableVector.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ComparableVector.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ComparableVector.java index 9940608..2d836f4 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ComparableVector.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ComparableVector.java @@ -121,7 +121,7 @@ public class ComparableVector { } protected final void set(int index, Tuple tuple, int field) { - if (tuple.isNull(field)) { + if (tuple.isBlankOrNull(field)) { nulls.set(index); return; } @@ -196,7 +196,7 @@ public class ComparableVector { public void set(Tuple tuple) { for (int i = 0; i < keyTypes.length; i++) { final int field = keyIndex[i]; - if (tuple.isNull(field)) { + if (tuple.isBlankOrNull(field)) { keys[i] = null; continue; } @@ -216,7 +216,7 @@ public class ComparableVector { case TEXT: case CHAR: case BLOB: keys[i] = tuple.getBytes(field); break; - case DATUM: keys[i] = tuple.get(field); break; + case DATUM: keys[i] = tuple.asDatum(field); break; default: throw new IllegalArgumentException(); } @@ -252,7 +252,7 @@ public class ComparableVector { for (int i = 0; i < keys.length; i++) { final int field = keyIndex[i]; final boolean n1 = keys[i] == null; - final boolean n2 = tuple.isNull(field); + final boolean n2 = tuple.isBlankOrNull(field); if (n1 && n2) { continue; } @@ -275,7 +275,7 @@ public class ComparableVector { case TEXT: case CHAR: case BLOB: if (!Arrays.equals((byte[])keys[i], tuple.getBytes(field))) return false; continue; - case DATUM: if (!keys[i].equals(tuple.get(field))) return false; continue; + case DATUM: if (!keys[i].equals(tuple.asDatum(field))) return false; continue; } } return true; http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java index 94429a0..7784817 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java @@ -175,7 +175,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { while(!context.isStopped() && (tuple = child.next()) != null) { Tuple groupingKey = new VTuple(groupingKeyIndexes.length); for (int i = 0; i < groupingKeyIndexes.length; i++) { - groupingKey.put(i, tuple.get(groupingKeyIndexes[i])); + groupingKey.put(i, tuple.asDatum(groupingKeyIndexes[i])); } for (int i = 0; i < distinctAggregators.length; i++) { distinctAggregators[i].compute(groupingKey, tuple); @@ -360,7 +360,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { public void compute(Tuple groupingKey, Tuple tuple) throws IOException { Tuple distinctKeyTuple = new VTuple(distinctKeyIndexes.length); for (int i = 0; i < distinctKeyIndexes.length; i++) { - distinctKeyTuple.put(i, tuple.get(distinctKeyIndexes[i])); + distinctKeyTuple.put(i, tuple.asDatum(distinctKeyIndexes[i])); } Set<Tuple> distinctEntry = distinctAggrDatas.get(groupingKey); @@ -415,7 +415,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { } // node sequence, groupingKeys, 1'st distinctKeys, 2'st distinctKeys, ... // If n'st == this.nodeSequence set with real data, otherwise set with NullDatum - Tuple tuple = new VTuple(resultTupleLength); + VTuple tuple = new VTuple(resultTupleLength); int tupleIndex = 0; tuple.put(tupleIndex++, nodeSequenceDatum); @@ -423,7 +423,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { Tuple groupingKeyTuple = currentGroupingTuples.getKey(); int groupingKeyLength = groupingKeyTuple.size(); for (int i = 0; i < groupingKeyLength; i++, tupleIndex++) { - tuple.put(tupleIndex, groupingKeyTuple.get(i)); + tuple.put(tupleIndex, groupingKeyTuple.asDatum(i)); } // merge distinctKey @@ -432,13 +432,13 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { Tuple distinctKeyTuple = distinctKeyIterator.next(); int distinctKeyLength = distinctKeyTuple.size(); for (int j = 0; j < distinctKeyLength; j++, tupleIndex++) { - tuple.put(tupleIndex, distinctKeyTuple.get(j)); + tuple.put(tupleIndex, distinctKeyTuple.asDatum(j)); } } else { Tuple dummyTuple = distinctAggregators[i].getDummyTuple(); int dummyTupleSize = dummyTuple.size(); for (int j = 0; j < dummyTupleSize; j++, tupleIndex++) { - tuple.put(tupleIndex, dummyTuple.get(j)); + tuple.put(tupleIndex, dummyTuple.asDatum(j)); } } } @@ -457,7 +457,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec { } int tupleSize = nonDistinctTuple.size(); for (int j = 0; j < tupleSize; j++, tupleIndex++) { - tuple.put(tupleIndex, nonDistinctTuple.get(j)); + tuple.put(tupleIndex, nonDistinctTuple.asDatum(j)); } } return tuple; http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java index 0f25d6c..c8a6588 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java @@ -236,10 +236,10 @@ public class DistinctGroupbyHashAggregationExec extends UnaryPhysicalExec { // set group key tuple // Because each hashAggregator has different number of tuples, // sometimes getting group key from each hashAggregator will be null value. - mergedTuple.put(mergeTupleIndex, distinctGroupingKey.get(mergeTupleIndex)); + mergedTuple.put(mergeTupleIndex, distinctGroupingKey.asDatum(mergeTupleIndex)); } else { if (tuples[i] != null) { - mergedTuple.put(mergeTupleIndex, tuples[i].get(j)); + mergedTuple.put(mergeTupleIndex, tuples[i].asDatum(j)); } else { mergedTuple.put(mergeTupleIndex, NullDatum.get()); } @@ -388,12 +388,12 @@ public class DistinctGroupbyHashAggregationExec extends UnaryPhysicalExec { public void compute(Tuple tuple) throws IOException { Tuple outerKeyTuple = new VTuple(distinctGroupingKeyIds.length); for (int i = 0; i < distinctGroupingKeyIds.length; i++) { - outerKeyTuple.put(i, tuple.get(distinctGroupingKeyIds[i])); + outerKeyTuple.put(i, tuple.asDatum(distinctGroupingKeyIds[i])); } Tuple keyTuple = new VTuple(groupingKeyIds.length); for (int i = 0; i < groupingKeyIds.length; i++) { - keyTuple.put(i, tuple.get(groupingKeyIds[i])); + keyTuple.put(i, tuple.asDatum(groupingKeyIds[i])); } Map<Tuple, FunctionContext[]> distinctEntry = hashTable.get(outerKeyTuple); @@ -428,7 +428,7 @@ public class DistinctGroupbyHashAggregationExec extends UnaryPhysicalExec { Tuple groupbyKey = entry.getKey(); int index = 0; for (; index < groupbyKey.size(); index++) { - tuple.put(index, groupbyKey.get(index)); + tuple.put(index, groupbyKey.asDatum(index)); } FunctionContext[] contexts = entry.getValue(); http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java index b394390..5a262a6 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java @@ -192,7 +192,7 @@ public class DistinctGroupbySecondAggregationExec extends UnaryPhysicalExec { throw new IOException(e.getMessage(), e); } - int distinctSeq = tuple.get(0).asInt2(); + int distinctSeq = tuple.getInt2(0); Tuple keyTuple = getKeyTuple(distinctSeq, tuple); if (prevKeyTuple == null) { @@ -267,12 +267,12 @@ public class DistinctGroupbySecondAggregationExec extends UnaryPhysicalExec { int[] columnIndexes = distinctKeyIndexes[distinctSeq]; Tuple keyTuple = new VTuple(numGroupingColumns + columnIndexes.length + 1); - keyTuple.put(0, tuple.get(0)); + keyTuple.put(0, tuple.asDatum(0)); for (int i = 0; i < numGroupingColumns; i++) { - keyTuple.put(i + 1, tuple.get(i + 1)); + keyTuple.put(i + 1, tuple.asDatum(i + 1)); } for (int i = 0; i < columnIndexes.length; i++) { - keyTuple.put(i + 1 + numGroupingColumns, tuple.get(columnIndexes[i])); + keyTuple.put(i + 1 + numGroupingColumns, tuple.asDatum(columnIndexes[i])); } return keyTuple; http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java index 9ff479b..c91dcca 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java @@ -117,7 +117,7 @@ public class DistinctGroupbySortAggregationExec extends PhysicalExec { int tupleSize = currentTuples[i].size(); for (int j = 0; j < tupleSize; j++) { if (resultColumnIdIndexes[mergeTupleIndex] >= 0) { - mergedTuple.put(resultColumnIdIndexes[mergeTupleIndex], currentTuples[i].get(j)); + mergedTuple.put(resultColumnIdIndexes[mergeTupleIndex], currentTuples[i].asDatum(j)); } mergeTupleIndex++; } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java index 267bd90..5791230 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java @@ -156,7 +156,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec { } for (int i = 0; i < numGroupingColumns; i++) { - resultTuple.put(resultTupleIndexes[i], prevTuple.get(i + 1)); + resultTuple.put(resultTupleIndexes[i], prevTuple.asDatum(i + 1)); } for (DistinctFinalAggregator eachAggr: aggregators) { eachAggr.terminate(resultTuple); @@ -171,7 +171,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec { throw new IOException(e.getMessage(), e); } - int distinctSeq = tuple.get(0).asInt2(); + int distinctSeq = tuple.getInt2(0); Tuple keyTuple = getGroupingKeyTuple(tuple); // First tuple @@ -186,7 +186,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec { if (!prevKeyTuple.equals(keyTuple)) { // new grouping key for (int i = 0; i < numGroupingColumns; i++) { - resultTuple.put(resultTupleIndexes[i], prevTuple.get(i + 1)); + resultTuple.put(resultTupleIndexes[i], prevTuple.asDatum(i + 1)); } for (DistinctFinalAggregator eachAggr: aggregators) { eachAggr.terminate(resultTuple); @@ -219,7 +219,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec { private Tuple getGroupingKeyTuple(Tuple tuple) { Tuple keyTuple = new VTuple(numGroupingColumns); for (int i = 0; i < numGroupingColumns; i++) { - keyTuple.put(i, tuple.get(i + 1)); + keyTuple.put(i, tuple.asDatum(i + 1)); } return keyTuple; @@ -272,7 +272,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec { } if (seq == 0 && nonDistinctAggr != null) { - if (!tuple.get(nonDistinctAggr.inTupleIndex).isNull()) { + if (!tuple.isBlankOrNull(nonDistinctAggr.inTupleIndex)) { nonDistinctAggr.merge(tuple); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java index 8ffd503..e6d1a96 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java @@ -52,7 +52,7 @@ public class HashAggregateExec extends AggregationExec { keyTuple = new VTuple(groupingKeyIds.length); // build one key tuple for(int i = 0; i < groupingKeyIds.length; i++) { - keyTuple.put(i, tuple.get(groupingKeyIds[i])); + keyTuple.put(i, tuple.asDatum(groupingKeyIds[i])); } FunctionContext [] contexts = hashTable.get(keyTuple); @@ -98,7 +98,7 @@ public class HashAggregateExec extends AggregationExec { int tupleIdx = 0; for (; tupleIdx < groupingKeyNum; tupleIdx++) { - tuple.put(tupleIdx, keyTuple.get(tupleIdx)); + tuple.put(tupleIdx, keyTuple.asDatum(tupleIdx)); } for (int funcIdx = 0; funcIdx < aggFunctionsNum; funcIdx++, tupleIdx++) { tuple.put(tupleIdx, aggFunctions[funcIdx].terminate(contexts[funcIdx])); http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java index e94bc26..0a812ee 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java @@ -20,7 +20,6 @@ package org.apache.tajo.engine.planner.physical; import org.apache.tajo.catalog.statistics.StatisticsUtil; import org.apache.tajo.catalog.statistics.TableStats; -import org.apache.tajo.datum.Datum; import org.apache.tajo.plan.logical.StoreTableNode; import org.apache.tajo.storage.Appender; import org.apache.tajo.storage.Tuple; @@ -72,11 +71,10 @@ public class HashBasedColPartitionStoreExec extends ColPartitionStoreExec { sb.delete(0, sb.length()); if (keyIds != null) { for(int i = 0; i < keyIds.length; i++) { - Datum datum = tuple.get(keyIds[i]); if(i > 0) sb.append("/"); sb.append(keyNames[i]).append("="); - sb.append(StringUtils.escapePathName(datum.asChars())); + sb.append(StringUtils.escapePathName(tuple.getText(keyIds[i]))); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java index a4215fa..3065c15 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java @@ -23,7 +23,10 @@ import org.apache.tajo.storage.Tuple; import org.apache.tajo.worker.TaskAttemptContext; import java.io.IOException; -import java.util.*; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; public class HashJoinExec extends CommonHashJoinExec<List<Tuple>> { @@ -50,6 +53,7 @@ public class HashJoinExec extends CommonHashJoinExec<List<Tuple>> { projector.eval(frameTuple, outTuple); return outTuple; } + Tuple leftTuple = leftChild.next(); // it comes from a disk if (leftTuple == null || leftFiltered(leftTuple)) { // if no more tuples in left tuples on disk, a join is completed. finished = leftTuple == null; http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java index 8613eac..27f683b 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java @@ -25,7 +25,8 @@ import org.apache.tajo.storage.Tuple; import org.apache.tajo.worker.TaskAttemptContext; import java.io.IOException; -import java.util.*; +import java.util.Iterator; +import java.util.List; public class HashLeftOuterJoinExec extends HashJoinExec { http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java index 233d6ec..20a9128 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java @@ -41,7 +41,7 @@ public class HashPartitioner extends Partitioner { // build one key tuple for (int i = 0; i < partitionKeyIds.length; i++) { - keyTuple.put(i, tuple.get(partitionKeyIds[i])); + keyTuple.put(i, tuple.asDatum(partitionKeyIds[i])); } return (keyTuple.hashCode() & Integer.MAX_VALUE) % numPartitions; } http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java index a59f8d9..515a2bd 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java @@ -64,13 +64,13 @@ public class JoinTupleComparator implements Comparator<Tuple> { if (outerTuple == null) { outer = NullDatum.get(); } else { - outer = outerTuple.get(outerSortKeyIds[i]); + outer = outerTuple.asDatum(outerSortKeyIds[i]); } if (innerTuple == null) { inner = NullDatum.get(); } else { - inner = innerTuple.get(innerSortKeyIds[i]); + inner = innerTuple.asDatum(innerSortKeyIds[i]); } if (outer instanceof NullDatum || inner instanceof NullDatum) { http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java index 7bf9e66..92e625c 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java @@ -90,20 +90,25 @@ public class RangeShuffleFileWriteExec extends UnaryPhysicalExec { @Override public Tuple next() throws IOException { Tuple tuple; - Tuple keyTuple; + VTuple keyTuple; Tuple prevKeyTuple = null; long offset; - while(!context.isStopped() && (tuple = child.next()) != null) { - offset = appender.getOffset(); - appender.addTuple(tuple); - keyTuple = new VTuple(keySchema.size()); - RowStoreUtil.project(tuple, keyTuple, indexKeys); - if (prevKeyTuple == null || !prevKeyTuple.equals(keyTuple)) { - indexWriter.write(keyTuple, offset); - prevKeyTuple = keyTuple; + try { + while(!context.isStopped() && (tuple = child.next()) != null) { + offset = appender.getOffset(); + appender.addTuple(tuple); + keyTuple = new VTuple(keySchema.size()); + RowStoreUtil.project(tuple, keyTuple, indexKeys); + if (prevKeyTuple == null || !prevKeyTuple.equals(keyTuple)) { + indexWriter.write(keyTuple, offset); + prevKeyTuple = keyTuple; + } } + } catch (RuntimeException e) { + e.printStackTrace(); + throw e; } return null; http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java index 9831d83..2feecd1 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java @@ -18,7 +18,6 @@ package org.apache.tajo.engine.planner.physical; -import org.apache.tajo.datum.NullDatum; import org.apache.tajo.plan.function.FunctionContext; import org.apache.tajo.plan.logical.GroupbyNode; import org.apache.tajo.storage.Tuple; @@ -61,7 +60,7 @@ public class SortAggregateExec extends AggregationExec { // get a key tuple currentKey = new VTuple(groupingKeyIds.length); for(int i = 0; i < groupingKeyIds.length; i++) { - currentKey.put(i, tuple.get(groupingKeyIds[i])); + currentKey.put(i, tuple.asDatum(groupingKeyIds[i])); } /** Aggregation State */ @@ -72,7 +71,7 @@ public class SortAggregateExec extends AggregationExec { // Merge when aggregator doesn't receive NullDatum if (!(groupingKeyNum == 0 && aggFunctionsNum == tuple.size() - && tuple.get(i) == NullDatum.get())) { + && tuple.isBlankOrNull(i))) { aggFunctions[i].merge(contexts[i], tuple); } } @@ -90,7 +89,7 @@ public class SortAggregateExec extends AggregationExec { int tupleIdx = 0; for(; tupleIdx < groupingKeyNum; tupleIdx++) { - outputTuple.put(tupleIdx, lastKey.get(tupleIdx)); + outputTuple.put(tupleIdx, lastKey.asDatum(tupleIdx)); } for(int aggFuncIdx = 0; aggFuncIdx < aggFunctionsNum; tupleIdx++, aggFuncIdx++) { outputTuple.put(tupleIdx, aggFunctions[aggFuncIdx].terminate(contexts[aggFuncIdx])); @@ -114,7 +113,7 @@ public class SortAggregateExec extends AggregationExec { outputTuple = new VTuple(outSchema.size()); int tupleIdx = 0; for(; tupleIdx < groupingKeyNum; tupleIdx++) { - outputTuple.put(tupleIdx, lastKey.get(tupleIdx)); + outputTuple.put(tupleIdx, lastKey.asDatum(tupleIdx)); } for(int aggFuncIdx = 0; aggFuncIdx < aggFunctionsNum; tupleIdx++, aggFuncIdx++) { outputTuple.put(tupleIdx, aggFunctions[aggFuncIdx].terminate(contexts[aggFuncIdx])); http://git-wip-us.apache.org/repos/asf/tajo/blob/5c2aee23/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java index ca90b0e..a40fc1d 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java @@ -52,7 +52,7 @@ public class SortBasedColPartitionStoreExec extends ColPartitionStoreExec { private void fillKeyTuple(Tuple inTuple, Tuple keyTuple) { for (int i = 0; i < keyIds.length; i++) { - keyTuple.put(i, inTuple.get(keyIds[i])); + keyTuple.put(i, inTuple.asDatum(keyIds[i])); } } @@ -60,12 +60,11 @@ public class SortBasedColPartitionStoreExec extends ColPartitionStoreExec { StringBuilder sb = new StringBuilder(); for(int i = 0; i < keyIds.length; i++) { - Datum datum = keyTuple.get(i); if(i > 0) { sb.append("/"); } sb.append(keyNames[i]).append("="); - sb.append(StringUtils.escapePathName(datum.asChars())); + sb.append(StringUtils.escapePathName(keyTuple.getText(i))); } return sb.toString(); }
