/*
 * 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.iotdb.db.queryengine.plan.analyze.schema;

import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.AlignedTimeseriesException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

class NormalSchemaFetcher {

  private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();

  private final DataNodeSchemaCache schemaCache;

  private final AutoCreateSchemaExecutor autoCreateSchemaExecutor;
  private final ClusterSchemaFetchExecutor clusterSchemaFetchExecutor;

  NormalSchemaFetcher(
      DataNodeSchemaCache schemaCache,
      AutoCreateSchemaExecutor autoCreateSchemaExecutor,
      ClusterSchemaFetchExecutor clusterSchemaFetchExecutor) {
    this.schemaCache = schemaCache;
    this.autoCreateSchemaExecutor = autoCreateSchemaExecutor;
    this.clusterSchemaFetchExecutor = clusterSchemaFetchExecutor;
  }

  /** Given full paths will be */
  private void computePatternTreeNeededReFetch(
      PathPatternTree patternTree, List<String> fullPaths) {
    for (String fullPath : fullPaths) {
      try {
        patternTree.appendFullPath(new PartialPath(fullPath));
      } catch (IllegalPathException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private PathPatternTree computePatternTreeNeededReFetch(
      PartialPath devicePath,
      String[] measurementList,
      List<Integer> indexOfMissingMeasurements,
      List<String> fullPaths) {
    PathPatternTree patternTree = new PathPatternTree();
    for (int index : indexOfMissingMeasurements) {
      patternTree.appendFullPath(devicePath, measurementList[index]);
    }
    for (String fullPath : fullPaths) {
      try {
        patternTree.appendFullPath(new PartialPath(fullPath));
      } catch (IllegalPathException e) {
        throw new RuntimeException(e);
      }
    }
    return patternTree;
  }

  private PathPatternTree computePatternTreeNeededReFetch(
      List<PartialPath> devicePathList,
      List<String[]> measurementsList,
      List<Integer> indexOfTargetDevices,
      List<List<Integer>> indexOfTargetMeasurementsList) {
    PathPatternTree patternTree = new PathPatternTree();
    int deviceIndex;
    for (int i = 0, size = indexOfTargetDevices.size(); i < size; i++) {
      deviceIndex = indexOfTargetDevices.get(i);
      for (int measurementIndex : indexOfTargetMeasurementsList.get(i)) {
        patternTree.appendFullPath(
            devicePathList.get(deviceIndex), measurementsList.get(deviceIndex)[measurementIndex]);
      }
    }
    return patternTree;
  }

  List<Integer> processNormalTimeSeries(
      ISchemaComputationWithAutoCreation schemaComputationWithAutoCreation,
      MPPQueryContext context) {
    // [Step 0] Record the input value.
    boolean isAlignedPutIn = schemaComputationWithAutoCreation.isAligned();

    // [Step 1] Cache 1. compute measurements and record logical views.
    List<Integer> indexOfMissingMeasurements =
        schemaCache.computeWithoutTemplate(schemaComputationWithAutoCreation);
    // [Step 2] Cache 2. process recorded logical views. If there is no views now, it returns empty
    // lists.
    Pair<List<Integer>, List<String>> missedIndexAndPathString =
        schemaCache.computeSourceOfLogicalView(schemaComputationWithAutoCreation);
    List<Integer> indexOfMissingLogicalView = missedIndexAndPathString.left;
    List<String> missedPathStringOfLogicalView = missedIndexAndPathString.right;
    // all schema can be taken from cache
    if (indexOfMissingMeasurements.isEmpty() && indexOfMissingLogicalView.isEmpty()) {
      return indexOfMissingMeasurements;
    }
    // [Step 3] Fetch 1. fetch schema from remote. Process logical view first; then process
    // measurements.
    // try fetch the missing raw schema from remote and cache fetched schema
    ClusterSchemaTree remoteSchemaTree;
    if (missedPathStringOfLogicalView.isEmpty()) {
      remoteSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaOfOneDevice(
              schemaComputationWithAutoCreation.getDevicePath(),
              schemaComputationWithAutoCreation.getMeasurements(),
              indexOfMissingMeasurements,
              context);
    } else {
      PathPatternTree patternTree =
          computePatternTreeNeededReFetch(
              schemaComputationWithAutoCreation.getDevicePath(),
              schemaComputationWithAutoCreation.getMeasurements(),
              indexOfMissingMeasurements,
              missedPathStringOfLogicalView);
      remoteSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaWithPatternTreeAndCache(patternTree, context);
    }
    // make sure all missed views are computed.
    remoteSchemaTree.computeSourceOfLogicalView(
        schemaComputationWithAutoCreation, indexOfMissingLogicalView);
    // check and compute the fetched schema
    indexOfMissingMeasurements =
        remoteSchemaTree.compute(schemaComputationWithAutoCreation, indexOfMissingMeasurements);
    schemaComputationWithAutoCreation.recordRangeOfLogicalViewSchemaListNow();

    // [Step 4] Fetch 2. Some fetched measurements in [Step 3] are views. Process them.
    missedIndexAndPathString =
        schemaCache.computeSourceOfLogicalView(schemaComputationWithAutoCreation);
    indexOfMissingLogicalView = missedIndexAndPathString.left;
    missedPathStringOfLogicalView = missedIndexAndPathString.right;
    if (!missedPathStringOfLogicalView.isEmpty()) {
      ClusterSchemaTree viewSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaWithFullPaths(
              missedPathStringOfLogicalView, context);
      viewSchemaTree.computeSourceOfLogicalView(
          schemaComputationWithAutoCreation, indexOfMissingLogicalView);
    }

    // all schema has been taken and processed
    if (indexOfMissingMeasurements.isEmpty()) {
      return indexOfMissingMeasurements;
    }

    // [Step 5] Auto Create and process the missing schema
    if (config.isAutoCreateSchemaEnabled()) {
      // Check the isAligned value. If the input value is different from the actual value of the
      // existing device, throw exception.
      PartialPath devicePath = schemaComputationWithAutoCreation.getDevicePath();
      validateIsAlignedValueIfAutoCreate(
          schemaComputationWithAutoCreation.isAligned(), isAlignedPutIn, devicePath);
      ClusterSchemaTree schemaTree = new ClusterSchemaTree();
      autoCreateSchemaExecutor.autoCreateTimeSeries(
          schemaTree,
          devicePath,
          indexOfMissingMeasurements,
          schemaComputationWithAutoCreation.getMeasurements(),
          schemaComputationWithAutoCreation::getDataType,
          isAlignedPutIn,
          context);
      indexOfMissingMeasurements =
          schemaTree.compute(schemaComputationWithAutoCreation, indexOfMissingMeasurements);
    }

    return indexOfMissingMeasurements;
  }

  void processNormalTimeSeries(
      List<? extends ISchemaComputationWithAutoCreation> schemaComputationWithAutoCreationList,
      MPPQueryContext context) {
    // [Step 0] Record the input value.
    List<Boolean> isAlignedPutInList = null;
    if (config.isAutoCreateSchemaEnabled()) {
      isAlignedPutInList =
          schemaComputationWithAutoCreationList.stream()
              .map(ISchemaComputationWithAutoCreation::isAligned)
              .collect(Collectors.toList());
    }

    // [Step 1] Cache 1. compute measurements and record logical views.
    List<Integer> indexOfDevicesWithMissingMeasurements = new ArrayList<>();
    List<List<Integer>> indexOfMissingMeasurementsList =
        new ArrayList<>(schemaComputationWithAutoCreationList.size());
    List<Integer> indexOfMissingMeasurements;
    for (int i = 0, size = schemaComputationWithAutoCreationList.size(); i < size; i++) {
      indexOfMissingMeasurements =
          schemaCache.computeWithoutTemplate(schemaComputationWithAutoCreationList.get(i));
      if (!indexOfMissingMeasurements.isEmpty()) {
        indexOfDevicesWithMissingMeasurements.add(i);
        indexOfMissingMeasurementsList.add(indexOfMissingMeasurements);
      }
    }
    // [Step 2] Cache 2. process recorded logical views.
    boolean hasUnFetchedLogicalView = false;
    List<Pair<List<Integer>, List<String>>> missedIndexAndPathStringOfViewList =
        new ArrayList<>(schemaComputationWithAutoCreationList.size());
    for (ISchemaComputationWithAutoCreation iSchemaComputationWithAutoCreation :
        schemaComputationWithAutoCreationList) {
      Pair<List<Integer>, List<String>> missedIndexAndPathString =
          schemaCache.computeSourceOfLogicalView(iSchemaComputationWithAutoCreation);
      if (!missedIndexAndPathString.left.isEmpty()) {
        hasUnFetchedLogicalView = true;
      }
      missedIndexAndPathStringOfViewList.add(missedIndexAndPathString);
    }
    // all schema can be taken from cache
    if (indexOfDevicesWithMissingMeasurements.isEmpty() && (!hasUnFetchedLogicalView)) {
      return;
    }
    // [Step 3] Fetch 1.fetch schema from remote. Process logical view first; then process
    // measurements.
    // try fetch the missing schema from remote
    ISchemaComputationWithAutoCreation schemaComputationWithAutoCreation;
    ClusterSchemaTree remoteSchemaTree;
    if (!hasUnFetchedLogicalView) {
      remoteSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaOfMultiDevices(
              schemaComputationWithAutoCreationList.stream()
                  .map(ISchemaComputationWithAutoCreation::getDevicePath)
                  .collect(Collectors.toList()),
              schemaComputationWithAutoCreationList.stream()
                  .map(ISchemaComputationWithAutoCreation::getMeasurements)
                  .collect(Collectors.toList()),
              indexOfDevicesWithMissingMeasurements,
              indexOfMissingMeasurementsList,
              context);
    } else {
      PathPatternTree patternTree =
          computePatternTreeNeededReFetch(
              schemaComputationWithAutoCreationList.stream()
                  .map(ISchemaComputationWithAutoCreation::getDevicePath)
                  .collect(Collectors.toList()),
              schemaComputationWithAutoCreationList.stream()
                  .map(ISchemaComputationWithAutoCreation::getMeasurements)
                  .collect(Collectors.toList()),
              indexOfDevicesWithMissingMeasurements,
              indexOfMissingMeasurementsList);
      List<String> fullPathsNeedReFetch = new ArrayList<>();
      for (Pair<List<Integer>, List<String>> pair : missedIndexAndPathStringOfViewList) {
        fullPathsNeedReFetch.addAll(pair.right);
      }
      computePatternTreeNeededReFetch(patternTree, fullPathsNeedReFetch);
      remoteSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaWithPatternTreeAndCache(patternTree, context);
    }
    // make sure all missed views are computed.
    for (int i = 0; i < schemaComputationWithAutoCreationList.size(); i++) {
      schemaComputationWithAutoCreation = schemaComputationWithAutoCreationList.get(i);
      remoteSchemaTree.computeSourceOfLogicalView(
          schemaComputationWithAutoCreation, missedIndexAndPathStringOfViewList.get(i).left);
    }
    // check and compute the fetched schema
    List<Integer> indexOfDevicesNeedAutoCreateSchema = new ArrayList<>();
    List<List<Integer>> indexOfMeasurementsNeedAutoCreate = new ArrayList<>();
    for (int i = 0; i < indexOfDevicesWithMissingMeasurements.size(); i++) {
      schemaComputationWithAutoCreation =
          schemaComputationWithAutoCreationList.get(indexOfDevicesWithMissingMeasurements.get(i));
      indexOfMissingMeasurements =
          remoteSchemaTree.compute(
              schemaComputationWithAutoCreation, indexOfMissingMeasurementsList.get(i));
      schemaComputationWithAutoCreation.recordRangeOfLogicalViewSchemaListNow();
      if (!indexOfMissingMeasurements.isEmpty()) {
        indexOfDevicesNeedAutoCreateSchema.add(indexOfDevicesWithMissingMeasurements.get(i));
        indexOfMeasurementsNeedAutoCreate.add(indexOfMissingMeasurements);
      }
    }

    // [Step 4] Fetch 2. Some fetched measurements in [Step 3] are views. Process them.
    hasUnFetchedLogicalView = false;
    for (int i = 0, size = schemaComputationWithAutoCreationList.size(); i < size; i++) {
      Pair<List<Integer>, List<String>> missedIndexAndPathString =
          schemaCache.computeSourceOfLogicalView(schemaComputationWithAutoCreationList.get(i));
      if (!missedIndexAndPathString.left.isEmpty()) {
        hasUnFetchedLogicalView = true;
      }
      missedIndexAndPathStringOfViewList.get(i).left = missedIndexAndPathString.left;
      missedIndexAndPathStringOfViewList.get(i).right = missedIndexAndPathString.right;
    }
    if (hasUnFetchedLogicalView) {
      List<String> fullPathsNeedRefetch = new ArrayList<>();
      for (Pair<List<Integer>, List<String>> pair : missedIndexAndPathStringOfViewList) {
        fullPathsNeedRefetch.addAll(pair.right);
      }
      ClusterSchemaTree viewSchemaTree =
          clusterSchemaFetchExecutor.fetchSchemaWithFullPaths(fullPathsNeedRefetch, context);
      for (int i = 0, size = schemaComputationWithAutoCreationList.size(); i < size; i++) {
        schemaComputationWithAutoCreation = schemaComputationWithAutoCreationList.get(i);
        viewSchemaTree.computeSourceOfLogicalView(
            schemaComputationWithAutoCreation, missedIndexAndPathStringOfViewList.get(i).left);
      }
    }

    // all schema has been taken and processed
    if (indexOfDevicesNeedAutoCreateSchema.isEmpty()) {
      return;
    }

    // [Step 5] Auto Create and process the missing schema
    if (config.isAutoCreateSchemaEnabled()) {
      List<PartialPath> devicePathList =
          schemaComputationWithAutoCreationList.stream()
              .map(ISchemaComputationWithAutoCreation::getDevicePath)
              .collect(Collectors.toList());
      List<Boolean> isAlignedRealList =
          schemaComputationWithAutoCreationList.stream()
              .map(ISchemaComputationWithAutoCreation::isAligned)
              .collect(Collectors.toList());
      // Check the isAligned value. If the input value is different from the actual value of the
      // existing device, throw exception.
      validateIsAlignedValueIfAutoCreate(isAlignedRealList, isAlignedPutInList, devicePathList);

      ClusterSchemaTree schemaTree = new ClusterSchemaTree();
      autoCreateSchemaExecutor.autoCreateTimeSeries(
          schemaTree,
          devicePathList,
          indexOfDevicesNeedAutoCreateSchema,
          indexOfMeasurementsNeedAutoCreate,
          schemaComputationWithAutoCreationList.stream()
              .map(ISchemaComputationWithAutoCreation::getMeasurements)
              .collect(Collectors.toList()),
          schemaComputationWithAutoCreationList.stream()
              .map(
                  o -> {
                    TSDataType[] dataTypes = new TSDataType[o.getMeasurements().length];
                    for (int i = 0, length = dataTypes.length; i < length; i++) {
                      dataTypes[i] = o.getDataType(i);
                    }
                    return dataTypes;
                  })
              .collect(Collectors.toList()),
          isAlignedPutInList,
          context);
      indexOfDevicesWithMissingMeasurements = new ArrayList<>();
      indexOfMissingMeasurementsList = new ArrayList<>();
      for (int i = 0; i < indexOfDevicesNeedAutoCreateSchema.size(); i++) {
        schemaComputationWithAutoCreation =
            schemaComputationWithAutoCreationList.get(indexOfDevicesNeedAutoCreateSchema.get(i));
        indexOfMissingMeasurements =
            schemaTree.compute(
                schemaComputationWithAutoCreation, indexOfMeasurementsNeedAutoCreate.get(i));
        if (!indexOfMissingMeasurements.isEmpty()) {
          indexOfDevicesWithMissingMeasurements.add(indexOfDevicesNeedAutoCreateSchema.get(i));
          indexOfMissingMeasurementsList.add(indexOfMissingMeasurements);
        }
      }

      // all schema has been taken and processed
      if (indexOfDevicesWithMissingMeasurements.isEmpty()) {
        return;
      }
    } else {
      indexOfDevicesWithMissingMeasurements = indexOfDevicesNeedAutoCreateSchema;
      indexOfMissingMeasurementsList = indexOfMeasurementsNeedAutoCreate;
    }

    // offer null for the rest missing schema processing
    for (int i = 0; i < indexOfDevicesWithMissingMeasurements.size(); i++) {
      schemaComputationWithAutoCreation =
          schemaComputationWithAutoCreationList.get(indexOfDevicesWithMissingMeasurements.get(i));
      for (int index : indexOfMissingMeasurementsList.get(i)) {
        schemaComputationWithAutoCreation.computeMeasurement(index, null);
      }
    }
  }

  private void validateIsAlignedValueIfAutoCreate(
      List<Boolean> realValueList, List<Boolean> putInValueList, List<PartialPath> devicePathList) {
    int checkLen =
        Math.min(Math.min(realValueList.size(), putInValueList.size()), devicePathList.size());
    for (int i = 0; i < checkLen; i++) {
      validateIsAlignedValueIfAutoCreate(
          realValueList.get(i), putInValueList.get(i), devicePathList.get(i));
    }
  }

  private void validateIsAlignedValueIfAutoCreate(
      boolean realValue, boolean putInValue, PartialPath devicePath) {
    if (realValue != putInValue) {
      String msg;
      if (realValue) {
        msg =
            "Timeseries under this device is aligned, please use createTimeseries or change device.";
      } else {
        msg =
            "Timeseries under this device is not aligned, please use createTimeseries or change device.";
      }
      throw new SemanticException(new AlignedTimeseriesException(msg, devicePath.getFullPath()));
    }
  }
}
