TAJO-939: Refactoring the column resolver in LogicalPlan

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java

@@ -95,11 +95,22 @@ private void init() {
    * @param qualifier The qualifier
    */
   public void setQualifier(String qualifier) {
+    Schema copy = null;
+    try {
+      copy = (Schema) clone();
+    } catch (CloneNotSupportedException e) {
+      throw new RuntimeException(e);
+    }
+
+    fields.clear();
     fieldsByQualifiedName.clear();
-    for (int i = 0; i < size(); i++) {
-      Column column = fields.get(i);
-      fields.set(i, new Column(qualifier + "." + column.getSimpleName(), column.getDataType()));
-      fieldsByQualifiedName.put(fields.get(i).getQualifiedName(), i);
+    fieldsByName.clear();
+
+    Column newColumn;
+    for (int i = 0; i < copy.size(); i++) {
+      Column column = copy.getColumn(i);
+      newColumn = new Column(qualifier + "." + column.getSimpleName(), column.getDataType());
+      addColumn(newColumn);
     }
   }
 

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

@@ -32,6 +32,8 @@
 import org.apache.tajo.engine.function.AggFunction;
 import org.apache.tajo.engine.function.GeneralFunction;
 import org.apache.tajo.engine.planner.logical.NodeType;
+import org.apache.tajo.engine.planner.nameresolver.NameResolvingMode;
+import org.apache.tajo.engine.planner.nameresolver.NameResolver;
 import org.apache.tajo.exception.InternalException;
 import org.apache.tajo.util.Pair;
 import org.apache.tajo.util.TUtil;
@@ -66,16 +68,19 @@ public ExprAnnotator(CatalogService catalog) {
   static class Context {
     LogicalPlan plan;
     LogicalPlan.QueryBlock currentBlock;
+    NameResolvingMode columnRsvLevel;
 
-    public Context(LogicalPlan plan, LogicalPlan.QueryBlock block) {
+    public Context(LogicalPlan plan, LogicalPlan.QueryBlock block, NameResolvingMode colRsvLevel) {
       this.plan = plan;
       this.currentBlock = block;
+      this.columnRsvLevel = colRsvLevel;
     }
   }
 
-  public EvalNode createEvalNode(LogicalPlan plan, LogicalPlan.QueryBlock block, Expr expr)
+  public EvalNode createEvalNode(LogicalPlan plan, LogicalPlan.QueryBlock block, Expr expr,
+                                 NameResolvingMode colRsvLevel)
       throws PlanningException {
-    Context context = new Context(plan, block);
+    Context context = new Context(plan, block, colRsvLevel);
     return AlgebraicUtil.eliminateConstantExprs(visit(context, new Stack<Expr>(), expr));
   }
 
@@ -540,7 +545,20 @@ public EvalNode visitSign(Context ctx, Stack<Expr> stack, SignedExpr expr) throw
   @Override
   public EvalNode visitColumnReference(Context ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
       throws PlanningException {
-    Column column = ctx.plan.resolveColumn(ctx.currentBlock, expr);
+    Column column;
+
+    switch (ctx.columnRsvLevel) {
+    case LEGACY:
+      column = ctx.plan.resolveColumn(ctx.currentBlock, expr);
+      break;
+    case RELS_ONLY:
+    case RELS_AND_SUBEXPRS:
+    case SUBEXPRS_AND_RELS:
+      column = NameResolver.resolve(ctx.plan, ctx.currentBlock, expr, ctx.columnRsvLevel);
+      break;
+    default:
+      throw new PlanningException("Unsupported column resolving level: " + ctx.columnRsvLevel.name());
+    }
     return new FieldEval(column);
   }
 

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

@@ -18,16 +18,16 @@
 
 package org.apache.tajo.engine.planner;
 
-import com.google.common.collect.Sets;
 import com.google.common.collect.Sets;
 import org.apache.tajo.algebra.*;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.engine.exception.NoSuchColumnException;
+import org.apache.tajo.engine.planner.nameresolver.NameResolvingMode;
+import org.apache.tajo.engine.planner.nameresolver.NameResolver;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
-import java.util.Set;
 import java.util.Stack;
 
 /**
@@ -330,11 +330,18 @@ public Expr visitCastExpr(ExprNormalizedResult ctx, Stack<Expr> stack, CastExpr
   @Override
   public Expr visitColumnReference(ExprNormalizedResult ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
       throws PlanningException {
+
+    if (ctx.block.isAliasedName(expr.getCanonicalName())) {
+      String originalName = ctx.block.getOriginalName(expr.getCanonicalName());
+      expr.setName(originalName);
+      return expr;
+    }
     // if a column reference is not qualified, it finds and sets the qualified column name.
     if (!(expr.hasQualifier() && CatalogUtil.isFQTableName(expr.getQualifier()))) {
       if (!ctx.block.namedExprsMgr.contains(expr.getCanonicalName()) && expr.getType() == OpType.Column) {
         try {
-          String normalized = ctx.plan.getNormalizedColumnName(ctx.block, expr);
+          String normalized =
+              NameResolver.resolve(ctx.plan, ctx.block, expr, NameResolvingMode.LEGACY).getQualifiedName();
           expr.setName(normalized);
         } catch (NoSuchColumnException nsc) {
         }