for review and discussion

common/src/main/java/org/apache/drill/common/config/DrillConfig.java

@@ -19,6 +19,7 @@
 
 import java.lang.management.ManagementFactory;
 import java.lang.management.RuntimeMXBean;
+import java.lang.reflect.Constructor;
 import java.net.URL;
 import java.util.Collection;
 import java.util.List;
@@ -27,17 +28,19 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.drill.common.exceptions.DrillConfigurationException;
+import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.scanner.ClassPathScanner;
 import org.reflections.util.ClasspathHelper;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableList;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 import com.typesafe.config.ConfigRenderOptions;
 
-public final class DrillConfig extends NestedConfig{
+public class DrillConfig extends NestedConfig {
   private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillConfig.class);
 
   private final ImmutableList<String> startupArguments;
@@ -58,6 +61,36 @@ public DrillConfig(Config config, boolean enableServerConfigs) {
     logger.debug("DrillConfig object initialized.");
   }
 
+  /**
+   * Get an instance of the provided interface using the configuration path provided. Construct the object based on the
+   * provided constructor arguments.
+   * @param path
+   *          The configuration path to use.
+   * @param iface
+   *          The Interface or Superclass of the instance you requested.
+   * @param constructorArgs
+   *          Any arguments required for constructing the requested type.
+   * @return The new Object instance that implements the provided Interface
+   */
+  @SuppressWarnings("unchecked")
+  public <T> T getInstance(String path, Class<T> iface, Object... constructorArgs) {
+    try{
+      String className = this.getString(path);
+      Class<?> clazz = Class.forName(className);
+      Preconditions.checkArgument(iface.isAssignableFrom(clazz));
+      Class<?>[] argClasses = new Class[constructorArgs.length];
+      for (int i = 0; i < constructorArgs.length; i++) {
+        argClasses[i] = constructorArgs[i].getClass();
+      }
+      Constructor<?> constructor = clazz.getConstructor(argClasses);
+      return (T) constructor.newInstance(constructorArgs);
+    }catch(Exception e){
+      throw UserException.unsupportedError(e)
+          .message("Failure while attempting to load instance of the class of type %s requested at path %s.",
+              iface.getName(), path).build(logger);
+    }
+  }
+
   public List<String> getStartupArguments() {
     return startupArguments;
   }
@@ -81,6 +114,7 @@ public static DrillConfig forClient() {
     return create(null, false);
   }
 
+
   /**
    * <p>
    * DrillConfig loads up Drill configuration information. It does this utilizing a combination of classpath scanning

common/src/main/java/org/apache/drill/common/exceptions/UserException.java

@@ -50,7 +50,16 @@ public class UserException extends DrillRuntimeException {
    */
   public static Builder memoryError(final Throwable cause) {
     return UserException.resourceError(cause)
-      .message(MEMORY_ERROR_MSG);
+      .message(MEMORY_ERROR_MSG).addContext(cause.getMessage());
+  }
+
+  public static Builder memoryError(final String format, final Object... args) {
+    Builder builder =  UserException.resourceError();
+    builder.message(MEMORY_ERROR_MSG);
+    if (!format.isEmpty()) {
+      builder.addContext(String.format(format, args));
+    }
+    return builder;
   }
 
   /**
@@ -59,7 +68,7 @@ public static Builder memoryError(final Throwable cause) {
    * @return resource error builder
    */
   public static Builder memoryError() {
-    return memoryError(null);
+    return memoryError("");
   }
 
   /**

common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java

@@ -89,6 +89,14 @@ private static Slf4jReporter getLogReporter() {
     }
   }
 
+  public synchronized static <T extends Metric> void register(String name, T metric) {
+    boolean removed = RegistryHolder.REGISTRY.remove(name);
+    if (removed) {
+      logger.warn("Removing old metric since name matched newly registered metric. Metric name: {}", name);
+    }
+    RegistryHolder.REGISTRY.register(name, metric);
+  }
+
   private static void registerAll(String prefix, MetricSet metricSet, MetricRegistry registry) {
     for (Entry<String, Metric> entry : metricSet.getMetrics().entrySet()) {
       if (entry.getValue() instanceof MetricSet) {

contrib/format-lucene/src/main/java/org/apache/drill/exec/planner/logical/LucenePushFilterIntoScan.java

@@ -23,9 +23,11 @@
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.plan.RelOptRuleOperand;
 import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.InvalidRelException;
 import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.planner.common.DrillJoinRelBase;
 import org.apache.drill.exec.store.StoragePluginOptimizerRule;
 import org.apache.drill.exec.store.lucene.LuceneGroupScan;
 import org.apache.drill.exec.store.lucene.LuceneScanSpec;
@@ -39,6 +41,7 @@ public abstract class LucenePushFilterIntoScan extends StoragePluginOptimizerRul
     @Override
     public boolean matches(RelOptRuleCall call) {
       final DrillScanRel scan = (DrillScanRel) call.rel(1);
+
       GroupScan groupScan = scan.getGroupScan();
       if (groupScan instanceof LuceneGroupScan) {
         if (((LuceneGroupScan) groupScan).supportsFilterPushDown()) {
@@ -60,6 +63,7 @@ public void onMatch(RelOptRuleCall call) {
       "LucenePushPartitionFilterIntoScan:Filter_On_Project") {
     @Override
     public boolean matches(RelOptRuleCall call) {
+
       final DrillScanRel scan = (DrillScanRel) call.rel(2);
       GroupScan groupScan = scan.getGroupScan();
       if (groupScan instanceof LuceneGroupScan) {
@@ -79,13 +83,90 @@ public void onMatch(RelOptRuleCall call) {
     }
   };
 
+  public static final StoragePluginOptimizerRule LUCENE_FILTER_ON_JOIN_REL = new LucenePushFilterIntoScan(RelOptHelper.some(DrillJoinRel.class, RelOptHelper.any(DrillScanRel.class)), "LucenePushPartitionFilterIntoScan:Filter_On_Join_Filter") {
+
+    @Override
+    public void doOnJoinMatch(RelOptRuleCall call, DrillJoinRel joinRel, DrillScanRel scanRel) {
+      super.doOnJoinMatch(call, joinRel, scanRel);
+    }
+
+    @Override
+    public boolean matches(RelOptRuleCall call) {
+
+      final DrillScanRel scan = (DrillScanRel) call.rel(1);
+      GroupScan groupScan = scan.getGroupScan();
+
+      if (groupScan instanceof LuceneGroupScan) {
+        if (((LuceneGroupScan) groupScan).supportsFilterPushDown()) {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    @Override
+    public void onMatch(RelOptRuleCall call) {
+      final DrillJoinRel joinRel = (DrillJoinRel) call.rel(0);
+      final DrillScanRel scanRel = (DrillScanRel) call.rel(1);
+      doOnJoinMatch(call, joinRel, scanRel);
+    }
+
+  };
+
+  public void doOnJoinMatch(RelOptRuleCall call, DrillJoinRel joinRel, DrillScanRel scanRel) {
+
+    SqlJoinFilterToLuceneQuery sqlFilterToLuceneQuery = new SqlJoinFilterToLuceneQuery(((LuceneGroupScan) scanRel.getGroupScan()).getIndexFields(), joinRel);
+
+    joinRel.getCondition().accept(sqlFilterToLuceneQuery);
+
+    LuceneGroupScan luceneGroupScan = (LuceneGroupScan) scanRel.getGroupScan();
+    LuceneScanSpec luceneScanSpec = luceneGroupScan.getLuceneScanSpec();
+    LuceneScanSpec newLuceneScanSpec = new LuceneScanSpec(luceneScanSpec.getSelectionRoot(), luceneScanSpec.getSelection(), sqlFilterToLuceneQuery.getLuceneQuery());
+    try {
+
+      LuceneGroupScan newLuceneGroupScan = new LuceneGroupScan(
+              luceneGroupScan.getUserName(),
+              newLuceneScanSpec,
+              luceneGroupScan.getFormatPlugin(),
+              luceneGroupScan.getColumns()
+      );
+
+      DrillScanRel newScanRel = new DrillScanRel(
+                scanRel.getCluster(),
+                scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
+                scanRel.getTable(),
+                newLuceneGroupScan,
+                scanRel.getRowType(),
+                scanRel.getColumns(),
+                true
+      );
+
+      /*
+      DrillJoinRel newScanRel = new DrillJoinRel(
+              scanRel.getCluster(),
+              scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
+              joinRel
+              scanRel.getTable(),
+              newLuceneGroupScan,
+              scanRel.getRowType(),
+              scanRel.getColumns()
+      );
+      */
+
+      //todo - figure out why this is breaking (see the re-writing above)
+      call.transformTo(newScanRel);
+    } catch (IOException e) {
+      throw new DrillRuntimeException(e);
+    }
+
+  }
 
   private LucenePushFilterIntoScan(RelOptRuleOperand operand, String id) {
     super(operand, id);
   }
 
-
   public void doOnMatch(RelOptRuleCall call, DrillFilterRel filterRel, DrillProjectRel projectRel, DrillScanRel scanRel) {
+
     DrillRel inputRel = projectRel != null ? projectRel : scanRel;
     SqlFilterToLuceneQuery sqlFilterToLuceneQuery = new SqlFilterToLuceneQuery(((LuceneGroupScan) scanRel.getGroupScan()).getIndexFields(), inputRel);