Update after rebase

presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java

@@ -30,6 +30,7 @@
 import com.facebook.airlift.tracetoken.TraceTokenModule;
 import com.facebook.drift.server.DriftServer;
 import com.facebook.drift.transport.netty.server.DriftNettyServerTransport;
+import com.facebook.presto.connector.ConnectorAwareNodeManager;
 import com.facebook.presto.dispatcher.QueryPrerequisitesManager;
 import com.facebook.presto.dispatcher.QueryPrerequisitesManagerModule;
 import com.facebook.presto.eventlistener.EventListenerManager;
@@ -39,12 +40,14 @@
 import com.facebook.presto.execution.warnings.WarningCollectorModule;
 import com.facebook.presto.metadata.Catalog;
 import com.facebook.presto.metadata.CatalogManager;
+import com.facebook.presto.metadata.InMemoryNodeManager;
 import com.facebook.presto.metadata.StaticCatalogStore;
 import com.facebook.presto.metadata.StaticFunctionNamespaceStore;
 import com.facebook.presto.security.AccessControlManager;
 import com.facebook.presto.security.AccessControlModule;
 import com.facebook.presto.server.security.PasswordAuthenticatorManager;
 import com.facebook.presto.server.security.ServerSecurityModule;
+import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.sql.parser.SqlParserOptions;
 import com.facebook.presto.sql.planner.PlanFragmenter;
@@ -180,10 +183,13 @@ public void run()
             injector.getInstance(TracerProviderManager.class).loadTracerProvider();
             injector.getInstance(NodeStatusNotificationManager.class).loadNodeStatusNotificationProvider();
             injector.getInstance(GracefulShutdownHandler.class).loadNodeStatusNotification();
+
+            ConnectorAwareNodeManager connectorAwareNodeManager = new ConnectorAwareNodeManager(injector.getInstance(InMemoryNodeManager.class), "", new ConnectorId("<NA>"));
             PlanCheckerProviderManager planCheckerProviderManager = injector.getInstance(PlanCheckerProviderManager.class);
-            planCheckerProviderManager.loadPlanCheckerProviders();
+            planCheckerProviderManager.loadPlanCheckerProviders(connectorAwareNodeManager);
             planCheckerProviderManager.updatePlanCheckerProviders(injector.getInstance(PlanChecker.class));
             planCheckerProviderManager.updatePlanCheckerProviders(injector.getInstance(PlanFragmenter.class));
+
             startAssociatedProcesses(injector);
 
             injector.getInstance(Announcer.class).start();

presto-main/src/main/java/com/facebook/presto/sql/planner/sanity/PlanCheckerProviderManager.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.sql.planner.sanity;
 
+import com.facebook.presto.spi.NodeManager;
 import com.facebook.presto.spi.plan.PlanCheckerProvider;
 import com.facebook.presto.spi.plan.PlanCheckerProviderFactory;
 import com.facebook.presto.spi.plan.SimplePlanFragmentSerde;
@@ -49,10 +50,10 @@ public void addPlanCheckerProviderFactory(PlanCheckerProviderFactory planChecker
         }
     }
 
-    public void loadPlanCheckerProviders()
+    public void loadPlanCheckerProviders(NodeManager nodeManager)
     {
         if (!providerFactories.isEmpty()) {
-            providers = providerFactories.values().stream().map(pc -> pc.create(simplePlanFragmentSerde)).collect(Collectors.toList());
+            providers = providerFactories.values().stream().map(pc -> pc.create(simplePlanFragmentSerde, nodeManager)).collect(Collectors.toList());
         }
         else {
             providers = ImmutableList.of(new EmptyPlanCheckerProvider());

presto-main/src/main/java/com/facebook/presto/testing/LocalQueryRunner.java

@@ -25,6 +25,7 @@
 import com.facebook.presto.common.block.SortOrder;
 import com.facebook.presto.common.type.BooleanType;
 import com.facebook.presto.common.type.Type;
+import com.facebook.presto.connector.ConnectorAwareNodeManager;
 import com.facebook.presto.connector.ConnectorManager;
 import com.facebook.presto.connector.ConnectorTypeSerdeManager;
 import com.facebook.presto.connector.system.AnalyzePropertiesSystemTable;
@@ -458,7 +459,8 @@ private LocalQueryRunner(Session defaultSession, FeaturesConfig featuresConfig,
         this.joinFilterFunctionCompiler = new JoinFilterFunctionCompiler(metadata);
 
         this.planCheckerProviderManager = new PlanCheckerProviderManager(new JsonCodecSimplePlanFragmentSerde(jsonCodec(SimplePlanFragment.class)));
-        planCheckerProviderManager.loadPlanCheckerProviders();
+
+        planCheckerProviderManager.loadPlanCheckerProviders(new ConnectorAwareNodeManager(nodeManager, "", new ConnectorId("<NA>")));
         planCheckerProviderManager.updatePlanCheckerProviders(distributedPlanChecker);
         planCheckerProviderManager.updatePlanCheckerProviders(singleNodePlanChecker);
         planCheckerProviderManager.updatePlanCheckerProviders(planFragmenter);

presto-plan-checker-providers/src/main/java/com/facebook/presto/plancheckerproviders/nativechecker/NativePlanCheckerProviderFactory.java

@@ -21,19 +21,27 @@
 import com.facebook.presto.spi.plan.PlanCheckerProviderFactory;
 import com.facebook.presto.spi.plan.SimplePlanFragment;
 import com.facebook.presto.spi.plan.SimplePlanFragmentSerde;
+import com.google.common.collect.ImmutableMap;
 import com.google.inject.Injector;
 import com.google.inject.Scopes;
 
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
 import java.util.Map;
+import java.util.Properties;
 
 import static com.facebook.airlift.configuration.ConfigBinder.configBinder;
 import static com.facebook.airlift.json.JsonBinder.jsonBinder;
 import static com.facebook.airlift.json.JsonCodecBinder.jsonCodecBinder;
+import static com.google.common.collect.Maps.fromProperties;
 import static java.util.Objects.requireNonNull;
 
 public class NativePlanCheckerProviderFactory
         implements PlanCheckerProviderFactory
 {
+    private static final File NATIVE_PLAN_CHECKER_PROVIDER_CONFIG = new File("etc/native-plan-checker-provider.properties");
     private final ClassLoader classLoader;
 
     public NativePlanCheckerProviderFactory(ClassLoader classLoader)
@@ -48,7 +56,7 @@ public String getName()
     }
 
     @Override
-    public PlanCheckerProvider create(Map<String, String> config, NodeManager nodeManager, SimplePlanFragmentSerde simplePlanFragmentSerde)
+    public PlanCheckerProvider create(SimplePlanFragmentSerde simplePlanFragmentSerde, NodeManager nodeManager)
     {
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
             Bootstrap app = new Bootstrap(
@@ -65,11 +73,33 @@ public PlanCheckerProvider create(Map<String, String> config, NodeManager nodeMa
             Injector injector = app
                     .noStrictConfig()
                     .doNotInitializeLogging()
-                    .setRequiredConfigurationProperties(config)
+                    .setRequiredConfigurationProperties(getConfig())
                     .quiet()
                     .initialize();
 
             return injector.getInstance(PlanCheckerProvider.class);
         }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private Map<String, String> getConfig()
+            throws IOException
+    {
+        if (NATIVE_PLAN_CHECKER_PROVIDER_CONFIG.exists()) {
+            return loadProperties(NATIVE_PLAN_CHECKER_PROVIDER_CONFIG);
+        }
+        return ImmutableMap.of();
+    }
+
+    static Map<String, String> loadProperties(File file)
+            throws IOException
+    {
+        Properties properties = new Properties();
+        try (InputStream in = Files.newInputStream(file.toPath())) {
+            properties.load(in);
+        }
+        return fromProperties(properties);
     }
 }

presto-spi/src/main/java/com/facebook/presto/spi/plan/PlanCheckerProviderFactory.java

@@ -13,9 +13,11 @@
  */
 package com.facebook.presto.spi.plan;
 
+import com.facebook.presto.spi.NodeManager;
+
 public interface PlanCheckerProviderFactory
 {
     String getName();
 
-    PlanCheckerProvider create(SimplePlanFragmentSerde simplePlanFragmentSerde);
+    PlanCheckerProvider create(SimplePlanFragmentSerde simplePlanFragmentSerde, NodeManager nodeManager);
 }

presto-tests/src/test/java/com/facebook/presto/execution/TestQueues.java

@@ -30,6 +30,7 @@
 import com.facebook.presto.sql.planner.PlanFragmenter;
 import com.facebook.presto.sql.planner.sanity.PlanChecker;
 import com.facebook.presto.sql.planner.sanity.PlanCheckerProviderManager;
+import com.facebook.presto.testing.TestingNodeManager;
 import com.facebook.presto.tests.DistributedQueryRunner;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
@@ -389,7 +390,7 @@ public void testEagerPlanValidation()
         queryRunner.installPlugin(new ResourceGroupManagerPlugin());
         queryRunner.installPlugin(new TestingPlanCheckerProviderPlugin(triggerValidationFailure));
         PlanCheckerProviderManager planCheckerProviderManager = queryRunner.getCoordinator().getInstance(Key.get(PlanCheckerProviderManager.class));
-        planCheckerProviderManager.loadPlanCheckerProviders();
+        planCheckerProviderManager.loadPlanCheckerProviders(new TestingNodeManager());
         planCheckerProviderManager.updatePlanCheckerProviders(queryRunner.getCoordinator().getInstance(Key.get(PlanChecker.class)));
         planCheckerProviderManager.updatePlanCheckerProviders(queryRunner.getCoordinator().getInstance(Key.get(PlanFragmenter.class)));
         queryRunner.getCoordinator().getResourceGroupManager().get().forceSetConfigurationManager("file", ImmutableMap.of("resource-groups.config-file", getResourceFilePath("resource_groups_config_eager_plan_validation.json")));

presto-tests/src/test/java/com/facebook/presto/execution/TestingPlanCheckerProviderPlugin.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.execution;
 
+import com.facebook.presto.spi.NodeManager;
 import com.facebook.presto.spi.Plugin;
 import com.facebook.presto.spi.plan.PlanChecker;
 import com.facebook.presto.spi.plan.PlanCheckerProvider;
@@ -46,7 +47,7 @@ public String getName()
     }
 
     @Override
-    public PlanCheckerProvider create(SimplePlanFragmentSerde simplePlanFragmentSerde)
+    public PlanCheckerProvider create(SimplePlanFragmentSerde simplePlanFragmentSerde, NodeManager nodeManager)
     {
         return this;
     }