/*
 * 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.drill.exec.physical.impl.metadata;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.Types;
import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.metastore.ColumnNamesOptions;
import org.apache.drill.exec.metastore.analyze.AnalyzeColumnUtils;
import org.apache.drill.exec.metastore.analyze.MetadataIdentifierUtils;
import org.apache.drill.exec.metastore.analyze.MetastoreAnalyzeConstants;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.MetadataControllerPOP;
import org.apache.drill.exec.physical.rowSet.DirectRowSet;
import org.apache.drill.exec.physical.rowSet.RowSetReader;
import org.apache.drill.exec.planner.common.DrillStatsTable;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.physical.WriterPrel;
import org.apache.drill.exec.record.AbstractBinaryRecordBatch;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.record.metadata.ColumnMetadata;
import org.apache.drill.exec.record.metadata.TupleMetadata;
import org.apache.drill.exec.store.EventBasedRecordWriter.FieldConverter;
import org.apache.drill.exec.store.StatisticsRecordCollector;
import org.apache.drill.exec.store.StatisticsRecordWriterImpl;
import org.apache.drill.exec.store.easy.json.StatisticsCollectorImpl;
import org.apache.drill.exec.store.parquet.ParquetTableMetadataUtils;
import org.apache.drill.exec.vector.BitVector;
import org.apache.drill.exec.vector.VarCharVector;
import org.apache.drill.exec.vector.accessor.ArrayReader;
import org.apache.drill.exec.vector.accessor.ObjectReader;
import org.apache.drill.exec.vector.accessor.ObjectType;
import org.apache.drill.exec.vector.accessor.TupleReader;
import org.apache.drill.exec.vector.complex.reader.FieldReader;
import org.apache.drill.metastore.components.tables.MetastoreTableInfo;
import org.apache.drill.metastore.components.tables.TableMetadataUnit;
import org.apache.drill.metastore.components.tables.Tables;
import org.apache.drill.metastore.expressions.FilterExpression;
import org.apache.drill.metastore.metadata.BaseMetadata;
import org.apache.drill.metastore.metadata.BaseTableMetadata;
import org.apache.drill.metastore.metadata.FileMetadata;
import org.apache.drill.metastore.metadata.MetadataInfo;
import org.apache.drill.metastore.metadata.MetadataType;
import org.apache.drill.metastore.metadata.PartitionMetadata;
import org.apache.drill.metastore.metadata.RowGroupMetadata;
import org.apache.drill.metastore.metadata.SegmentMetadata;
import org.apache.drill.metastore.metadata.TableInfo;
import org.apache.drill.metastore.operate.Modify;
import org.apache.drill.metastore.statistics.BaseStatisticsKind;
import org.apache.drill.metastore.statistics.ColumnStatistics;
import org.apache.drill.metastore.statistics.ColumnStatisticsKind;
import org.apache.drill.metastore.statistics.ExactStatisticsConstants;
import org.apache.drill.metastore.statistics.StatisticsHolder;
import org.apache.drill.metastore.statistics.StatisticsKind;
import org.apache.drill.metastore.statistics.TableStatisticsKind;
import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
import org.apache.drill.shaded.guava.com.google.common.collect.ArrayListMultimap;
import org.apache.drill.shaded.guava.com.google.common.collect.Multimap;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Terminal operator for producing ANALYZE statement. This operator is responsible for converting
 * obtained metadata, fetching absent metadata from the Metastore and storing resulting metadata into the Metastore.
 * <p>
 * This operator has two inputs: left input contains metadata and right input contains statistics metadata.
 */
public class MetadataControllerBatch extends AbstractBinaryRecordBatch<MetadataControllerPOP> {
  private static final Logger logger = LoggerFactory.getLogger(MetadataControllerBatch.class);

  private final Tables tables;
  private final TableInfo tableInfo;
  private final Map<String, MetadataInfo> metadataToHandle;
  private final StatisticsRecordCollector statisticsCollector;
  private final List<TableMetadataUnit> metadataUnits;
  private final ColumnNamesOptions columnNamesOptions;

  private boolean firstLeft = true;
  private boolean firstRight = true;
  private boolean finished;
  private boolean finishedRight;
  private int recordCount;

  protected MetadataControllerBatch(MetadataControllerPOP popConfig,
      FragmentContext context, RecordBatch left, RecordBatch right) throws OutOfMemoryException {
    super(popConfig, context, false, left, right);
    this.tables = context.getMetastoreRegistry().get().tables();
    this.tableInfo = popConfig.getContext().tableInfo();
    this.metadataToHandle = popConfig.getContext().metadataToHandle() == null
        ? null
        : popConfig.getContext().metadataToHandle().stream()
            .collect(Collectors.toMap(MetadataInfo::identifier, Function.identity()));
    this.metadataUnits = new ArrayList<>();
    this.statisticsCollector = new StatisticsCollectorImpl();
    this.columnNamesOptions = new ColumnNamesOptions(context.getOptions());
  }

  protected boolean setupNewSchema() {
    container.clear();
    container.addOrGet(MetastoreAnalyzeConstants.OK_FIELD_NAME, Types.required(TypeProtos.MinorType.BIT), null);
    container.addOrGet(MetastoreAnalyzeConstants.SUMMARY_FIELD_NAME, Types.required(TypeProtos.MinorType.VARCHAR), null);
    container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
    container.setEmpty();
    return true;
  }

  @Override
  public IterOutcome innerNext() {
    IterOutcome outcome;
    boolean finishedLeft;
    if (finished) {
      return IterOutcome.NONE;
    }

    if (!finishedRight) {
      outcome = handleRightIncoming();
      if (outcome != null) {
        return outcome;
      }
    }

    outer:
    while (true) {
      outcome = next(0, left);
      switch (outcome) {
        case NONE:
          // all incoming data was processed when returned OK_NEW_SCHEMA
          finishedLeft = !firstLeft;
          break outer;
        case OUT_OF_MEMORY:
        case NOT_YET:
        case STOP:
          return outcome;
        case OK_NEW_SCHEMA:
          if (firstLeft) {
            firstLeft = false;
            if (!setupNewSchema()) {
              outcome = IterOutcome.OK;
            }
            IterOutcome out = handleLeftIncoming();
            if (out != IterOutcome.OK) {
              return out;
            }
            return outcome;
          }
          //fall through
        case OK:
          assert !firstLeft : "First batch should be OK_NEW_SCHEMA";
          IterOutcome out = handleLeftIncoming();
          if (out != IterOutcome.OK) {
            return out;
          }
          break;
        default:
          context.getExecutorState()
              .fail(new UnsupportedOperationException("Unsupported upstream state " + outcome));
          close();
          killIncoming(false);
          return IterOutcome.STOP;
      }
    }

    if (finishedLeft) {
      IterOutcome out = writeToMetastore();
      finished = true;
      return out;
    }
    return outcome;
  }

  private IterOutcome handleRightIncoming() {
    IterOutcome outcome;
    outer:
    while (true) {
      outcome = next(0, right);
      switch (outcome) {
        case NONE:
          // all incoming data was processed
          finishedRight = true;
          break outer;
        case OUT_OF_MEMORY:
        case NOT_YET:
        case STOP:
          return outcome;
        case OK_NEW_SCHEMA:
          firstRight = false;
          //fall through
        case OK:
          assert !firstRight : "First batch should be OK_NEW_SCHEMA";
          try {
            appendStatistics(statisticsCollector);
          } catch (IOException e) {
            context.getExecutorState().fail(e);
            close();
            killIncoming(false);
            return IterOutcome.STOP;
          }
          break;
        default:
          context.getExecutorState()
              .fail(new UnsupportedOperationException("Unsupported upstream state " + outcome));
          close();
          killIncoming(false);
          return IterOutcome.STOP;
      }
    }
    return null;
  }

  private IterOutcome handleLeftIncoming() {
    try {
      metadataUnits.addAll(getMetadataUnits(left.getContainer()));
    } catch (Exception e) {
      context.getExecutorState().fail(e);
      close();
      killIncoming(false);
      return IterOutcome.STOP;
    }
    return IterOutcome.OK;
  }

  private IterOutcome writeToMetastore() {
    FilterExpression deleteFilter = popConfig.getContext().tableInfo().toFilter();

    for (MetadataInfo metadataInfo : popConfig.getContext().metadataToRemove()) {
      deleteFilter = FilterExpression.and(deleteFilter,
          FilterExpression.equal(MetadataInfo.METADATA_KEY, metadataInfo.key()));
    }

    Modify<TableMetadataUnit> modify = tables.modify();
    if (!popConfig.getContext().metadataToRemove().isEmpty()) {
      modify.delete(deleteFilter);
    }

    MetastoreTableInfo metastoreTableInfo = popConfig.getContext().metastoreTableInfo();

    if (tables.basicRequests().hasMetastoreTableInfoChanged(metastoreTableInfo)) {
      context.getExecutorState()
          .fail(new IllegalStateException(String.format("Metadata for table [%s] was changed before analyze is finished", tableInfo.name())));
      close();
      killIncoming(false);
      return IterOutcome.STOP;
    }

    modify.overwrite(metadataUnits)
        .execute();

    BitVector bitVector =
        container.addOrGet(MetastoreAnalyzeConstants.OK_FIELD_NAME, Types.required(TypeProtos.MinorType.BIT), null);
    VarCharVector varCharVector =
        container.addOrGet(MetastoreAnalyzeConstants.SUMMARY_FIELD_NAME, Types.required(TypeProtos.MinorType.VARCHAR), null);

    bitVector.allocateNew();
    varCharVector.allocateNew();

    bitVector.getMutator().set(0, 1);
    varCharVector.getMutator().setSafe(0,
        String.format("Collected / refreshed metadata for table [%s.%s.%s]",
            popConfig.getContext().tableInfo().storagePlugin(),
            popConfig.getContext().tableInfo().workspace(),
            popConfig.getContext().tableInfo().name()).getBytes());

    bitVector.getMutator().setValueCount(1);
    varCharVector.getMutator().setValueCount(1);
    container.setRecordCount(++recordCount);

    return IterOutcome.OK;
  }

  private List<TableMetadataUnit> getMetadataUnits(VectorContainer container) {
    List<TableMetadataUnit> metadataUnits = new ArrayList<>();
    RowSetReader reader = DirectRowSet.fromContainer(container).reader();
    while (reader.next()) {
      metadataUnits.addAll(getMetadataUnits(reader, 0));
    }

    if (!metadataToHandle.isEmpty()) {
      // leaves only table metadata and metadata which belongs to segments to be overridden
      metadataUnits = metadataUnits.stream()
          .filter(tableMetadataUnit ->
              metadataToHandle.values().stream()
                  .map(MetadataInfo::key)
                  .anyMatch(s -> s.equals(tableMetadataUnit.metadataKey()))
              || MetadataType.TABLE.name().equals(tableMetadataUnit.metadataType()))
          .collect(Collectors.toList());

      // leaves only metadata which should be fetched from the Metastore
      metadataUnits.stream()
          .map(TableMetadataUnit::metadataIdentifier)
          .forEach(metadataToHandle::remove);

      List<TableMetadataUnit> metadata = metadataToHandle.isEmpty()
          ? Collections.emptyList()
          : tables.basicRequests().metadata(popConfig.getContext().tableInfo(), metadataToHandle.values());

      metadataUnits.addAll(metadata);
    }

    boolean insertDefaultSegment = metadataUnits.stream()
        .noneMatch(metadataUnit -> metadataUnit.metadataType().equals(MetadataType.SEGMENT.name()));

    if (insertDefaultSegment) {
      TableMetadataUnit defaultSegmentMetadata = getDefaultSegment(metadataUnits);
      metadataUnits.add(defaultSegmentMetadata);
    }

    return metadataUnits;
  }

  private TableMetadataUnit getDefaultSegment(List<TableMetadataUnit> metadataUnits) {
    TableMetadataUnit tableMetadataUnit = metadataUnits.stream()
        .filter(metadataUnit -> metadataUnit.metadataType().equals(MetadataType.TABLE.name()))
        .findAny()
        .orElseThrow(() -> new IllegalStateException("Table metadata wasn't found among collected metadata."));

    List<String> paths = metadataUnits.stream()
        .filter(metadataUnit -> metadataUnit.metadataType().equals(MetadataType.FILE.name()))
        .map(TableMetadataUnit::path)
        .collect(Collectors.toList());

    return tableMetadataUnit.toBuilder()
        .metadataType(MetadataType.SEGMENT.name())
        .metadataKey(MetadataInfo.DEFAULT_SEGMENT_KEY)
        .owner(null)
        .tableType(null)
        .metadataStatistics(Collections.emptyList())
        .columnsStatistics(Collections.emptyMap())
        .path(tableMetadataUnit.location())
        .schema(null)
        .locations(paths)
        .build();
  }

  @SuppressWarnings("rawtypes")
  private List<TableMetadataUnit> getMetadataUnits(TupleReader reader, int nestingLevel) {
    List<TableMetadataUnit> metadataUnits = new ArrayList<>();

    TupleMetadata columnMetadata = reader.tupleSchema();
    ObjectReader metadataColumnReader = reader.column(MetastoreAnalyzeConstants.METADATA_TYPE);
    Preconditions.checkNotNull(metadataColumnReader, "metadataType column wasn't found");

    ObjectReader underlyingMetadataReader = reader.column(MetastoreAnalyzeConstants.COLLECTED_MAP_FIELD);
    if (underlyingMetadataReader != null) {
      if (!underlyingMetadataReader.schema().isArray()) {
        throw new IllegalStateException("Incoming vector with name `collected_map` should be repeated map");
      }
      // current row contains information about underlying metadata
      ArrayReader array = underlyingMetadataReader.array();
      while (array.next()) {
        metadataUnits.addAll(getMetadataUnits(array.tuple(), nestingLevel + 1));
      }
    }

    List<StatisticsHolder> metadataStatistics = getMetadataStatistics(reader, columnMetadata);

    Long rowCount = (Long) metadataStatistics.stream()
        .filter(statisticsHolder -> statisticsHolder.getStatisticsKind() == TableStatisticsKind.ROW_COUNT)
        .findAny()
        .map(StatisticsHolder::getStatisticsValue)
        .orElse(null);

    Map<SchemaPath, ColumnStatistics> columnStatistics = getColumnStatistics(reader, columnMetadata, rowCount);

    MetadataType metadataType = MetadataType.valueOf(metadataColumnReader.scalar().getString());

    BaseMetadata metadata;

    switch (metadataType) {
      case TABLE: {
        metadata = getTableMetadata(reader, metadataStatistics, columnStatistics);
        break;
      }
      case SEGMENT: {
        metadata = getSegmentMetadata(reader, metadataStatistics, columnStatistics, nestingLevel);
        break;
      }
      case PARTITION: {
        metadata = getPartitionMetadata(reader, metadataStatistics, columnStatistics, nestingLevel);
        break;
      }
      case FILE: {
        metadata = getFileMetadata(reader, metadataStatistics, columnStatistics, nestingLevel);
        break;
      }
      case ROW_GROUP: {
        metadata = getRowGroupMetadata(reader, metadataStatistics, columnStatistics, nestingLevel);
        break;
      }
      default:
        throw new UnsupportedOperationException("Unsupported metadata type: " + metadataType);
    }
    metadataUnits.add(metadata.toMetadataUnit());

    return metadataUnits;
  }

  @SuppressWarnings("rawtypes")
  private PartitionMetadata getPartitionMetadata(TupleReader reader, List<StatisticsHolder> metadataStatistics,
      Map<SchemaPath, ColumnStatistics> columnStatistics, int nestingLevel) {
    List<String> segmentColumns = popConfig.getContext().segmentColumns();

    String segmentKey = segmentColumns.size() > 0
        ? reader.column(segmentColumns.iterator().next()).scalar().getString()
        : MetadataInfo.DEFAULT_SEGMENT_KEY;

    List<String> partitionValues = segmentColumns.stream()
        .limit(nestingLevel)
        .map(columnName -> reader.column(columnName).scalar().getString())
        .collect(Collectors.toList());
    String metadataIdentifier = MetadataIdentifierUtils.getMetadataIdentifierKey(partitionValues);
    MetadataInfo metadataInfo = MetadataInfo.builder()
        .type(MetadataType.PARTITION)
        .key(segmentKey)
        .identifier(StringUtils.defaultIfEmpty(metadataIdentifier, null))
        .build();
    return PartitionMetadata.builder()
        .tableInfo(tableInfo)
        .metadataInfo(metadataInfo)
        .columnsStatistics(columnStatistics)
        .metadataStatistics(metadataStatistics)
        .locations(getIncomingLocations(reader))
        .lastModifiedTime(Long.parseLong(reader.column(columnNamesOptions.lastModifiedTime()).scalar().getString()))
//            .column(SchemaPath.getSimplePath("dir1"))
//            .partitionValues()
        .schema(TupleMetadata.of(reader.column(MetastoreAnalyzeConstants.SCHEMA_FIELD).scalar().getString()))
        .build();
  }

  @SuppressWarnings("rawtypes")
  private BaseTableMetadata getTableMetadata(TupleReader reader, List<StatisticsHolder> metadataStatistics,
      Map<SchemaPath, ColumnStatistics> columnStatistics) {
    List<StatisticsHolder> updatedMetaStats = new ArrayList<>(metadataStatistics);
    updatedMetaStats.add(new StatisticsHolder<>(popConfig.getContext().analyzeMetadataLevel(), TableStatisticsKind.ANALYZE_METADATA_LEVEL));

    MetadataInfo metadataInfo = MetadataInfo.builder()
        .type(MetadataType.TABLE)
        .key(MetadataInfo.GENERAL_INFO_KEY)
        .build();

    BaseTableMetadata tableMetadata = BaseTableMetadata.builder()
        .tableInfo(tableInfo)
        .metadataInfo(metadataInfo)
        .columnsStatistics(columnStatistics)
        .metadataStatistics(updatedMetaStats)
        .partitionKeys(Collections.emptyMap())
        .interestingColumns(popConfig.getContext().interestingColumns())
        .location(popConfig.getContext().location())
        .lastModifiedTime(Long.parseLong(reader.column(columnNamesOptions.lastModifiedTime()).scalar().getString()))
        .schema(TupleMetadata.of(reader.column(MetastoreAnalyzeConstants.SCHEMA_FIELD).scalar().getString()))
        .build();

    if (context.getOptions().getOption(PlannerSettings.STATISTICS_USE)) {
      DrillStatsTable statistics = new DrillStatsTable(statisticsCollector.getStatistics());
      Map<SchemaPath, ColumnStatistics> tableColumnStatistics =
          ParquetTableMetadataUtils.getColumnStatistics(tableMetadata.getSchema(), statistics);
      tableMetadata = tableMetadata.cloneWithStats(tableColumnStatistics, DrillStatsTable.getEstimatedTableStats(statistics));
    }

    return tableMetadata;
  }

  @SuppressWarnings("rawtypes")
  private SegmentMetadata getSegmentMetadata(TupleReader reader, List<StatisticsHolder> metadataStatistics,
      Map<SchemaPath, ColumnStatistics> columnStatistics, int nestingLevel) {
    List<String> segmentColumns = popConfig.getContext().segmentColumns();

    String segmentKey = segmentColumns.size() > 0
        ? reader.column(segmentColumns.iterator().next()).scalar().getString()
        : MetadataInfo.DEFAULT_SEGMENT_KEY;

    // for the case of multi-value segments, there is no nesting
    // and therefore all values should be used when forming metadata identifier
    if (popConfig.getContext().multiValueSegments()) {
      nestingLevel = segmentColumns.size();
    }

    List<String> allPartitionValues = segmentColumns.stream()
        .limit(nestingLevel)
        .map(columnName -> reader.column(columnName).scalar().getString())
        .collect(Collectors.toList());
    String metadataIdentifier = MetadataIdentifierUtils.getMetadataIdentifierKey(allPartitionValues);

    MetadataInfo metadataInfo = MetadataInfo.builder()
        .type(MetadataType.SEGMENT)
        .key(segmentKey)
        .identifier(StringUtils.defaultIfEmpty(metadataIdentifier, null))
        .build();

    int segmentLevel = nestingLevel - 1;

    // for the case of multi-value segments, there is no nesting,
    // so all partition column values should be used
    List<String> partitionValues = popConfig.getContext().multiValueSegments()
        ? allPartitionValues
        : Collections.singletonList(allPartitionValues.get(segmentLevel));

    return SegmentMetadata.builder()
        .tableInfo(tableInfo)
        .metadataInfo(metadataInfo)
        .columnsStatistics(columnStatistics)
        .metadataStatistics(metadataStatistics)
        .path(new Path(reader.column(MetastoreAnalyzeConstants.LOCATION_FIELD).scalar().getString()))
        .locations(getIncomingLocations(reader))
        .column(segmentColumns.size() > 0 ? SchemaPath.getSimplePath(segmentColumns.get(segmentLevel)) : null)
        .partitionValues(partitionValues)
        .lastModifiedTime(Long.parseLong(reader.column(columnNamesOptions.lastModifiedTime()).scalar().getString()))
        .schema(TupleMetadata.of(reader.column(MetastoreAnalyzeConstants.SCHEMA_FIELD).scalar().getString()))
        .build();
  }

  @SuppressWarnings("rawtypes")
  private FileMetadata getFileMetadata(TupleReader reader, List<StatisticsHolder> metadataStatistics,
      Map<SchemaPath, ColumnStatistics> columnStatistics, int nestingLevel) {
    List<String> segmentColumns = popConfig.getContext().segmentColumns();

    String segmentKey = segmentColumns.size() > 0
        ? reader.column(segmentColumns.iterator().next()).scalar().getString()
        : MetadataInfo.DEFAULT_SEGMENT_KEY;

    List<String> partitionValues = segmentColumns.stream()
        .limit(nestingLevel - 1)
        .map(columnName -> reader.column(columnName).scalar().getString())
        .collect(Collectors.toList());

    Path path = new Path(reader.column(MetastoreAnalyzeConstants.LOCATION_FIELD).scalar().getString());
    String metadataIdentifier = MetadataIdentifierUtils.getFileMetadataIdentifier(partitionValues, path);

    MetadataInfo metadataInfo = MetadataInfo.builder()
        .type(MetadataType.FILE)
        .key(segmentKey)
        .identifier(StringUtils.defaultIfEmpty(metadataIdentifier, null))
        .build();

    return FileMetadata.builder()
        .tableInfo(tableInfo)
        .metadataInfo(metadataInfo)
        .columnsStatistics(columnStatistics)
        .metadataStatistics(metadataStatistics)
        .path(path)
        .lastModifiedTime(Long.parseLong(reader.column(columnNamesOptions.lastModifiedTime()).scalar().getString()))
        .schema(TupleMetadata.of(reader.column(MetastoreAnalyzeConstants.SCHEMA_FIELD).scalar().getString()))
        .build();
  }

  @SuppressWarnings("rawtypes")
  private RowGroupMetadata getRowGroupMetadata(TupleReader reader,List<StatisticsHolder> metadataStatistics,
      Map<SchemaPath, ColumnStatistics> columnStatistics, int nestingLevel) {

    List<String> segmentColumns = popConfig.getContext().segmentColumns();
    String segmentKey = segmentColumns.size() > 0
        ? reader.column(segmentColumns.iterator().next()).scalar().getString()
        : MetadataInfo.DEFAULT_SEGMENT_KEY;

    List<String> partitionValues = segmentColumns.stream()
        .limit(nestingLevel - 2)
        .map(columnName -> reader.column(columnName).scalar().getString())
        .collect(Collectors.toList());

    Path path = new Path(reader.column(MetastoreAnalyzeConstants.LOCATION_FIELD).scalar().getString());

    int rowGroupIndex = Integer.parseInt(reader.column(columnNamesOptions.rowGroupIndex()).scalar().getString());

    String metadataIdentifier = MetadataIdentifierUtils.getRowGroupMetadataIdentifier(partitionValues, path, rowGroupIndex);

    MetadataInfo metadataInfo = MetadataInfo.builder()
        .type(MetadataType.ROW_GROUP)
        .key(segmentKey)
        .identifier(StringUtils.defaultIfEmpty(metadataIdentifier, null))
        .build();

    return RowGroupMetadata.builder()
        .tableInfo(tableInfo)
        .metadataInfo(metadataInfo)
        .columnsStatistics(columnStatistics)
        .metadataStatistics(metadataStatistics)
        .hostAffinity(Collections.emptyMap())
        .rowGroupIndex(rowGroupIndex)
        .path(path)
        .lastModifiedTime(Long.parseLong(reader.column(columnNamesOptions.lastModifiedTime()).scalar().getString()))
        .schema(TupleMetadata.of(reader.column(MetastoreAnalyzeConstants.SCHEMA_FIELD).scalar().getString()))
        .build();
  }

  @SuppressWarnings("rawtypes")
  private Map<SchemaPath, ColumnStatistics> getColumnStatistics(TupleReader reader, TupleMetadata columnMetadata, Long rowCount) {
    Multimap<String, StatisticsHolder> columnStatistics = ArrayListMultimap.create();
    Map<String, TypeProtos.MinorType> columnTypes = new HashMap<>();
    for (ColumnMetadata column : columnMetadata) {
      String fieldName = AnalyzeColumnUtils.getColumnName(column.name());

      if (AnalyzeColumnUtils.isColumnStatisticsField(column.name())) {
        StatisticsKind<?> statisticsKind = AnalyzeColumnUtils.getStatisticsKind(column.name());
        columnStatistics.put(fieldName,
            new StatisticsHolder<>(getConvertedColumnValue(reader.column(column.name())), statisticsKind));
        if (statisticsKind.getName().equalsIgnoreCase(ColumnStatisticsKind.MIN_VALUE.getName())
            || statisticsKind.getName().equalsIgnoreCase(ColumnStatisticsKind.MAX_VALUE.getName())) {
          columnTypes.putIfAbsent(fieldName, column.type());
        }
      }
    }

    // adds NON_NULL_COUNT to use it during filter pushdown
    if (rowCount != null) {
      Map<String, StatisticsHolder> nullsCountColumnStatistics = new HashMap<>();
      columnStatistics.asMap().forEach((key, value) ->
          value.stream()
              .filter(statisticsHolder -> statisticsHolder.getStatisticsKind() == ColumnStatisticsKind.NON_NULL_VALUES_COUNT)
              .findAny()
              .map(statisticsHolder -> (Long) statisticsHolder.getStatisticsValue())
              .ifPresent(nonNullCount ->
                  nullsCountColumnStatistics.put(
                      key,
                      new StatisticsHolder<>(rowCount - nonNullCount, ColumnStatisticsKind.NULLS_COUNT))));

      nullsCountColumnStatistics.forEach(columnStatistics::put);
    }

    Map<SchemaPath, ColumnStatistics> resultingStats = new HashMap<>();

    columnStatistics.asMap().forEach((fieldName, statisticsHolders) ->
        resultingStats.put(SchemaPath.parseFromString(fieldName), new ColumnStatistics<>(statisticsHolders, columnTypes.get(fieldName))));
    return resultingStats;
  }

  @SuppressWarnings("rawtypes")
  private List<StatisticsHolder> getMetadataStatistics(TupleReader reader, TupleMetadata columnMetadata) {
    List<StatisticsHolder> metadataStatistics = new ArrayList<>();
    String rgs = columnNamesOptions.rowGroupStart();
    String rgl = columnNamesOptions.rowGroupLength();
    for (ColumnMetadata column : columnMetadata) {
      String columnName = column.name();
      ObjectReader objectReader = reader.column(columnName);
      if (AnalyzeColumnUtils.isMetadataStatisticsField(columnName)) {
        metadataStatistics.add(new StatisticsHolder<>(objectReader.getObject(),
            AnalyzeColumnUtils.getStatisticsKind(columnName)));
      } else if (!objectReader.isNull()) {
        if (columnName.equals(rgs)) {
          metadataStatistics.add(new StatisticsHolder<>(Long.parseLong(objectReader.scalar().getString()),
              new BaseStatisticsKind<>(ExactStatisticsConstants.START, true)));
        } else if (columnName.equals(rgl)) {
          metadataStatistics.add(new StatisticsHolder<>(Long.parseLong(objectReader.scalar().getString()),
              new BaseStatisticsKind<>(ExactStatisticsConstants.LENGTH, true)));
        }
      }
    }
    return metadataStatistics;
  }

  private void appendStatistics(StatisticsRecordCollector statisticsCollector) throws IOException {
    if (context.getOptions().getOption(PlannerSettings.STATISTICS_USE)) {
      List<FieldConverter> fieldConverters = new ArrayList<>();
      int fieldId = 0;

      for (VectorWrapper<?> wrapper : right) {
        if (wrapper.getField().getName().equalsIgnoreCase(WriterPrel.PARTITION_COMPARATOR_FIELD)) {
          continue;
        }
        FieldReader reader = wrapper.getValueVector().getReader();
        FieldConverter converter =
            StatisticsRecordWriterImpl.getConverter(statisticsCollector, fieldId++, wrapper.getField().getName(), reader);
        fieldConverters.add(converter);
      }

      for (int counter = 0; counter < right.getRecordCount(); counter++) {
        statisticsCollector.startStatisticsRecord();
        // write the current record
        for (FieldConverter converter : fieldConverters) {
          converter.setPosition(counter);
          converter.startField();
          converter.writeField();
          converter.endField();
        }
        statisticsCollector.endStatisticsRecord();
      }
    }
  }

  private Object getConvertedColumnValue(ObjectReader objectReader) {
    switch (objectReader.schema().type()) {
      case VARBINARY:
      case FIXEDBINARY:
        return new String(objectReader.scalar().getBytes());
      default:
        return objectReader.getObject();
    }

  }

  private Set<Path> getIncomingLocations(TupleReader reader) {
    Set<Path> childLocations = new HashSet<>();

    ObjectReader metadataColumnReader = reader.column(MetastoreAnalyzeConstants.METADATA_TYPE);
    Preconditions.checkNotNull(metadataColumnReader, "metadataType column wasn't found");

    MetadataType metadataType = MetadataType.valueOf(metadataColumnReader.scalar().getString());

    switch (metadataType) {
      case SEGMENT:
      case PARTITION: {
        ObjectReader locationsReader = reader.column(MetastoreAnalyzeConstants.LOCATIONS_FIELD);
        // populate list of file locations from "locations" field if it is present in the schema
        if (locationsReader != null && locationsReader.type() == ObjectType.ARRAY) {
          ArrayReader array = locationsReader.array();
          while (array.next()) {
            childLocations.add(new Path(array.scalar().getString()));
          }
          break;
        }
        // in the opposite case, populate list of file locations using underlying metadata
        ObjectReader underlyingMetadataReader = reader.column(MetastoreAnalyzeConstants.COLLECTED_MAP_FIELD);
        if (underlyingMetadataReader != null) {
          // current row contains information about underlying metadata
          ArrayReader array = underlyingMetadataReader.array();
          array.rewind();
          while (array.next()) {
            childLocations.addAll(getIncomingLocations(array.tuple()));
          }
        }
        break;
      }
      case FILE: {
        childLocations.add(new Path(reader.column(MetastoreAnalyzeConstants.LOCATION_FIELD).scalar().getString()));
      }
    }

    return childLocations;
  }

  @Override
  protected void killIncoming(boolean sendUpstream) {
    left.kill(sendUpstream);
    right.kill(sendUpstream);
  }

  @Override
  public void dump() {
    logger.error("MetadataHandlerBatch[container={}, popConfig={}]", container, popConfig);
  }

  @Override
  public int getRecordCount() {
    return recordCount;
  }
}