TAJO-922: Add NTH_VALUE window function

tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4

@@ -270,6 +270,7 @@ MONTH : M O N T H;
 NATIONAL : N A T I O N A L;
 NULLIF : N U L L I F;
 NO : N O;
+NTH_VALUE : N T H UNDERLINE V A L U E;
 
 OVERWRITE : O V E R W R I T E;
 OTHERS: O T H E R S;

tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4

@@ -639,6 +639,7 @@ window_function_type
   | LAST_VALUE LEFT_PAREN column_reference RIGHT_PAREN
   | LAG LEFT_PAREN column_reference ( COMMA numeric_value_expression ( COMMA common_value_expression )? )? RIGHT_PAREN
   | LEAD LEFT_PAREN column_reference ( COMMA numeric_value_expression ( COMMA common_value_expression )? )? RIGHT_PAREN
+  | NTH_VALUE LEFT_PAREN column_reference COMMA numeric_value_expression RIGHT_PAREN
   ;
 
 rank_function_type

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValue.java

@@ -0,0 +1,69 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.plan.function.AggFunction;
+import org.apache.tajo.plan.function.FunctionContext;
+import org.apache.tajo.storage.Tuple;
+
+
+public abstract class NthValue extends AggFunction<Datum> {
+
+  public NthValue(Column[] columns) {
+    super(columns);
+  }
+
+  @Override
+  public FunctionContext newContext() {
+    return new NthValueContext();
+  }
+
+  @Override
+  public void eval(FunctionContext ctx, Tuple params) {
+    NthValueContext nThValueCtx = (NthValueContext) ctx;
+    Datum datum = params.get(1);
+    nThValueCtx.rowCount++;
+    if (nThValueCtx.rowCount == datum.asInt8()) {
+      nThValueCtx.nThValue = params.get(0);
+    }
+  }
+
+  @Override
+  public Datum getPartialResult(FunctionContext ctx) {
+    return ((NthValueContext) ctx).nThValue;
+  }
+
+  @Override
+  public Datum terminate(FunctionContext ctx) {
+    if (((NthValueContext) ctx).nThValue == null) {
+      return NullDatum.get();
+    }
+    else {
+      return ((NthValueContext) ctx).nThValue;
+    }
+  }
+
+  protected static class NthValueContext implements FunctionContext {
+    Datum nThValue = null;
+    long rowCount = 0;
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueDate.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.DATE,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.DATE, TajoDataTypes.Type.INT4})}
+)
+public class NthValueDate extends NthValue {
+
+  public NthValueDate() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.DATE),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.DATE);
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueDouble.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.FLOAT8,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.FLOAT8, TajoDataTypes.Type.INT4})}
+)
+public class NthValueDouble extends NthValue {
+
+  public NthValueDouble() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.FLOAT8),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueFloat.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.FLOAT4,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.FLOAT4, TajoDataTypes.Type.INT4})}
+)
+public class NthValueFloat extends NthValue {
+
+  public NthValueFloat() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.FLOAT4),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueInt.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.INT4,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.INT4, TajoDataTypes.Type.INT4})}
+)
+public class NthValueInt extends NthValue {
+
+  public NthValueInt() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.INT4),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT4);
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueLong.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.INT8,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.INT8, TajoDataTypes.Type.INT4})}
+)
+public class NthValueLong extends NthValue {
+
+  public NthValueLong() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.INT8),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
+  }
+}

tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/NthValueString.java

@@ -0,0 +1,47 @@
+/**
+ * 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.tajo.engine.function.builtin;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+
+@Description(
+    functionName = "nth_value",
+    description = "the nth value of expr",
+    example = "> SELECT nth_value(col, expr);",
+    returnType = TajoDataTypes.Type.TEXT,
+    paramTypes = {@ParamTypes(paramTypes = {TajoDataTypes.Type.TEXT, TajoDataTypes.Type.INT4})}
+)
+public class NthValueString extends NthValue {
+
+  public NthValueString() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.TEXT),
+        new Column("num", TajoDataTypes.Type.INT4),
+    });
+  }
+
+  @Override
+  public TajoDataTypes.DataType getPartialResultType() {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TEXT);
+  }
+}