/*
 * 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 java.io.IOException;
import java.util.Collection;
import java.util.Optional;
import java.util.Properties;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
import org.apache.carbondata.hadoop.api.CarbonTableOutputFormat;
import org.apache.carbondata.hive.MapredCarbonOutputCommitter;
import org.apache.carbondata.hive.util.HiveCarbonUtil;
import org.apache.carbondata.presto.impl.CarbonTableConfig;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;

import com.google.common.collect.ImmutableMap;
import io.airlift.json.JsonCodec;
import io.airlift.slice.Slice;
import io.prestosql.plugin.base.CatalogName;
import io.prestosql.plugin.hive.HdfsEnvironment;
import io.prestosql.plugin.hive.HiveInsertTableHandle;
import io.prestosql.plugin.hive.HiveMetadata;
import io.prestosql.plugin.hive.HivePartitionManager;
import io.prestosql.plugin.hive.LocationService;
import io.prestosql.plugin.hive.PartitionUpdate;
import io.prestosql.plugin.hive.TypeTranslator;
import io.prestosql.plugin.hive.authentication.HiveIdentity;
import io.prestosql.plugin.hive.metastore.MetastoreUtil;
import io.prestosql.plugin.hive.metastore.SemiTransactionalHiveMetastore;
import io.prestosql.plugin.hive.metastore.Table;
import io.prestosql.plugin.hive.security.AccessControlMetadata;
import io.prestosql.plugin.hive.statistics.HiveStatisticsProvider;
import io.prestosql.plugin.hive.util.ConfigurationUtils;
import io.prestosql.spi.connector.ConnectorInsertTableHandle;
import io.prestosql.spi.connector.ConnectorOutputMetadata;
import io.prestosql.spi.connector.ConnectorSession;
import io.prestosql.spi.connector.ConnectorTableHandle;
import io.prestosql.spi.connector.SchemaTableName;
import io.prestosql.spi.statistics.ComputedStatistics;
import io.prestosql.spi.type.TypeManager;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobContextImpl;
import org.apache.hadoop.mapred.JobID;

import org.apache.log4j.Logger;
import org.joda.time.DateTimeZone;

public class CarbonDataMetaData extends HiveMetadata {

  private static final Logger LOG =
      LogServiceFactory.getLogService(CarbonDataMetaData.class.getName());

  private HdfsEnvironment hdfsEnvironment;
  private SemiTransactionalHiveMetastore metastore;
  private MapredCarbonOutputCommitter carbonOutputCommitter;
  private JobContextImpl jobContext;

  public CarbonDataMetaData(
      CatalogName catalogName,
      SemiTransactionalHiveMetastore metastore,
      HdfsEnvironment hdfsEnvironment,
      HivePartitionManager partitionManager,
      DateTimeZone timeZone,
      boolean allowCorruptWritesForTesting,
      boolean writesToNonManagedTablesEnabled,
      boolean createsOfNonManagedTablesEnabled,
      boolean translateHiveViews, TypeManager typeManager,
      LocationService locationService,
      JsonCodec<PartitionUpdate> partitionUpdateCodec,
      TypeTranslator typeTranslator,
      String prestoVersion,
      HiveStatisticsProvider hiveStatisticsProvider,
      AccessControlMetadata accessControlMetadata) {
    super(catalogName, metastore, hdfsEnvironment, partitionManager, timeZone,
        allowCorruptWritesForTesting, writesToNonManagedTablesEnabled,
        createsOfNonManagedTablesEnabled, translateHiveViews, typeManager, locationService,
        partitionUpdateCodec, typeTranslator, prestoVersion, hiveStatisticsProvider,
        accessControlMetadata);
    this.hdfsEnvironment = hdfsEnvironment;
    this.metastore = metastore;
  }

  @Override
  public CarbonDataInsertTableHandle beginInsert(ConnectorSession session,
      ConnectorTableHandle tableHandle) {
    HiveInsertTableHandle hiveInsertTableHandle = super.beginInsert(session, tableHandle);
    SchemaTableName tableName = hiveInsertTableHandle.getSchemaTableName();
    Optional<Table> table = this.metastore
        .getTable(new HiveIdentity(session), tableName.getSchemaName(), tableName.getTableName());
    Path outputPath =
        new Path(hiveInsertTableHandle.getLocationHandle().getJsonSerializableTargetPath());
    JobConf jobConf = ConfigurationUtils.toJobConf(this.hdfsEnvironment
        .getConfiguration(
            new HdfsEnvironment.HdfsContext(session, hiveInsertTableHandle.getSchemaName(),
                hiveInsertTableHandle.getTableName()),
            new Path(hiveInsertTableHandle.getLocationHandle().getJsonSerializableWritePath())));
    jobConf.set("location", outputPath.toString());
    Properties hiveSchema = MetastoreUtil.getHiveSchema(table.get());
    try {
      CarbonLoadModel carbonLoadModel =
          HiveCarbonUtil.getCarbonLoadModel(hiveSchema, jobConf);

      CarbonTableOutputFormat.setLoadModel(jobConf, carbonLoadModel);
    } catch (IOException ex) {
      LOG.error("Error while creating carbon load model", ex);
      throw new RuntimeException(ex);
    }
    try {
      carbonOutputCommitter = new MapredCarbonOutputCommitter();
      jobContext = new JobContextImpl(jobConf, new JobID());
      carbonOutputCommitter.setupJob(jobContext);
      ThreadLocalSessionInfo.setConfigurationToCurrentThread(jobConf);
    } catch (IOException e) {
      LOG.error("error setting the output committer", e);
      throw new RuntimeException("error setting the output committer");
    }
    return new CarbonDataInsertTableHandle(hiveInsertTableHandle.getSchemaTableName().getSchemaName(),
        hiveInsertTableHandle.getTableName(),
        hiveInsertTableHandle.getInputColumns(),
        hiveInsertTableHandle.getPageSinkMetadata(),
        hiveInsertTableHandle.getLocationHandle(),
        hiveInsertTableHandle.getBucketProperty(), hiveInsertTableHandle.getTableStorageFormat(),
        hiveInsertTableHandle.getPartitionStorageFormat(),
        ImmutableMap.of(CarbonTableConfig.CARBON_PRESTO_LOAD_MODEL,
            jobContext.getConfiguration().get(CarbonTableOutputFormat.LOAD_MODEL)));
  }

  @Override
  public Optional<ConnectorOutputMetadata> finishInsert(ConnectorSession session,
      ConnectorInsertTableHandle insertHandle, Collection<Slice> fragments,
      Collection<ComputedStatistics> computedStatistics) {
    Optional<ConnectorOutputMetadata> connectorOutputMetadata =
        super.finishInsert(session, insertHandle, fragments, computedStatistics);
    try {
      carbonOutputCommitter.commitJob(jobContext);
    } catch (IOException e) {
      LOG.error("Error occurred while committing the insert job.", e);
      throw new RuntimeException(e);
    }
    return connectorOutputMetadata;
  }
}