// 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.impala.service;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import org.apache.impala.analysis.BoolLiteral;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.TableName;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TCacheJarParams;
import org.apache.impala.thrift.TCacheJarResult;
import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TCatalogServiceRequestHeader;
import org.apache.impala.thrift.TColumnValue;
import org.apache.impala.thrift.TErrorCode;
import org.apache.impala.thrift.TExprBatch;
import org.apache.impala.thrift.TGetPartitionStatsRequest;
import org.apache.impala.thrift.TGetPartitionStatsResponse;
import org.apache.impala.thrift.TParseDateStringResult;
import org.apache.impala.thrift.TPrioritizeLoadRequest;
import org.apache.impala.thrift.TPrioritizeLoadResponse;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TResultRow;
import org.apache.impala.thrift.TSymbolLookupParams;
import org.apache.impala.thrift.TSymbolLookupResult;
import org.apache.impala.thrift.TTable;
import org.apache.impala.util.NativeLibUtil;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;

/**
 * This class provides the Impala executor functionality to the FE.
 * fe-support.cc implements all the native calls.
 * If the planner is executed inside Impalad, Impalad would have registered all the JNI
 * native functions already. There's no need to load the shared library.
 * For unit test (mvn test), load the shared library because the native function has not
 * been loaded yet.
 */
public class FeSupport {
  private final static Logger LOG = LoggerFactory.getLogger(FeSupport.class);
  private static boolean loaded_ = false;

  // Only called if this library is explicitly loaded. This only happens
  // when running FE tests.
  public native static void NativeFeTestInit();

  // Returns a serialized TResultRow
  public native static byte[] NativeEvalExprsWithoutRow(
      byte[] thriftExprBatch, byte[] thriftQueryGlobals, long maxResultSize);

  // Returns a serialized TSymbolLookupResult
  public native static byte[] NativeLookupSymbol(byte[] thriftSymbolLookup);

  // Returns a serialized TCacheJarResult
  public native static byte[] NativeCacheJar(byte[] thriftCacheJar);

  // Adds a topic item to the backend's pending metadata-topic update.
  // 'serializationBuffer' is a serialized TCatalogObject.
  // The return value is true if the operation succeeds and false otherwise.
  public native static boolean NativeAddPendingTopicItem(long nativeCatalogServerPtr,
      String key, long version, byte[] serializationBuffer, boolean deleted);

  // Get a catalog object update from the backend. A pair of isDeletion flag and
  // serialized TCatalogObject is returned.
  public native static Pair<Boolean, ByteBuffer> NativeGetNextCatalogObjectUpdate(
      long nativeIteratorPtr);

  // The return value is true if the operation succeeds and false otherwise.
  public native static boolean NativeLibCacheSetNeedsRefresh(String hdfsLocation);
  public native static boolean NativeLibCacheRemoveEntry(String hdfsLibFile);

  // The two methods below make RPCs to the Catalog Server. To keep our kerberos
  // configuration consolidated, we make make all RPCs in the BE layer instead of calling
  // the Catalog Server using Java Thrift bindings.

  // Does an RPC to the Catalog Server to prioritize the metadata loading of a
  // one or more catalog objects.
  public native static byte[] NativePrioritizeLoad(byte[] thriftReq);

  public native static byte[] NativeGetPartialCatalogObject(byte[] thriftReq)
      throws InternalException;

  // Does an RPC to the Catalog Server to fetch specified table partition statistics.
  public native static byte[] NativeGetPartitionStats(byte[] thriftReq);

  public native static byte[] NativeUpdateTableUsage(byte[] thriftReq);

  // Does an RPC to the Catalog Server to check if the given user is a Sentry admin.
  public native static byte[] NativeSentryAdminCheck(byte[] thriftReq);

  // Parses a string of comma-separated key=value query options ('csvQueryOptions'),
  // updates the existing query options ('queryOptions') with them and returns the
  // resulting serialized TQueryOptions object.
  // A note about the function's interface: ideally we wouldn't have to pass in the
  // existing query options. We could just return the newly set query options to the
  // caller and let the caller update the existing query options with the new ones.
  // Unfortunately due to a bug in the thrift-generated TQueryOptions class, in some cases
  // it is impossible to figure out whether a query option has been set explicitly or left
  // at its default setting, therefore this approach would not work.
  public native static byte[] NativeParseQueryOptions(String csvQueryOptions,
      byte[] queryOptions);

  public native static int MinLogSpaceForBloomFilter(long ndv, double fpp);

  // Parses date string, verifies if it is valid and returns the resulting
  // TParseDateStringResult object. Different date string variations are accepted.
  // E.g.: '2011-01-01', '2011-01-1', '2011-1-01', '2011-01-01'.
  public native static byte[] nativeParseDateString(String date);

  /**
   * Locally caches the jar at the specified HDFS location.
   *
   * @param hdfsLocation The path to the jar in HDFS
   * @return The result of the call to cache the jar, includes a status and the local
   *         path of the cached jar if the operation was successful.
   */
  public static TCacheJarResult CacheJar(String hdfsLocation) throws InternalException {
    Preconditions.checkNotNull(hdfsLocation);
    TCacheJarParams params = new TCacheJarParams(hdfsLocation);
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    byte[] result;
    try {
      result = CacheJar(serializer.serialize(params));
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TCacheJarResult thriftResult = new TCacheJarResult();
      deserializer.deserialize(thriftResult, result);
      return thriftResult;
    } catch (TException e) {
      // this should never happen
      throw new InternalException(
          "Couldn't cache jar at HDFS location " + hdfsLocation, e);
    }
  }

  private static byte[] CacheJar(byte[] thriftParams) {
    try {
      return NativeCacheJar(thriftParams);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeCacheJar(thriftParams);
  }

  /**
   * If it is known that the size of the evaluated expression is fixed, e.g.,
   * the size of an integer, then this method will be called to perform the evaluation.
   * Otherwise, the method EvalExprWithoutRowBounded that takes an additional argument
   * specifying the upper bound on the size of evaluated expression should be invoked.
   */
  public static TColumnValue EvalExprWithoutRow(Expr expr, TQueryCtx queryCtx)
    throws InternalException {
    return EvalExprWithoutRowBounded(expr, queryCtx, 0);
  }

  public static TColumnValue EvalExprWithoutRowBounded(Expr expr, TQueryCtx queryCtx,
    int maxResultSize) throws InternalException {
    Preconditions.checkState(!expr.contains(SlotRef.class));
    TExprBatch exprBatch = new TExprBatch();
    exprBatch.addToExprs(expr.treeToThrift());
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    byte[] result;
    try {
      result = EvalExprsWithoutRowBounded(
          serializer.serialize(exprBatch), serializer.serialize(queryCtx), maxResultSize);
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TResultRow val = new TResultRow();
      deserializer.deserialize(val, result);
      if (val.getColValsSize() != 1) {
        throw new IllegalStateException(String.format("Illegal expr eval result. " +
            "Expr=%s\nTExpBatch=%s\nResult=%s", expr.toSql(), exprBatch, val));
      }
      return val.getColVals().get(0);
    } catch (TException e) {
      // this should never happen
      throw new InternalException("couldn't execute expr " + expr.toSql(), e);
    }
  }

  private static byte[] LookupSymbol(byte[] thriftParams) {
    try {
      return NativeLookupSymbol(thriftParams);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeLookupSymbol(thriftParams);
  }

  public static TSymbolLookupResult LookupSymbol(TSymbolLookupParams params)
      throws InternalException {
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    try {
      byte[] resultBytes = LookupSymbol(serializer.serialize(params));
      Preconditions.checkNotNull(resultBytes);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TSymbolLookupResult result = new TSymbolLookupResult();
      deserializer.deserialize(result, resultBytes);
      return result;
    } catch (TException e) {
      // this should never happen
      throw new InternalException("couldn't perform symbol lookup.", e);
    }
  }

  /**
   * If it is known that the size of the evaluated expression is fixed, e.g.,
   * the size of an integer, then this method will be called to perform the evaluation.
   * Otherwise, the method EvalExprsWithoutRowBounded that takes an additional argument
   * specifying the upper bound on the size of rewritten expression should be invoked.
   */
  private static byte[] EvalExprsWithoutRow(
      byte[] thriftExprBatch, byte[] thriftQueryContext) {
    return EvalExprsWithoutRowBounded(thriftExprBatch, thriftQueryContext, 0);
  }

  private static byte[] EvalExprsWithoutRowBounded(
      byte[] thriftExprBatch, byte[] thriftQueryContext, int maxResultSize) {
    try {
      return NativeEvalExprsWithoutRow(thriftExprBatch, thriftQueryContext,
        maxResultSize);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeEvalExprsWithoutRow(thriftExprBatch, thriftQueryContext, maxResultSize);
  }

  public static boolean EvalPredicate(Expr pred, TQueryCtx queryCtx)
      throws InternalException {
    // Shortcuts to avoid expensive BE evaluation.
    if (pred instanceof BoolLiteral) return ((BoolLiteral) pred).getValue();
    if (Expr.IS_NULL_LITERAL.apply(pred)) return false;
    Preconditions.checkState(pred.getType().isBoolean());
    TColumnValue val = EvalExprWithoutRow(pred, queryCtx);
    // Return false if pred evaluated to false or NULL. True otherwise.
    return val.isBool_val() && val.bool_val;
  }

  /**
   * Evaluate a batch of predicates in the BE. The results are stored in a
   * TResultRow object, where each TColumnValue in it stores the result of
   * a predicate evaluation.
   *
   * TODO: This function is currently used for improving the performance of
   * partition pruning (see IMPALA-887), hence it only supports boolean
   * exprs. In the future, we can extend it to support arbitrary exprs without
   * SlotRefs.
   */
  public static TResultRow EvalPredicateBatch(List<Expr> exprs,
      TQueryCtx queryCtx) throws InternalException {
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    TExprBatch exprBatch = new TExprBatch();
    for (Expr expr: exprs) {
      // Make sure we only process boolean exprs.
      Preconditions.checkState(expr.getType().isBoolean());
      Preconditions.checkState(!expr.contains(SlotRef.class));
      exprBatch.addToExprs(expr.treeToThrift());
    }
    byte[] result;
    try {
      result = EvalExprsWithoutRow(
          serializer.serialize(exprBatch), serializer.serialize(queryCtx));
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TResultRow val = new TResultRow();
      deserializer.deserialize(val, result);
      return val;
    } catch (TException e) {
      // this should never happen
      throw new InternalException("couldn't execute a batch of exprs.", e);
    }
  }

  private static byte[] PrioritizeLoad(byte[] thriftReq) {
    try {
      return NativePrioritizeLoad(thriftReq);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativePrioritizeLoad(thriftReq);
  }

  public static void PrioritizeLoad(Set<TableName> tableNames)
      throws InternalException {
    Preconditions.checkNotNull(tableNames);

    LOG.info(String.format("Requesting prioritized load of table(s): %s",
        Joiner.on(", ").join(tableNames)));

    List<TCatalogObject> objectDescs = new ArrayList<TCatalogObject>(tableNames.size());
    for (TableName tableName: tableNames) {
      TCatalogObject catalogObject = new TCatalogObject();
      catalogObject.setType(TCatalogObjectType.TABLE);
      catalogObject.setTable(new TTable(tableName.getDb(), tableName.getTbl()));
      objectDescs.add(catalogObject);
    }

    TPrioritizeLoadRequest request = new TPrioritizeLoadRequest ();
    request.setHeader(new TCatalogServiceRequestHeader());
    request.setObject_descs(objectDescs);

    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    try {
      byte[] result = PrioritizeLoad(serializer.serialize(request));
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TPrioritizeLoadResponse response = new TPrioritizeLoadResponse();
      deserializer.deserialize(response, result);
      if (response.getStatus().getStatus_code() != TErrorCode.OK) {
        throw new InternalException("Error requesting prioritized load: " +
            Joiner.on("\n").join(response.getStatus().getError_msgs()));
      }
    } catch (TException e) {
      // this should never happen
      throw new InternalException("Error processing request: " + e.getMessage(), e);
    }
  }

  private static byte[] GetPartitionStats(byte[] thriftReq) {
    try {
      return NativeGetPartitionStats(thriftReq);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeGetPartitionStats(thriftReq);
  }

  public static TGetPartitionStatsResponse GetPartitionStats(
      TableName table) throws InternalException {
    Preconditions.checkNotNull(table);

    LOG.info("Fetching partition statistics for table {} from catalog.", table);

    TGetPartitionStatsRequest request = new TGetPartitionStatsRequest();
    request.setTable_name(table.toThrift());
    TGetPartitionStatsResponse response = new TGetPartitionStatsResponse();
    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    try {
      byte[] result = GetPartitionStats(serializer.serialize(request));
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());

      deserializer.deserialize(response, result);
      if (response.getStatus().getStatus_code() != TErrorCode.OK) {
        throw new InternalException("Error requesting GetPartitionStats: "
            + Joiner.on("\n").join(response.getStatus().getError_msgs()));
      }
    } catch (TException e) {
      // this should never happen
      throw new InternalException("Error processing request: " + e.getMessage(), e);
    }
    int numPartitionsFetched = 0;
    if (response.isSetPartition_stats()) {
      numPartitionsFetched = response.partition_stats.size();
    }
    LOG.info("Fetched statistics for {} partitions of table {}.", numPartitionsFetched,
        table);
    return response;
  }

  private static byte[] ParseQueryOptions(String csvQueryOptions, byte[] queryOptions) {
    try {
      return NativeParseQueryOptions(csvQueryOptions, queryOptions);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeParseQueryOptions(csvQueryOptions, queryOptions);
  }

  /**
   * Parses a string of comma-separated key=value query options. Returns a TQueryOptions
   * object that contains the updated query options.
   */
  public static TQueryOptions ParseQueryOptions(String csvQueryOptions,
      TQueryOptions queryOptions) throws InternalException {
    Preconditions.checkNotNull(csvQueryOptions);
    Preconditions.checkNotNull(queryOptions);

    TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
    try {
      byte[] result = ParseQueryOptions(csvQueryOptions,
          serializer.serialize(queryOptions));
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TQueryOptions updatedQueryOptions = new TQueryOptions();
      deserializer.deserialize(updatedQueryOptions, result);
      return updatedQueryOptions;
    } catch (TException e) {
      throw new InternalException("Could not parse query options: " + e.getMessage(), e);
    }
  }

  /**
   * Returns the log (base 2) of the minimum number of bytes we need for a Bloom
   * filter with 'ndv' unique elements and a false positive probability of less
   * than 'fpp'.
   */
  public static int GetMinLogSpaceForBloomFilter(long ndv, double fpp) {
    try {
      return MinLogSpaceForBloomFilter(ndv, fpp);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return MinLogSpaceForBloomFilter(ndv, fpp);
  }

  public static byte[] GetPartialCatalogObject(byte[] thriftReq)
      throws InternalException {
    try {
      return NativeGetPartialCatalogObject(thriftReq);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeGetPartialCatalogObject(thriftReq);
  }

  public static byte[] CheckSentryAdmin(byte[] thriftReq) {
    try {
      return NativeSentryAdminCheck(thriftReq);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return NativeSentryAdminCheck(thriftReq);
  }

  private static byte[] parseDateStringUtil(String date) {
    try {
      return nativeParseDateString(date);
    } catch (UnsatisfiedLinkError e) {
      loadLibrary();
    }
    return nativeParseDateString(date);
  }

  /**
   * Parses date string, verifies if it is valid and returns the resulting
   * TParseDateStringResult object. Different date string variations are accepted.
   * E.g.: '2011-01-01', '2011-01-1', '2011-1-01', '2011-01-01'.
   */
  public static TParseDateStringResult parseDateString(String date)
      throws InternalException {
    Preconditions.checkNotNull(date);
    try {
      byte[] result = parseDateStringUtil(date);
      Preconditions.checkNotNull(result);
      TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
      TParseDateStringResult res = new TParseDateStringResult();
      deserializer.deserialize(res, result);
      return res;
    } catch (TException e) {
      throw new InternalException("Could not parse date string: " + e.getMessage(), e);
    }
  }

  /**
   * This function should be called explicitly by the FeSupport to ensure that
   * native functions are loaded. Tests that depend on JniCatalog or JniFrontend
   * being instantiated should also call this function.
   */
  public static synchronized void loadLibrary() {
    if (loaded_) return;
    LOG.info("Loading libfesupport.so");
    NativeLibUtil.loadLibrary("libfesupport.so");
    LOG.info("Loaded libfesupport.so");
    loaded_ = true;
    NativeFeTestInit();
  }
}