/*
 * 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.spark.sql.secondaryindex.jobs;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CacheProvider;
import org.apache.carbondata.core.cache.CacheType;
import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder;
import org.apache.carbondata.core.index.IndexInputSplit;
import org.apache.carbondata.core.index.IndexStoreManager;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.index.dev.CacheableIndex;
import org.apache.carbondata.core.index.dev.IndexFactory;
import org.apache.carbondata.core.index.dev.expr.IndexExprWrapper;
import org.apache.carbondata.core.indexstore.BlockMetaInfo;
import org.apache.carbondata.core.indexstore.BlockletIndexStore;
import org.apache.carbondata.core.indexstore.BlockletIndexWrapper;
import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifier;
import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifierWrapper;
import org.apache.carbondata.core.indexstore.blockletindex.BlockletIndexInputSplit;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.readcommitter.ReadCommittedScope;
import org.apache.carbondata.core.util.BlockletIndexUtil;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.log4j.Logger;

/**
 * class to load blocklet index
 */
public class BlockletIndexInputFormat
    extends FileInputFormat<TableBlockIndexUniqueIdentifier, BlockletIndexDetailsWithSchema>
    implements Serializable {

  /**
   * Attribute for Carbon LOGGER.
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(BlockletIndexInputFormat.class.getName());

  private static final long serialVersionUID = 1;

  private CarbonTable table;

  private transient IndexExprWrapper indexExprWrapper;

  private transient List<Segment> validSegments;

  private transient Set<String> keys;

  private ReadCommittedScope readCommittedScope;

  public BlockletIndexInputFormat(CarbonTable table, IndexExprWrapper indexExprWrapper,
      List<Segment> validSegments) {
    this.table = table;
    this.indexExprWrapper = indexExprWrapper;
    this.validSegments = validSegments;
  }

  @Override
  public List<InputSplit> getSplits(JobContext job) throws IOException {
    IndexFactory indexFactory =
        IndexStoreManager.getInstance().getDefaultIndex(table).getIndexFactory();
    CacheableIndex factory = (CacheableIndex) indexFactory;
    List<IndexInputSplit> validDistributables =
        factory.getAllUncached(validSegments, indexExprWrapper);
    if (!validSegments.isEmpty()) {
      this.readCommittedScope = validSegments.get(0).getReadCommittedScope();
    }
    CarbonBlockLoaderHelper instance = CarbonBlockLoaderHelper.getInstance();
    int distributableSize = validDistributables.size();
    List<InputSplit> inputSplits = new ArrayList<>(distributableSize);
    keys = new HashSet<>();
    Iterator<IndexInputSplit> iterator = validDistributables.iterator();
    while (iterator.hasNext()) {
      BlockletIndexInputSplit next = (BlockletIndexInputSplit) iterator.next();
      String key = next.getSegmentPath();
      if (instance.checkAlreadySubmittedBlock(table.getAbsoluteTableIdentifier(), key)) {
        inputSplits.add(next);
        keys.add(key);
      }
    }
    int sizeOfDistToBeLoaded = inputSplits.size();
    LOGGER.info("Submitted blocks " + sizeOfDistToBeLoaded + ", " + distributableSize
        + " . Rest already considered for load in other job.");
    return inputSplits;
  }

  @Override
  public RecordReader<TableBlockIndexUniqueIdentifier,
      BlockletIndexDetailsWithSchema> createRecordReader(InputSplit inputSplit,
      TaskAttemptContext taskAttemptContext) {
    return new RecordReader<TableBlockIndexUniqueIdentifier, BlockletIndexDetailsWithSchema>() {
      private BlockletIndexWrapper wrapper = null;
      private TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier = null;
      private TableBlockIndexUniqueIdentifierWrapper tableBlockIndexUniqueIdentifierWrapper;
      Cache<TableBlockIndexUniqueIdentifierWrapper, BlockletIndexWrapper> cache =
          CacheProvider.getInstance().createCache(CacheType.DRIVER_BLOCKLET_INDEX);
      private Iterator<TableBlockIndexUniqueIdentifier> iterator;
      // Cache to avoid multiple times listing of files
      private Map<String, Map<String, BlockMetaInfo>> segInfoCache = new HashMap<>();

      @Override
      public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
          throws IOException, InterruptedException {
        BlockletIndexInputSplit segmentDistributable =
            (BlockletIndexInputSplit) inputSplit;
        TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier =
            segmentDistributable.getTableBlockIndexUniqueIdentifier();
        Segment segment =
            Segment.toSegment(tableSegmentUniqueIdentifier.getSegmentId(), readCommittedScope);
        iterator =
            BlockletIndexUtil.getTableBlockUniqueIdentifiers(segment).iterator();
      }

      @Override
      public boolean nextKeyValue() {
        if (iterator.hasNext()) {
          TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier = iterator.next();
          this.tableBlockIndexUniqueIdentifier  = tableBlockIndexUniqueIdentifier;
          TableBlockIndexUniqueIdentifierWrapper tableBlockIndexUniqueIdentifierWrapper =
              new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier, table,
                  false, true, true);
          this.tableBlockIndexUniqueIdentifierWrapper = tableBlockIndexUniqueIdentifierWrapper;
          wrapper = ((BlockletIndexStore) cache)
              .get(tableBlockIndexUniqueIdentifierWrapper, segInfoCache);
          return true;
        }
        return false;
      }

      @Override
      public TableBlockIndexUniqueIdentifier getCurrentKey() {
        return tableBlockIndexUniqueIdentifier;
      }

      @Override
      public BlockletIndexDetailsWithSchema getCurrentValue() {
        BlockletIndexDetailsWithSchema blockletIndexDetailsWithSchema =
            new BlockletIndexDetailsWithSchema(wrapper, table.getTableInfo().isSchemaModified());
        return blockletIndexDetailsWithSchema;
      }

      @Override
      public float getProgress() {
        return 0;
      }

      @Override
      public void close() {
        if (null != tableBlockIndexUniqueIdentifierWrapper) {
          if (null != wrapper && null != wrapper.getIndexes() && !wrapper.getIndexes()
              .isEmpty()) {
            String segmentId =
                tableBlockIndexUniqueIdentifierWrapper.getTableBlockIndexUniqueIdentifier()
                    .getSegmentId();
            // as segmentId will be same for all the indexes and segmentProperties cache is
            // maintained at segment level so it need to be called only once for clearing
            SegmentPropertiesAndSchemaHolder.getInstance()
                .invalidate(segmentId, wrapper.getIndexes().get(0).getSegmentPropertiesWrapper(),
                    tableBlockIndexUniqueIdentifierWrapper.isAddTableBlockToUnsafeAndLRUCache());
          }
        }
      }

    };
  }

  public void invalidate() {
    if (null != keys) {
      CarbonBlockLoaderHelper instance = CarbonBlockLoaderHelper.getInstance();
      instance.clear(table.getAbsoluteTableIdentifier(), keys);
    }
  }
}