/*
 * 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.tsfile.read.reader.page;

import org.apache.tsfile.encoding.decoder.Decoder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.header.PageHeader;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.common.BatchData;
import org.apache.tsfile.read.common.BatchDataFactory;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.TsBlockUtil;
import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.read.reader.IPageReader;
import org.apache.tsfile.read.reader.IPointReader;
import org.apache.tsfile.read.reader.series.PaginationController;
import org.apache.tsfile.utils.TsPrimitiveType;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

import static org.apache.tsfile.read.reader.series.PaginationController.UNLIMITED_PAGINATION_CONTROLLER;

public class AlignedPageReader implements IPageReader {

  private final TimePageReader timePageReader;
  private final List<ValuePageReader> valuePageReaderList;
  private final int valueCount;

  private final Filter globalTimeFilter;
  private Filter pushDownFilter;
  private PaginationController paginationController = UNLIMITED_PAGINATION_CONTROLLER;

  private boolean isModified;
  private TsBlockBuilder builder;

  private static final int MASK = 0x80;

  @SuppressWarnings("squid:S107")
  public AlignedPageReader(
      PageHeader timePageHeader,
      ByteBuffer timePageData,
      Decoder timeDecoder,
      List<PageHeader> valuePageHeaderList,
      List<ByteBuffer> valuePageDataList,
      List<TSDataType> valueDataTypeList,
      List<Decoder> valueDecoderList,
      Filter globalTimeFilter) {
    timePageReader = new TimePageReader(timePageHeader, timePageData, timeDecoder);
    isModified = timePageReader.isModified();
    valuePageReaderList = new ArrayList<>(valuePageHeaderList.size());
    for (int i = 0; i < valuePageHeaderList.size(); i++) {
      if (valuePageHeaderList.get(i) != null) {
        ValuePageReader valuePageReader =
            new ValuePageReader(
                valuePageHeaderList.get(i),
                valuePageDataList.get(i),
                valueDataTypeList.get(i),
                valueDecoderList.get(i));
        valuePageReaderList.add(valuePageReader);
        isModified = isModified || valuePageReader.isModified();
      } else {
        valuePageReaderList.add(null);
      }
    }
    this.globalTimeFilter = globalTimeFilter;
    this.valueCount = valuePageReaderList.size();
  }

  @SuppressWarnings("squid:S107")
  public AlignedPageReader(
      PageHeader timePageHeader,
      ByteBuffer timePageData,
      Decoder timeDecoder,
      List<PageHeader> valuePageHeaderList,
      // The reason for using Array here, rather than passing in
      // List<LazyLoadPageData> as a parameter, is that after type erasure, it would
      // conflict with the existing constructor.
      LazyLoadPageData[] lazyLoadPageDataArray,
      List<TSDataType> valueDataTypeList,
      List<Decoder> valueDecoderList,
      Filter globalTimeFilter) {
    timePageReader = new TimePageReader(timePageHeader, timePageData, timeDecoder);
    isModified = timePageReader.isModified();
    valuePageReaderList = new ArrayList<>(valuePageHeaderList.size());
    for (int i = 0; i < valuePageHeaderList.size(); i++) {
      if (valuePageHeaderList.get(i) != null) {
        ValuePageReader valuePageReader =
            new ValuePageReader(
                valuePageHeaderList.get(i),
                lazyLoadPageDataArray[i],
                valueDataTypeList.get(i),
                valueDecoderList.get(i));
        valuePageReaderList.add(valuePageReader);
        isModified = isModified || valuePageReader.isModified();
      } else {
        valuePageReaderList.add(null);
      }
    }
    this.globalTimeFilter = globalTimeFilter;
    this.valueCount = valuePageReaderList.size();
  }

  @Override
  public BatchData getAllSatisfiedPageData(boolean ascending) throws IOException {
    BatchData pageData = BatchDataFactory.createBatchData(TSDataType.VECTOR, ascending, false);
    int timeIndex = -1;
    Object[] rowValues = new Object[valueCount];
    while (timePageReader.hasNextTime()) {
      long timestamp = timePageReader.nextTime();
      timeIndex++;

      TsPrimitiveType[] v = new TsPrimitiveType[valueCount];
      // if all the sub sensors' value are null in current row, just discard it
      boolean hasNotNullValues = false;
      for (int i = 0; i < valueCount; i++) {
        ValuePageReader pageReader = valuePageReaderList.get(i);
        if (pageReader != null) {
          v[i] = pageReader.nextValue(timestamp, timeIndex);
          rowValues[i] = (v[i] == null) ? null : v[i].getValue();
        } else {
          v[i] = null;
          rowValues[i] = null;
        }
        if (rowValues[i] != null) {
          hasNotNullValues = true;
        }
      }

      if (hasNotNullValues && satisfyRecordFilter(timestamp, rowValues)) {
        pageData.putVector(timestamp, v);
      }
    }
    return pageData.flip();
  }

  private boolean satisfyRecordFilter(long timestamp, Object[] rowValues) {
    return (globalTimeFilter == null || globalTimeFilter.satisfyRow(timestamp, rowValues))
        && (pushDownFilter == null || pushDownFilter.satisfyRow(timestamp, rowValues));
  }

  @Override
  public boolean timeAllSelected() {
    for (int index = 0; index < getMeasurementCount(); index++) {
      if (!hasNullValue(index)) {
        // When there is any value page point number that is the same as the time page,
        // it means that all timestamps in time page will be selected.
        return true;
      }
    }
    return false;
  }

  @Override
  public int getMeasurementCount() {
    return valueCount;
  }

  public IPointReader getLazyPointReader() throws IOException {
    return new LazyLoadAlignedPagePointReader(timePageReader, valuePageReaderList);
  }

  private boolean allPageDataSatisfy() {
    return !isModified
        && timeAllSelected()
        && globalTimeFilterAllSatisfy()
        && pushDownFilterAllSatisfy();
  }

  private boolean globalTimeFilterAllSatisfy() {
    return globalTimeFilter == null || globalTimeFilter.allSatisfy(this);
  }

  private boolean pushDownFilterAllSatisfy() {
    return pushDownFilter == null || pushDownFilter.allSatisfy(this);
  }

  @Override
  public TsBlock getAllSatisfiedData() throws IOException {
    long[] timeBatch = timePageReader.getNextTimeBatch();

    if (allPageDataSatisfy()) {
      buildResultWithoutAnyFilterAndDelete(timeBatch);
      return builder.build();
    }

    // if all the sub sensors' value are null in current row, just discard it
    // if !filter.satisfy, discard this row
    boolean[] keepCurrentRow = new boolean[timeBatch.length];
    boolean globalTimeFilterAllSatisfy = globalTimeFilterAllSatisfy();
    if (globalTimeFilterAllSatisfy) {
      Arrays.fill(keepCurrentRow, true);
    } else {
      updateKeepCurrentRowThroughGlobalTimeFilter(keepCurrentRow, timeBatch);
    }

    boolean[][] isDeleted = null;
    if ((isModified || !timeAllSelected()) && valueCount != 0) {
      // using bitMap in valuePageReaders to indicate whether columns of current row are all null.
      byte[] bitmask = new byte[(timeBatch.length - 1) / 8 + 1];
      Arrays.fill(bitmask, (byte) 0x00);
      isDeleted = new boolean[valueCount][timeBatch.length];

      fillIsDeletedAndBitMask(timeBatch, isDeleted, bitmask);

      updateKeepCurrentRowThroughBitmask(keepCurrentRow, bitmask);
    }

    boolean pushDownFilterAllSatisfy = pushDownFilterAllSatisfy();

    // construct time column
    // when pushDownFilterAllSatisfy = true, we can skip rows by OFFSET & LIMIT
    int readEndIndex = buildTimeColumn(timeBatch, keepCurrentRow, pushDownFilterAllSatisfy);

    // construct value columns
    buildValueColumns(readEndIndex, keepCurrentRow, isDeleted);

    TsBlock unFilteredBlock = builder.build();
    if (pushDownFilterAllSatisfy) {
      // OFFSET & LIMIT has been consumed in buildTimeColumn
      return unFilteredBlock;
    }
    builder.reset();
    return TsBlockUtil.applyFilterAndLimitOffsetToTsBlock(
        unFilteredBlock, builder, pushDownFilter, paginationController);
  }

  private void buildResultWithoutAnyFilterAndDelete(long[] timeBatch) throws IOException {
    if (paginationController.hasCurOffset(timeBatch.length)) {
      paginationController.consumeOffset(timeBatch.length);
    } else {
      int readStartIndex = 0;
      if (paginationController.hasCurOffset()) {
        readStartIndex = (int) paginationController.getCurOffset();
        // consume the remaining offset
        paginationController.consumeOffset(readStartIndex);
      }

      // not included
      int readEndIndex = timeBatch.length;
      if (paginationController.hasCurLimit() && paginationController.getCurLimit() > 0) {
        readEndIndex =
            Math.min(readEndIndex, readStartIndex + (int) paginationController.getCurLimit());
        paginationController.consumeLimit((long) readEndIndex - readStartIndex);
      }

      // construct time column
      for (int i = readStartIndex; i < readEndIndex; i++) {
        builder.getTimeColumnBuilder().writeLong(timeBatch[i]);
        builder.declarePosition();
      }

      // construct value columns
      for (int i = 0; i < valueCount; i++) {
        ValuePageReader pageReader = valuePageReaderList.get(i);
        if (pageReader != null) {
          pageReader.writeColumnBuilderWithNextBatch(
              readStartIndex, readEndIndex, builder.getColumnBuilder(i));
        } else {
          builder.getColumnBuilder(i).appendNull(readEndIndex - readStartIndex);
        }
      }
    }
  }

  private int buildTimeColumn(
      long[] timeBatch, boolean[] keepCurrentRow, boolean pushDownFilterAllSatisfy) {
    if (pushDownFilterAllSatisfy) {
      return buildTimeColumnWithPagination(timeBatch, keepCurrentRow);
    } else {
      return buildTimeColumnWithoutPagination(timeBatch, keepCurrentRow);
    }
  }

  private int buildTimeColumnWithPagination(long[] timeBatch, boolean[] keepCurrentRow) {
    int readEndIndex = timeBatch.length;
    for (int rowIndex = 0; rowIndex < timeBatch.length; rowIndex++) {
      if (keepCurrentRow[rowIndex]) {
        if (paginationController.hasCurOffset()) {
          paginationController.consumeOffset();
          keepCurrentRow[rowIndex] = false;
        } else if (paginationController.hasCurLimit()) {
          builder.getTimeColumnBuilder().writeLong(timeBatch[rowIndex]);
          builder.declarePosition();
          paginationController.consumeLimit();
        } else {
          readEndIndex = rowIndex;
          break;
        }
      }
    }
    return readEndIndex;
  }

  private int buildTimeColumnWithoutPagination(long[] timeBatch, boolean[] keepCurrentRow) {
    int readEndIndex = 0;
    for (int i = 0; i < timeBatch.length; i++) {
      if (keepCurrentRow[i]) {
        builder.getTimeColumnBuilder().writeLong(timeBatch[i]);
        builder.declarePosition();
        readEndIndex = i;
      }
    }
    return readEndIndex + 1;
  }

  private void buildValueColumns(int readEndIndex, boolean[] keepCurrentRow, boolean[][] isDeleted)
      throws IOException {
    for (int i = 0; i < valueCount; i++) {
      ValuePageReader pageReader = valuePageReaderList.get(i);
      if (pageReader != null) {
        if (pageReader.isModified()) {
          pageReader.writeColumnBuilderWithNextBatch(
              readEndIndex,
              builder.getColumnBuilder(i),
              keepCurrentRow,
              Objects.requireNonNull(isDeleted)[i]);
        } else {
          pageReader.writeColumnBuilderWithNextBatch(
              readEndIndex, builder.getColumnBuilder(i), keepCurrentRow);
        }
      } else {
        for (int j = 0; j < readEndIndex; j++) {
          if (keepCurrentRow[j]) {
            builder.getColumnBuilder(i).appendNull();
          }
        }
      }
    }
  }

  private void fillIsDeletedAndBitMask(long[] timeBatch, boolean[][] isDeleted, byte[] bitmask)
      throws IOException {
    for (int columnIndex = 0; columnIndex < valueCount; columnIndex++) {
      ValuePageReader pageReader = valuePageReaderList.get(columnIndex);
      if (pageReader != null) {
        byte[] bitmap = pageReader.getBitmap();

        if (pageReader.isModified()) {
          pageReader.fillIsDeleted(timeBatch, isDeleted[columnIndex]);
          updateBitmapThroughIsDeleted(bitmap, isDeleted[columnIndex]);
        }

        for (int i = 0, n = bitmask.length; i < n; i++) {
          bitmask[i] = (byte) (bitmap[i] | bitmask[i]);
        }
      }
    }
  }

  private void updateBitmapThroughIsDeleted(byte[] bitmap, boolean[] isDeleted) {
    for (int i = 0, n = isDeleted.length; i < n; i++) {
      if (isDeleted[i]) {
        int shift = i % 8;
        bitmap[i / 8] = (byte) (bitmap[i / 8] & (~(MASK >>> shift)));
      }
    }
  }

  private void updateKeepCurrentRowThroughGlobalTimeFilter(
      boolean[] keepCurrentRow, long[] timeBatch) {
    for (int i = 0, n = timeBatch.length; i < n; i++) {
      keepCurrentRow[i] = globalTimeFilter.satisfy(timeBatch[i], null);
    }
  }

  private void updateKeepCurrentRowThroughBitmask(boolean[] keepCurrentRow, byte[] bitmask) {
    for (int i = 0, n = bitmask.length; i < n; i++) {
      if (bitmask[i] == (byte) 0xFF) {
        // 8 rows are not all null, do nothing
      } else if (bitmask[i] == (byte) 0x00) {
        Arrays.fill(keepCurrentRow, i * 8, Math.min(i * 8 + 8, keepCurrentRow.length), false);
      } else {
        for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) {
          if (((bitmask[i] & 0xFF) & (MASK >>> j)) == 0) {
            keepCurrentRow[i * 8 + j] = false;
          }
        }
      }
    }
  }

  public void setDeleteIntervalList(List<List<TimeRange>> list) {
    for (int i = 0; i < valueCount; i++) {
      if (valuePageReaderList.get(i) != null) {
        valuePageReaderList.get(i).setDeleteIntervalList(list.get(i));
      }
    }
  }

  @Override
  public Statistics<? extends Serializable> getStatistics() {
    return valuePageReaderList.size() == 1 && valuePageReaderList.get(0) != null
        ? valuePageReaderList.get(0).getStatistics()
        : timePageReader.getStatistics();
  }

  @Override
  public Statistics<? extends Serializable> getTimeStatistics() {
    return timePageReader.getStatistics();
  }

  @Override
  public Optional<Statistics<? extends Serializable>> getMeasurementStatistics(
      int measurementIndex) {
    ValuePageReader valuePageReader = valuePageReaderList.get(measurementIndex);
    return Optional.ofNullable(valuePageReader == null ? null : valuePageReader.getStatistics());
  }

  @Override
  public boolean hasNullValue(int measurementIndex) {
    long rowCount = getTimeStatistics().getCount();
    Optional<Statistics<? extends Serializable>> statistics =
        getMeasurementStatistics(measurementIndex);
    return statistics.map(stat -> stat.hasNullValue(rowCount)).orElse(true);
  }

  @Override
  public void addRecordFilter(Filter filter) {
    this.pushDownFilter = filter;
  }

  @Override
  public void setLimitOffset(PaginationController paginationController) {
    this.paginationController = paginationController;
  }

  @Override
  public boolean isModified() {
    return isModified;
  }

  @Override
  public void initTsBlockBuilder(List<TSDataType> dataTypes) {
    if (paginationController.hasLimit()) {
      builder =
          new TsBlockBuilder(
              (int)
                  Math.min(
                      paginationController.getCurLimit(),
                      timePageReader.getStatistics().getCount()),
              dataTypes);
    } else {
      builder = new TsBlockBuilder((int) timePageReader.getStatistics().getCount(), dataTypes);
    }
  }

  public TimePageReader getTimePageReader() {
    return timePageReader;
  }

  public List<ValuePageReader> getValuePageReaderList() {
    return valuePageReaderList;
  }
}
