integration/presto/src/main/prestosql/org/apache/carbondata/presto/InternalCarbonDataConnectorFactory.java

/*
 * 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.carbondata.presto;

import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Module;
import com.google.inject.TypeLiteral;
import io.airlift.bootstrap.Bootstrap;
import io.airlift.bootstrap.LifeCycleManager;
import io.airlift.event.client.EventModule;
import io.airlift.json.JsonModule;
import io.prestosql.plugin.base.CatalogName;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeConnectorAccessControl;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeConnectorPageSinkProvider;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeConnectorPageSourceProvider;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeConnectorSplitManager;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeEventListener;
import io.prestosql.plugin.base.classloader.ClassLoaderSafeNodePartitioningProvider;
import io.prestosql.plugin.base.jmx.ConnectorObjectNameGeneratorModule;
import io.prestosql.plugin.base.jmx.MBeanServerModule;
import io.prestosql.plugin.hive.HiveAnalyzeProperties;
import io.prestosql.plugin.hive.HiveMetadataFactory;
import io.prestosql.plugin.hive.HiveSchemaProperties;
import io.prestosql.plugin.hive.HiveSessionProperties;
import io.prestosql.plugin.hive.HiveTableProperties;
import io.prestosql.plugin.hive.HiveTransactionManager;
import io.prestosql.plugin.hive.NodeVersion;
import io.prestosql.plugin.hive.authentication.HiveAuthenticationModule;
import io.prestosql.plugin.hive.azure.HiveAzureModule;
import io.prestosql.plugin.hive.gcs.HiveGcsModule;
import io.prestosql.plugin.hive.metastore.HiveMetastore;
import io.prestosql.plugin.hive.metastore.HiveMetastoreModule;
import io.prestosql.plugin.hive.procedure.HiveProcedureModule;
import io.prestosql.plugin.hive.rubix.RubixEnabledConfig;
import io.prestosql.plugin.hive.rubix.RubixInitializer;
import io.prestosql.plugin.hive.rubix.RubixModule;
import io.prestosql.plugin.hive.s3.HiveS3Module;
import io.prestosql.plugin.hive.security.HiveSecurityModule;
import io.prestosql.plugin.hive.security.SystemTableAwareAccessControl;
import io.prestosql.spi.NodeManager;
import io.prestosql.spi.PageIndexerFactory;
import io.prestosql.spi.PageSorter;
import io.prestosql.spi.VersionEmbedder;
import io.prestosql.spi.classloader.ThreadContextClassLoader;
import io.prestosql.spi.connector.Connector;
import io.prestosql.spi.connector.ConnectorAccessControl;
import io.prestosql.spi.connector.ConnectorContext;
import io.prestosql.spi.connector.ConnectorNodePartitioningProvider;
import io.prestosql.spi.connector.ConnectorPageSinkProvider;
import io.prestosql.spi.connector.ConnectorPageSourceProvider;
import io.prestosql.spi.connector.ConnectorSplitManager;
import io.prestosql.spi.connector.SystemTable;
import io.prestosql.spi.eventlistener.EventListener;
import io.prestosql.spi.procedure.Procedure;
import io.prestosql.spi.type.TypeManager;
import org.weakref.jmx.guice.MBeanModule;

import java.util.Map;
import java.util.Optional;
import java.util.Set;

import org.apache.carbondata.presto.impl.CarbonTableConfig;

import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static com.google.inject.multibindings.Multibinder.newSetBinder;
import static io.airlift.configuration.ConditionalModule.installModuleIf;
import static io.airlift.configuration.ConfigBinder.configBinder;
import static java.util.Objects.requireNonNull;

/**
 * Similar class to InternalHiveConnectorFactory,
 * to handle the module bindings for carbonData connector.
 */
public final class InternalCarbonDataConnectorFactory {
  private InternalCarbonDataConnectorFactory() {
  }

  public static Connector createConnector(String catalogName, Map<String, String> config,
      ConnectorContext context, Module module) {
    return createConnector(catalogName, config, context, module, Optional.empty());
  }

  public static Connector createConnector(String catalogName, Map<String, String> config,
      ConnectorContext context, Module module, Optional<HiveMetastore> metastore) {
    requireNonNull(config, "config is null");

    ClassLoader classLoader = InternalCarbonDataConnectorFactory.class.getClassLoader();
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
      Bootstrap app = new Bootstrap(
          new EventModule(),
          new MBeanModule(),
          new ConnectorObjectNameGeneratorModule(catalogName, "io.prestosql.plugin.carbondata",
              "presto.plugin.carbondata"),
          new JsonModule(),
          new CarbondataModule(),
          new HiveS3Module(),
          new HiveGcsModule(),
          new HiveAzureModule(),
          installModuleIf(RubixEnabledConfig.class, RubixEnabledConfig::isCacheEnabled,
              new RubixModule()),
          new HiveMetastoreModule(metastore),
          new HiveSecurityModule(catalogName),
          new HiveAuthenticationModule(),
          new HiveProcedureModule(),
          new MBeanServerModule(),
          binder -> {
          binder.bind(NodeVersion.class)
              .toInstance(new NodeVersion(context.getNodeManager().getCurrentNode().getVersion()));
          binder.bind(NodeManager.class).toInstance(context.getNodeManager());
          binder.bind(VersionEmbedder.class).toInstance(context.getVersionEmbedder());
          binder.bind(TypeManager.class).toInstance(context.getTypeManager());
          binder.bind(PageIndexerFactory.class).toInstance(context.getPageIndexerFactory());
          binder.bind(PageSorter.class).toInstance(context.getPageSorter());
          binder.bind(CatalogName.class).toInstance(new CatalogName(catalogName));
          configBinder(binder).bindConfig(CarbonTableConfig.class);
          },
          binder -> newSetBinder(binder, EventListener.class), module);

      Injector injector =
          app.strictConfig().doNotInitializeLogging().setRequiredConfigurationProperties(config)
              .initialize();

      if (injector.getInstance(RubixEnabledConfig.class).isCacheEnabled()) {
        // RubixInitializer needs ConfigurationInitializers, hence kept outside RubixModule
        RubixInitializer rubixInitializer = injector.getInstance(RubixInitializer.class);
        rubixInitializer.initializeRubix(context.getNodeManager());
      }

      LifeCycleManager lifeCycleManager = injector.getInstance(LifeCycleManager.class);
      HiveMetadataFactory metadataFactory = injector.getInstance(HiveMetadataFactory.class);
      HiveTransactionManager transactionManager =
          injector.getInstance(HiveTransactionManager.class);
      ConnectorSplitManager splitManager = injector.getInstance(ConnectorSplitManager.class);
      ConnectorPageSourceProvider connectorPageSource =
          injector.getInstance(ConnectorPageSourceProvider.class);
      ConnectorPageSinkProvider pageSinkProvider =
          injector.getInstance(ConnectorPageSinkProvider.class);
      ConnectorNodePartitioningProvider connectorDistributionProvider =
          injector.getInstance(ConnectorNodePartitioningProvider.class);
      HiveSessionProperties hiveSessionProperties =
          injector.getInstance(HiveSessionProperties.class);
      HiveTableProperties hiveTableProperties = injector.getInstance(HiveTableProperties.class);
      HiveAnalyzeProperties hiveAnalyzeProperties =
          injector.getInstance(HiveAnalyzeProperties.class);
      ConnectorAccessControl accessControl = new ClassLoaderSafeConnectorAccessControl(
          new SystemTableAwareAccessControl(injector.getInstance(ConnectorAccessControl.class)),
          classLoader);
      Set<Procedure> procedures = injector.getInstance(Key.get(new TypeLiteral<Set<Procedure>>() {
      }));
      Set<SystemTable> systemTables =
          injector.getInstance(Key.get(new TypeLiteral<Set<SystemTable>>() {
          }));
      Set<EventListener> eventListeners =
          injector.getInstance(Key.get(new TypeLiteral<Set<EventListener>>() {
          })).stream().map(listener -> new ClassLoaderSafeEventListener(listener, classLoader))
              .collect(toImmutableSet());

      return new CarbonDataConnector(lifeCycleManager, metadataFactory, transactionManager,
          new ClassLoaderSafeConnectorSplitManager(splitManager, classLoader),
          new ClassLoaderSafeConnectorPageSourceProvider(connectorPageSource, classLoader),
          new ClassLoaderSafeConnectorPageSinkProvider(pageSinkProvider, classLoader),
          new ClassLoaderSafeNodePartitioningProvider(connectorDistributionProvider, classLoader),
          systemTables, procedures, eventListeners, hiveSessionProperties.getSessionProperties(),
          HiveSchemaProperties.SCHEMA_PROPERTIES, hiveTableProperties.getTableProperties(),
          hiveAnalyzeProperties.getAnalyzeProperties(), accessControl, classLoader);
    }
  }
}