TAJO-1833: Refine LogicalPlanPreprocessor to add new rules easily

tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java

@@ -126,6 +126,10 @@ public QueryBlock getQueryBlock() {
       return queryBlock;
     }
 
+    public LogicalPlan getPlan() {
+      return plan;
+    }
+
     public OverridableConf getQueryContext() {
       return queryContext;
     }
@@ -160,7 +164,7 @@ public LogicalPlan createPlan(OverridableConf queryContext, Expr expr, boolean d
 
     QueryBlock rootBlock = plan.newAndGetBlock(LogicalPlan.ROOT_BLOCK);
     PlanContext context = new PlanContext(queryContext, plan, rootBlock, evalOptimizer, debug);
-    preprocessor.visit(context, new Stack<Expr>(), expr);
+    preprocessor.process(context, expr);
     plan.resetGeneratedId();
     LogicalNode topMostNode = this.visit(context, new Stack<Expr>(), expr);
 

tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/BaseLogicalPlanPreprocessEngine.java

@@ -0,0 +1,100 @@
+/**
+ * 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.plan.rewrite;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.algebra.Expr;
+import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.exception.TajoException;
+import org.apache.tajo.plan.ExprAnnotator;
+import org.apache.tajo.plan.LogicalPlanner.PlanContext;
+import org.apache.tajo.plan.logical.LogicalNode;
+
+import java.lang.reflect.Constructor;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+public class BaseLogicalPlanPreprocessEngine implements LogicalPlanPreprocessEngine {
+  private final CatalogService catalogService;
+  private final ExprAnnotator exprAnnotator;
+
+  public BaseLogicalPlanPreprocessEngine(CatalogService catalogService, ExprAnnotator exprAnnotator) {
+    this.catalogService = catalogService;
+    this.exprAnnotator = exprAnnotator;
+  }
+
+  /** class logger */
+  private Log LOG = LogFactory.getLog(BaseLogicalPlanPreprocessEngine.class);
+
+  /** a map for pre-process phases */
+  private Map<String, LogicalPlanPreprocessPhase> preprocessPhases = new LinkedHashMap<>();
+
+  /**
+   * Add a pre-process phase to this engine.
+   *
+   * @param phases pre-process phase
+   */
+  public void addProcessPhase(Iterable<Class<? extends LogicalPlanPreprocessPhase>> phases) {
+    for (Class<? extends LogicalPlanPreprocessPhase> clazz : phases) {
+      try {
+        Constructor cons = clazz.getConstructor(CatalogService.class, ExprAnnotator.class);
+        LogicalPlanPreprocessPhase rule = (LogicalPlanPreprocessPhase) cons.newInstance(catalogService, exprAnnotator);
+        addProcessPhase(rule);
+      } catch (Throwable t) {
+        throw new RuntimeException(t);
+      }
+    }
+  }
+
+  /**
+   * Add a pre-process phase to this engine.
+   *
+   * @param phase The pre-process phase to be added to this engine.
+   */
+  public void addProcessPhase(LogicalPlanPreprocessPhase phase) {
+    if (!preprocessPhases.containsKey(phase.getName())) {
+      preprocessPhases.put(phase.getName(), phase);
+    }
+  }
+
+  /**
+   * Do every pre-process phase added to this engine.
+   *
+   * @param context
+   * @return The rewritten logical node.
+   */
+  @Override
+  public LogicalNode process(PlanContext context, Expr expr) throws TajoException {
+    LogicalPlanPreprocessPhase rule;
+    LogicalNode node = null;
+    for (Entry<String, LogicalPlanPreprocessPhase> preprocessPhase : preprocessPhases.entrySet()) {
+      rule = preprocessPhase.getValue();
+      if (rule.isEligible(context, expr)) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("The rule \"" + rule.getName() + " \" rewrites the query.");
+        }
+        node = rule.process(context, expr);
+      }
+    }
+
+    return node;
+  }
+}

tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/BaseLogicalPlanPreprocessPhaseProvider.java

@@ -0,0 +1,34 @@
+/**
+ * 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.plan.rewrite;
+
+import org.apache.tajo.util.TUtil;
+
+import java.util.Collection;
+import java.util.List;
+
+public class BaseLogicalPlanPreprocessPhaseProvider extends LogicalPlanPreprocessPhaseProvider {
+  @Override
+  public Collection<Class<? extends LogicalPlanPreprocessPhase>> getPhases() {
+    List phases = TUtil.newList(
+        BaseSchemaBuildPhase.class
+    );
+    return phases;
+  }
+}

tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/BaseLogicalPlanRewriteEngine.java

@@ -37,7 +37,7 @@ public class BaseLogicalPlanRewriteEngine implements LogicalPlanRewriteEngine {
   private Log LOG = LogFactory.getLog(BaseLogicalPlanRewriteEngine.class);
 
   /** a map for query rewrite rules  */
-  private Map<String, LogicalPlanRewriteRule> rewriteRules = new LinkedHashMap<String, LogicalPlanRewriteRule>();
+  private Map<String, LogicalPlanRewriteRule> rewriteRules = new LinkedHashMap<>();
 
   /**
    * Add a query rewrite rule to this engine.