TAJO-962: Column reference used in LIMIT clause incurs NPE.

tajo-core/src/main/java/org/apache/tajo/engine/planner/PreLogicalPlanVerifier.java

@@ -89,6 +89,20 @@ public Expr visitProjection(Context context, Stack<Expr> stack, Projection expr)
     return expr;
   }
 
+  @Override
+  public Expr visitLimit(Context context, Stack<Expr> stack, Limit expr) throws PlanningException {
+    stack.push(expr);
+
+    if (ExprFinder.finds(expr.getFetchFirstNum(), OpType.Column).size() > 0) {
+      context.state.addVerification("argument of LIMIT must not contain variables");
+    }
+
+    visit(context, stack, expr.getFetchFirstNum());
+    Expr result = visit(context, stack, expr.getChild());
+    stack.pop();
+    return result;
+  }
+
   @Override
   public Expr visitGroupBy(Context context, Stack<Expr> stack, Aggregation expr) throws PlanningException {
     super.visitGroupBy(context, stack, expr);

tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java

@@ -18,6 +18,7 @@
 
 package org.apache.tajo.master;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
 import org.apache.commons.logging.Log;
@@ -115,6 +116,26 @@ public void stop() {
     super.stop();
   }
 
+  @VisibleForTesting
+  public SQLAnalyzer getAnalyzer() {
+    return analyzer;
+  }
+
+  @VisibleForTesting
+  public PreLogicalPlanVerifier getPreLogicalPlanVerifier() {
+    return preVerifier;
+  }
+
+  @VisibleForTesting
+  public LogicalPlanner getLogicalPlanner() {
+    return planner;
+  }
+
+  @VisibleForTesting
+  public LogicalOptimizer getLogicalOptimizer() {
+    return optimizer;
+  }
+
   private QueryContext createQueryContext(Session session) {
     QueryContext newQueryContext =  new QueryContext(context.getConf(), session);
 

tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPreLogicalPlanVerifier.java

@@ -0,0 +1,74 @@
+/*
+ * Lisensed 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.planner;
+
+import org.apache.tajo.LocalTajoTestingUtility;
+import org.apache.tajo.QueryTestCaseBase;
+import org.apache.tajo.algebra.Expr;
+import org.apache.tajo.engine.parser.SQLAnalyzer;
+import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.master.GlobalEngine;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.fail;
+
+public class TestPreLogicalPlanVerifier extends QueryTestCaseBase {
+
+  private static SQLAnalyzer analyzer;
+  private static PreLogicalPlanVerifier verifier;
+
+  @BeforeClass
+  public static void setUp() {
+    GlobalEngine engine = testingCluster.getMaster().getContext().getGlobalEngine();
+    analyzer = engine.getAnalyzer();
+    verifier = engine.getPreLogicalPlanVerifier();
+  }
+
+  public static VerificationState verify(String query) throws PlanningException {
+
+    VerificationState state = new VerificationState();
+    QueryContext context = LocalTajoTestingUtility.createDummyContext(conf);
+
+    Expr expr = analyzer.parse(query);
+    verifier.verify(context, state, expr);
+
+    return state;
+  }
+
+  public static void valid(String query) throws PlanningException {
+    VerificationState state = verify(query);
+    if (state.errorMessages.size() > 0) {
+      fail(state.getErrorMessages().get(0));
+    }
+  }
+
+  public static void invalid(String query) throws PlanningException {
+    VerificationState state = verify(query);
+    if (state.errorMessages.size() == 0) {
+      fail(PreLogicalPlanVerifier.class.getSimpleName() + " cannot catch any verification error: " + query);
+    }
+  }
+
+  @Test
+  public void testLimitWithFieldReference() throws PlanningException {
+    valid("select * from lineitem limit 3");
+    invalid("select * from lineitem limit l_orderkey");
+  }
+}