// 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.

#include "service/query-options.h"

#include "runtime/runtime-filter.h"
#include "util/debug-util.h"
#include "util/mem-info.h"
#include "util/parse-util.h"
#include "util/string-parser.h"
#include "exprs/timezone_db.h"
#include "gen-cpp/ImpalaInternalService_types.h"

#include <zstd.h>
#include <sstream>
#include <boost/algorithm/string.hpp>
#include <gutil/strings/strip.h>
#include <gutil/strings/substitute.h>

#include "common/names.h"

DECLARE_int64(min_buffer_size);

using boost::algorithm::iequals;
using boost::algorithm::is_any_of;
using boost::algorithm::token_compress_on;
using boost::algorithm::split;
using boost::algorithm::trim;
using std::to_string;
using beeswax::TQueryOptionLevel;
using namespace impala;
using namespace strings;

DECLARE_int32(idle_session_timeout);

// Utility method to wrap ParseUtil::ParseMemSpec() by returning a Status instead of an
// int.
static Status ParseMemValue(const string& value, const string& key, int64_t* result) {
  bool is_percent;
  *result = ParseUtil::ParseMemSpec(value, &is_percent, MemInfo::physical_mem());
  if (*result < 0) {
    return Status("Failed to parse " + key + " from '" + value + "'.");
  }
  if (is_percent) {
    return Status("Invalid " + key + " with percent '" + value + "'.");
  }
  return Status::OK();
}

void impala::OverlayQueryOptions(const TQueryOptions& src, const QueryOptionsMask& mask,
    TQueryOptions* dst) {
  DCHECK_GT(mask.size(), _TImpalaQueryOptions_VALUES_TO_NAMES.size()) <<
      "Size of QueryOptionsMask must be increased.";
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
  if (src.__isset.NAME && mask[TImpalaQueryOptions::ENUM]) dst->__set_##NAME(src.NAME);
#define REMOVED_QUERY_OPT_FN(NAME, ENUM)
  QUERY_OPTS_TABLE
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
}

// Choose different print function based on the type.
// TODO: In thrift 0.11.0 operator << is implemented for enums and this indirection can be
// removed.
template <typename T, typename std::enable_if_t<std::is_enum<T>::value>* = nullptr>
string PrintQueryOptionValue(const T& option) {
  return PrintThriftEnum(option);
}

template <typename T, typename std::enable_if_t<std::is_arithmetic<T>::value>* = nullptr>
string PrintQueryOptionValue(const T& option) {
  return std::to_string(option);
}

const string& PrintQueryOptionValue(const std::string& option)  {
  return option;
}

const string PrintQueryOptionValue(const impala::TCompressionCodec& compression_codec) {
  if (compression_codec.codec != THdfsCompression::ZSTD) {
    return Substitute("$0", PrintThriftEnum(compression_codec.codec));
  } else {
    return Substitute("$0:$1", PrintThriftEnum(compression_codec.codec),
        compression_codec.compression_level);
  }
}

void impala::TQueryOptionsToMap(const TQueryOptions& query_options,
    map<string, string>* configuration) {
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
  {\
    if (query_options.__isset.NAME) { \
      (*configuration)[#ENUM] = PrintQueryOptionValue(query_options.NAME); \
    } else { \
      (*configuration)[#ENUM] = ""; \
    }\
  }
#define REMOVED_QUERY_OPT_FN(NAME, ENUM) (*configuration)[#ENUM] = "";
  QUERY_OPTS_TABLE
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
}

// Resets query_options->option to its default value.
static void ResetQueryOption(const int option, TQueryOptions* query_options) {
  const static TQueryOptions defaults;
  switch (option) {
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
    case TImpalaQueryOptions::ENUM:\
      query_options->__isset.NAME = defaults.__isset.NAME;\
      query_options->NAME = defaults.NAME;\
      break;
#define REMOVED_QUERY_OPT_FN(NAME, ENUM)
  QUERY_OPTS_TABLE
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
  }
}

static TQueryOptions DefaultQueryOptions() {
  TQueryOptions defaults;
  // default value of idle_session_timeout is set by a command line flag.
  defaults.__set_idle_session_timeout(FLAGS_idle_session_timeout);
  return defaults;
}

inline bool operator!=(const TCompressionCodec& a,
  const TCompressionCodec& b) {
  return (a.codec != b.codec || a.compression_level != b.compression_level);
}

string impala::DebugQueryOptions(const TQueryOptions& query_options) {
  const static TQueryOptions defaults = DefaultQueryOptions();
  int i = 0;
  stringstream ss;
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
  if (query_options.__isset.NAME &&\
      (!defaults.__isset.NAME || query_options.NAME != defaults.NAME)) {\
    if (i++ > 0) ss << ",";\
    ss << #ENUM << "=" << query_options.NAME;\
  }
#define REMOVED_QUERY_OPT_FN(NAME, ENUM)
  QUERY_OPTS_TABLE
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
  return ss.str();
}

// Returns the TImpalaQueryOptions enum for the given "key". Input is case insensitive.
// Return -1 if the input is an invalid option.
static int GetQueryOptionForKey(const string& key) {
  map<int, const char*>::const_iterator itr =
      _TImpalaQueryOptions_VALUES_TO_NAMES.begin();
  for (; itr != _TImpalaQueryOptions_VALUES_TO_NAMES.end(); ++itr) {
    if (iequals(key, (*itr).second)) {
      return itr->first;
    }
  }
  return -1;
}

// Return true if we can ignore a reference to this removed query option.
static bool IsRemovedQueryOption(const string& key) {
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)
#define REMOVED_QUERY_OPT_FN(NAME, ENUM) \
  if (iequals(key, #NAME)) { \
    return true; \
  }
  QUERY_OPTS_TABLE
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
  return false;
}

// Return all enum values in a string format, e.g. FOO(1), BAR(2), BAZ(3).
static string GetThriftEnumValues(const map<int, const char*>& enum_values_to_names) {
  bool first = true;
  stringstream ss;
  for (const auto& e : enum_values_to_names) {
    if (!first) {
      ss << ", ";
    } else {
      first = false;
    }
    ss << e.second << "(" << e.first << ")";
  }
  return ss.str();
}

// Return false for an invalid Thrift enum value.
template<typename ENUM_TYPE>
static Status GetThriftEnum(const string& value, const string& key,
    const map<int, const char*>& enum_values_to_names, ENUM_TYPE* enum_value) {
  for (const auto& e : enum_values_to_names) {
    if (iequals(value, to_string(e.first)) || iequals(value, e.second)) {
      *enum_value = static_cast<ENUM_TYPE>(e.first);
      return Status::OK();
    }
  }
  return Status(Substitute("Invalid $0: '$1'. Valid values are $2.", key, value,
      GetThriftEnumValues(enum_values_to_names)));
}

// Return true if the given value is true (case-insensitive) or 1.
static bool IsTrue(const string& value) {
  return iequals(value, "true") || iequals(value, "1");
}

// Note that we allow numerical values for boolean and enum options. This is because
// TQueryOptionsToMap() will output the numerical values, and we need to parse its output
// configuration.
Status impala::SetQueryOption(const string& key, const string& value,
    TQueryOptions* query_options, QueryOptionsMask* set_query_options_mask) {
  int option = GetQueryOptionForKey(key);
  if (option < 0) {
    return Status(Substitute("Invalid query option: $0", key));
  } else if (value == "") {
    ResetQueryOption(option, query_options);
    if (set_query_options_mask != nullptr) {
      DCHECK_LT(option, set_query_options_mask->size());
      set_query_options_mask->reset(option);
    }
  } else {
    switch (option) {
      case TImpalaQueryOptions::ABORT_ON_ERROR:
        query_options->__set_abort_on_error(IsTrue(value));
        break;
      case TImpalaQueryOptions::MAX_ERRORS:
        query_options->__set_max_errors(atoi(value.c_str()));
        break;
      case TImpalaQueryOptions::DISABLE_CODEGEN:
        query_options->__set_disable_codegen(IsTrue(value));
        break;
      case TImpalaQueryOptions::BATCH_SIZE: {
        StringParser::ParseResult status;
        int val = StringParser::StringToInt<int>(value.c_str(),
            static_cast<int>(value.size()), &status);
        if (status != StringParser::PARSE_SUCCESS || val < 0 || val > 65536) {
          return Status(Substitute("Invalid batch size '$0'. Valid sizes are in"
              "[0, 65536]", value));
        }
        query_options->__set_batch_size(val);
        break;
      }
      case TImpalaQueryOptions::MEM_LIMIT: {
        // Parse the mem limit spec and validate it.
        int64_t bytes_limit;
        RETURN_IF_ERROR(ParseMemValue(value, "query memory limit", &bytes_limit));
        query_options->__set_mem_limit(bytes_limit);
        break;
      }
      case TImpalaQueryOptions::NUM_NODES:
        query_options->__set_num_nodes(atoi(value.c_str()));
        break;
      case TImpalaQueryOptions::MAX_SCAN_RANGE_LENGTH: {
        int64_t scan_length = 0;
        RETURN_IF_ERROR(ParseMemValue(value, "scan range length", &scan_length));
        query_options->__set_max_scan_range_length(scan_length);
        break;
      }
      case TImpalaQueryOptions::NUM_SCANNER_THREADS:
        query_options->__set_num_scanner_threads(atoi(value.c_str()));
        break;
      case TImpalaQueryOptions::DEBUG_ACTION:
        query_options->__set_debug_action(value.c_str());
        break;
      case TImpalaQueryOptions::COMPRESSION_CODEC: {
        // Acceptable values are:
        // - zstd:compression_level
        // - codec
        vector<string> tokens;
        split(tokens, value, is_any_of(":"), token_compress_on);
        if (tokens.size() > 2) return Status("Invalid compression codec value");

        string& codec_name = tokens[0];
        trim(codec_name);
        int compression_level = ZSTD_CLEVEL_DEFAULT;
        THdfsCompression::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(codec_name, "compression codec",
            _THdfsCompression_VALUES_TO_NAMES, &enum_type));

        if (tokens.size() == 2) {
          if (enum_type != THdfsCompression::ZSTD) {
            return Status("Compression level only supported for ZSTD");
          }
          StringParser::ParseResult status;
          string& clevel = tokens[1];
          trim(clevel);
          compression_level = StringParser::StringToInt<int>(
            clevel.c_str(), static_cast<int>(clevel.size()), &status);
          if (status != StringParser::PARSE_SUCCESS || compression_level < 1
              || compression_level > ZSTD_maxCLevel()) {
            return Status(Substitute("Invalid ZSTD compression level '$0'."
                " Valid values are in [1,$1]", clevel, ZSTD_maxCLevel()));
          }
        }

        TCompressionCodec compression_codec;
        compression_codec.__set_codec(enum_type);
        if (enum_type == THdfsCompression::ZSTD) {
          compression_codec.__set_compression_level(compression_level);
        }
        query_options->__set_compression_codec(compression_codec);
        break;
      }
      case TImpalaQueryOptions::HBASE_CACHING:
        query_options->__set_hbase_caching(atoi(value.c_str()));
        break;
      case TImpalaQueryOptions::HBASE_CACHE_BLOCKS:
        query_options->__set_hbase_cache_blocks(IsTrue(value));
        break;
      case TImpalaQueryOptions::PARQUET_FILE_SIZE: {
        int64_t file_size;
        RETURN_IF_ERROR(ParseMemValue(value, "parquet file size", &file_size));
        if (file_size > numeric_limits<int32_t>::max()) {
          // Do not allow values greater than or equal to 2GB since hdfsOpenFile() from
          // the HDFS API gets an int32 blocksize parameter (see HDFS-8949).
          stringstream ss;
          ss << "The PARQUET_FILE_SIZE query option must be less than 2GB.";
          return Status(ss.str());
        } else {
          query_options->__set_parquet_file_size(file_size);
        }
        break;
      }
      case TImpalaQueryOptions::EXPLAIN_LEVEL: {
        TExplainLevel::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "explain level",
            _TExplainLevel_VALUES_TO_NAMES, &enum_type));
        query_options->__set_explain_level(enum_type);
        break;
      }
      case TImpalaQueryOptions::SYNC_DDL:
        query_options->__set_sync_ddl(IsTrue(value));
        break;
      case TImpalaQueryOptions::REQUEST_POOL:
        query_options->__set_request_pool(value);
        break;
      case TImpalaQueryOptions::DISABLE_OUTERMOST_TOPN:
        query_options->__set_disable_outermost_topn(IsTrue(value));
        break;
      case TImpalaQueryOptions::QUERY_TIMEOUT_S: {
        StringParser::ParseResult result;
        const int32_t timeout_s =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || timeout_s < 0) {
          return Status(
              Substitute("Invalid query timeout: '$0'. "
                         "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_query_timeout_s(timeout_s);
        break;
      }
      case TImpalaQueryOptions::BUFFER_POOL_LIMIT: {
        int64_t mem;
        RETURN_IF_ERROR(ParseMemValue(value, "buffer pool limit", &mem));
        query_options->__set_buffer_pool_limit(mem);
        break;
      }
      case TImpalaQueryOptions::APPX_COUNT_DISTINCT: {
        query_options->__set_appx_count_distinct(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::DISABLE_UNSAFE_SPILLS: {
        query_options->__set_disable_unsafe_spills(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::EXEC_SINGLE_NODE_ROWS_THRESHOLD:
        query_options->__set_exec_single_node_rows_threshold(atoi(value.c_str()));
        break;
      case TImpalaQueryOptions::OPTIMIZE_PARTITION_KEY_SCANS:
        query_options->__set_optimize_partition_key_scans(IsTrue(value));
        break;
      case TImpalaQueryOptions::REPLICA_PREFERENCE: {
        map<int, const char *> valid_enums_values = {
            {0, "CACHE_LOCAL"},
            {2, "DISK_LOCAL"},
            {4, "REMOTE"}
        };
        TReplicaPreference::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "replica memory distance preference",
            valid_enums_values, &enum_type));
        query_options->__set_replica_preference(enum_type);
        break;
      }
      case TImpalaQueryOptions::SCHEDULE_RANDOM_REPLICA:
        query_options->__set_schedule_random_replica(IsTrue(value));
        break;
      case TImpalaQueryOptions::DISABLE_STREAMING_PREAGGREGATIONS:
        query_options->__set_disable_streaming_preaggregations(IsTrue(value));
        break;
      case TImpalaQueryOptions::RUNTIME_FILTER_MODE: {
        TRuntimeFilterMode::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "runtime filter mode",
            _TRuntimeFilterMode_VALUES_TO_NAMES, &enum_type));
        query_options->__set_runtime_filter_mode(enum_type);
        break;
      }
      case TImpalaQueryOptions::RUNTIME_FILTER_MAX_SIZE:
      case TImpalaQueryOptions::RUNTIME_FILTER_MIN_SIZE:
      case TImpalaQueryOptions::RUNTIME_BLOOM_FILTER_SIZE: {
        int64_t size;
        RETURN_IF_ERROR(ParseMemValue(value, "Bloom filter size", &size));
        if (size < RuntimeFilterBank::MIN_BLOOM_FILTER_SIZE ||
            size > RuntimeFilterBank::MAX_BLOOM_FILTER_SIZE) {
          return Status(Substitute("$0 is not a valid Bloom filter size for $1. "
                  "Valid sizes are in [$2, $3].", value, PrintThriftEnum(
                      static_cast<TImpalaQueryOptions::type>(option)),
                  RuntimeFilterBank::MIN_BLOOM_FILTER_SIZE,
                  RuntimeFilterBank::MAX_BLOOM_FILTER_SIZE));
        }
        if (option == TImpalaQueryOptions::RUNTIME_FILTER_MAX_SIZE
            && size < FLAGS_min_buffer_size
            // last condition is to unblock the highly improbable case where the
            // min_buffer_size is greater than RuntimeFilterBank::MAX_BLOOM_FILTER_SIZE.
            && FLAGS_min_buffer_size <= RuntimeFilterBank::MAX_BLOOM_FILTER_SIZE) {
          return Status(Substitute("$0 should not be less than $1 which is the minimum "
              "buffer size that can be allocated by the buffer pool",
              PrintThriftEnum(static_cast<TImpalaQueryOptions::type>(option)),
              FLAGS_min_buffer_size));
        }
        if (option == TImpalaQueryOptions::RUNTIME_BLOOM_FILTER_SIZE) {
          query_options->__set_runtime_bloom_filter_size(size);
        } else if (option == TImpalaQueryOptions::RUNTIME_FILTER_MIN_SIZE) {
          query_options->__set_runtime_filter_min_size(size);
        } else if (option == TImpalaQueryOptions::RUNTIME_FILTER_MAX_SIZE) {
          query_options->__set_runtime_filter_max_size(size);
        }
        break;
      }
      case TImpalaQueryOptions::RUNTIME_FILTER_WAIT_TIME_MS: {
        StringParser::ParseResult result;
        const int32_t time_ms =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || time_ms < 0) {
          return Status(
              Substitute("$0 is not a valid wait time. Valid sizes are in [0, $1].",
                  value, numeric_limits<int32_t>::max()));
        }
        query_options->__set_runtime_filter_wait_time_ms(time_ms);
        break;
      }
      case TImpalaQueryOptions::DISABLE_ROW_RUNTIME_FILTERING:
        query_options->__set_disable_row_runtime_filtering(IsTrue(value));
        break;
      case TImpalaQueryOptions::MAX_NUM_RUNTIME_FILTERS: {
        StringParser::ParseResult status;
        int val = StringParser::StringToInt<int>(value.c_str(), value.size(), &status);
        if (status != StringParser::PARSE_SUCCESS) {
          return Status(Substitute("Invalid number of runtime filters: '$0'.", value));
        }
        if (val < 0) {
          return Status(Substitute("Invalid number of runtime filters: '$0'. "
              "Only positive values are allowed.", val));
        }
        query_options->__set_max_num_runtime_filters(val);
        break;
      }
      case TImpalaQueryOptions::PARQUET_ANNOTATE_STRINGS_UTF8: {
        query_options->__set_parquet_annotate_strings_utf8(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PARQUET_FALLBACK_SCHEMA_RESOLUTION: {
        TParquetFallbackSchemaResolution::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "parquet fallback schema resolution",
            _TParquetFallbackSchemaResolution_VALUES_TO_NAMES, &enum_type));
        query_options->__set_parquet_fallback_schema_resolution(enum_type);
        break;
      }
      case TImpalaQueryOptions::PARQUET_ARRAY_RESOLUTION: {
        TParquetArrayResolution::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "parquet array resolution",
            _TParquetArrayResolution_VALUES_TO_NAMES, &enum_type));
        query_options->__set_parquet_array_resolution(enum_type);
        break;
      }
      case TImpalaQueryOptions::MT_DOP: {
        StringParser::ParseResult result;
        const int32_t dop =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || dop < 0 || dop > 64) {
          return Status(
              Substitute("$0 is not valid for mt_dop. Valid values are in "
                "[0, 64].", value));
        }
        query_options->__set_mt_dop(dop);
        break;
      }
      case TImpalaQueryOptions::S3_SKIP_INSERT_STAGING: {
        query_options->__set_s3_skip_insert_staging(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PREFETCH_MODE: {
        TPrefetchMode::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "prefetch mode",
            _TPrefetchMode_VALUES_TO_NAMES, &enum_type));
        query_options->__set_prefetch_mode(enum_type);
        break;
      }
      case TImpalaQueryOptions::STRICT_MODE: {
        query_options->__set_strict_mode(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::SCRATCH_LIMIT: {
        // Parse the scratch limit spec and validate it.
        if (iequals(value, "-1")) {
          query_options->__set_scratch_limit(-1);
        } else {
          int64_t bytes_limit;
          RETURN_IF_ERROR(ParseMemValue(value, "Scratch space memory limit",
              &bytes_limit));
          query_options->__set_scratch_limit(bytes_limit);
        }
        break;
      }
      case TImpalaQueryOptions::ENABLE_EXPR_REWRITES: {
        query_options->__set_enable_expr_rewrites(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::DECIMAL_V2: {
        query_options->__set_decimal_v2(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PARQUET_DICTIONARY_FILTERING: {
        query_options->__set_parquet_dictionary_filtering(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PARQUET_READ_STATISTICS: {
        query_options->__set_parquet_read_statistics(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::DEFAULT_JOIN_DISTRIBUTION_MODE: {
        TJoinDistributionMode::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "default join distribution mode",
            _TJoinDistributionMode_VALUES_TO_NAMES, &enum_type));
        query_options->__set_default_join_distribution_mode(enum_type);
        break;
      }
      case TImpalaQueryOptions::DISABLE_CODEGEN_ROWS_THRESHOLD: {
        StringParser::ParseResult status;
        int val = StringParser::StringToInt<int>(value.c_str(), value.size(), &status);
        if (status != StringParser::PARSE_SUCCESS) {
          return Status(Substitute("Invalid threshold: '$0'.", value));
        }
        if (val < 0) {
          return Status(Substitute(
              "Invalid threshold: '$0'. Only positive values are allowed.", val));
        }
        query_options->__set_disable_codegen_rows_threshold(val);
        break;
      }
      case TImpalaQueryOptions::DEFAULT_SPILLABLE_BUFFER_SIZE: {
        int64_t buffer_size_bytes;
        RETURN_IF_ERROR(
            ParseMemValue(value, "Default spillable buffer size", &buffer_size_bytes));
        if (!BitUtil::IsPowerOf2(buffer_size_bytes)) {
          return Status(
              Substitute("Default spillable buffer size must be a power of two: $0",
                  buffer_size_bytes));
        }
        if (buffer_size_bytes > SPILLABLE_BUFFER_LIMIT) {
          return Status(Substitute(
              "Default spillable buffer size must be less than or equal to: $0",
              SPILLABLE_BUFFER_LIMIT));
        }
        query_options->__set_default_spillable_buffer_size(buffer_size_bytes);
        break;
      }
      case TImpalaQueryOptions::MIN_SPILLABLE_BUFFER_SIZE: {
        int64_t buffer_size_bytes;
        RETURN_IF_ERROR(
            ParseMemValue(value, "Minimum spillable buffer size", &buffer_size_bytes));
        if (!BitUtil::IsPowerOf2(buffer_size_bytes)) {
          return Status(
              Substitute("Minimum spillable buffer size must be a power of two: $0",
                  buffer_size_bytes));
        }
        if (buffer_size_bytes > SPILLABLE_BUFFER_LIMIT) {
          return Status(Substitute(
              "Minimum spillable buffer size must be less than or equal to: $0",
              SPILLABLE_BUFFER_LIMIT));
        }
        query_options->__set_min_spillable_buffer_size(buffer_size_bytes);
        break;
      }
      case TImpalaQueryOptions::MAX_ROW_SIZE: {
        int64_t max_row_size_bytes;
        RETURN_IF_ERROR(ParseMemValue(value, "Max row size", &max_row_size_bytes));
        if (max_row_size_bytes <= 0 || max_row_size_bytes > ROW_SIZE_LIMIT) {
          return Status(
              Substitute("Invalid max row size of $0. Valid sizes are in [$1, $2]", value,
                  1, ROW_SIZE_LIMIT));
        }
        query_options->__set_max_row_size(max_row_size_bytes);
        break;
      }
      case TImpalaQueryOptions::IDLE_SESSION_TIMEOUT: {
        StringParser::ParseResult result;
        const int32_t requested_timeout =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || requested_timeout < 0) {
          return Status(
              Substitute("Invalid idle session timeout: '$0'. "
                         "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_idle_session_timeout(requested_timeout);
        break;
      }
      case TImpalaQueryOptions::COMPUTE_STATS_MIN_SAMPLE_SIZE: {
        int64_t min_sample_size;
        RETURN_IF_ERROR(ParseMemValue(value, "Min sample size", &min_sample_size));
        if (min_sample_size < 0) {
          return Status(
              Substitute("Min sample size must be greater or equal to zero: $0", value));
        }
        query_options->__set_compute_stats_min_sample_size(min_sample_size);
        break;
      }
      case TImpalaQueryOptions::EXEC_TIME_LIMIT_S: {
        StringParser::ParseResult result;
        const int32_t time_limit =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || time_limit < 0) {
          return Status(
              Substitute("Invalid query time limit: '$0'. "
                         "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_exec_time_limit_s(time_limit);
        break;
      }
      case TImpalaQueryOptions::SHUFFLE_DISTINCT_EXPRS: {
        query_options->__set_shuffle_distinct_exprs(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::MAX_MEM_ESTIMATE_FOR_ADMISSION: {
        int64_t bytes_limit;
        RETURN_IF_ERROR(ParseMemValue(
            value, "max memory estimate for admission", &bytes_limit));
        query_options->__set_max_mem_estimate_for_admission(bytes_limit);
        break;
      }
      case TImpalaQueryOptions::THREAD_RESERVATION_LIMIT:
      case TImpalaQueryOptions::THREAD_RESERVATION_AGGREGATE_LIMIT: {
        // Parsing logic is identical for these two options.
        StringParser::ParseResult status;
        int val = StringParser::StringToInt<int>(value.c_str(), value.size(), &status);
        if (status != StringParser::PARSE_SUCCESS) {
          return Status(Substitute("Invalid thread count: '$0'.", value));
        }
        if (val < -1) {
          return Status(Substitute("Invalid thread count: '$0'. "
              "Only -1 and non-negative values are allowed.", val));
        }
        if (option == TImpalaQueryOptions::THREAD_RESERVATION_LIMIT) {
          query_options->__set_thread_reservation_limit(val);
        } else {
          DCHECK_EQ(option, TImpalaQueryOptions::THREAD_RESERVATION_AGGREGATE_LIMIT);
          query_options->__set_thread_reservation_aggregate_limit(val);
        }
        break;
      }
      case TImpalaQueryOptions::KUDU_READ_MODE: {
        TKuduReadMode::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "Kudu read mode",
            _TKuduReadMode_VALUES_TO_NAMES, &enum_type));
        query_options->__set_kudu_read_mode(enum_type);
        break;
      }
      case TImpalaQueryOptions::ALLOW_ERASURE_CODED_FILES: {
        query_options->__set_allow_erasure_coded_files(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::TIMEZONE: {
        // Leading/trailing " and ' characters are stripped because the / character
        // cannot be entered unquoted in some contexts.
        string timezone = value;
        TrimString(&timezone, "'\"");
        timezone = timezone.empty() ? TimezoneDatabase::LocalZoneName() : timezone;
        if (TimezoneDatabase::FindTimezone(timezone) == nullptr) {
          return Status(Substitute("Invalid timezone name '$0'.", timezone));
        }
        query_options->__set_timezone(timezone);
        break;
      }
      case TImpalaQueryOptions::SCAN_BYTES_LIMIT: {
        int64_t bytes_limit;
        RETURN_IF_ERROR(ParseMemValue(value, "query scan bytes limit", &bytes_limit));
        query_options->__set_scan_bytes_limit(bytes_limit);
        break;
      }
      case TImpalaQueryOptions::CPU_LIMIT_S: {
        StringParser::ParseResult result;
        const int64_t cpu_limit_s =
            StringParser::StringToInt<int64_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || cpu_limit_s < 0) {
          return Status(
              Substitute("Invalid CPU limit: '$0'. "
                         "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_cpu_limit_s(cpu_limit_s);
        break;
      }
      case TImpalaQueryOptions::TOPN_BYTES_LIMIT: {
        int64_t topn_bytes_limit;
        RETURN_IF_ERROR(ParseMemValue(value, "topn bytes limit", &topn_bytes_limit));
        query_options->__set_topn_bytes_limit(topn_bytes_limit);
        break;
      }
      case TImpalaQueryOptions::CLIENT_IDENTIFIER: {
        query_options->__set_client_identifier(value);
        break;
      }
      case TImpalaQueryOptions::RESOURCE_TRACE_RATIO: {
        StringParser::ParseResult result;
        const double val =
            StringParser::StringToFloat<double>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || val < 0 || val > 1) {
          return Status(Substitute("Invalid resource trace ratio: '$0'. "
                                   "Only values from 0 to 1 are allowed.",
              value));
        }
        query_options->__set_resource_trace_ratio(val);
      }
      case TImpalaQueryOptions::PLANNER_TESTCASE_MODE: {
        query_options->__set_planner_testcase_mode(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::NUM_REMOTE_EXECUTOR_CANDIDATES: {
        StringParser::ParseResult result;
        const int64_t num_remote_executor_candidates =
            StringParser::StringToInt<int64_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS ||
            num_remote_executor_candidates < 0 || num_remote_executor_candidates > 16) {
          return Status(
              Substitute("$0 is not valid for num_remote_executor_candidates. "
                         "Valid values are in [0, 16].", value));
        }
        query_options->__set_num_remote_executor_candidates(
            num_remote_executor_candidates);
        break;
      }
      case TImpalaQueryOptions::NUM_ROWS_PRODUCED_LIMIT: {
        StringParser::ParseResult result;
        const int64_t num_rows_produced_limit =
            StringParser::StringToInt<int64_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || num_rows_produced_limit < 0) {
          return Status(Substitute("Invalid rows returned limit: '$0'. "
                                   "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_num_rows_produced_limit(num_rows_produced_limit);
        break;
      }
      case TImpalaQueryOptions::DEFAULT_FILE_FORMAT: {
        THdfsFileFormat::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "default file format",
            _THdfsFileFormat_VALUES_TO_NAMES, &enum_type));
        query_options->__set_default_file_format(enum_type);
        break;
      }
      case TImpalaQueryOptions::PARQUET_TIMESTAMP_TYPE: {
        TParquetTimestampType::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "Parquet timestamp type",
            _TParquetTimestampType_VALUES_TO_NAMES, &enum_type));
        query_options->__set_parquet_timestamp_type(enum_type);
        break;
      }
      case TImpalaQueryOptions::PARQUET_READ_PAGE_INDEX: {
        query_options->__set_parquet_read_page_index(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PARQUET_WRITE_PAGE_INDEX: {
        query_options->__set_parquet_write_page_index(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::PARQUET_PAGE_ROW_COUNT_LIMIT: {
        StringParser::ParseResult result;
        const int32_t row_count_limit =
            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || row_count_limit <= 0) {
          return Status("Parquet page row count limit must be a positive integer.");
        }
        query_options->__set_parquet_page_row_count_limit(row_count_limit);
        break;
      }
      case TImpalaQueryOptions::DISABLE_HDFS_NUM_ROWS_ESTIMATE: {
        query_options->__set_disable_hdfs_num_rows_estimate(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::DEFAULT_HINTS_INSERT_STATEMENT: {
        query_options->__set_default_hints_insert_statement(value);
        break;
      }
      case TImpalaQueryOptions::SPOOL_QUERY_RESULTS: {
        query_options->__set_spool_query_results(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::MAX_RESULT_SPOOLING_MEM: {
        int64_t max_result_spooling_mem;
        RETURN_IF_ERROR(ParseMemValue(value, "max result spooling memory",
            &max_result_spooling_mem));
        query_options->__set_max_result_spooling_mem(
            max_result_spooling_mem);
        break;
      }
      case TImpalaQueryOptions::MAX_SPILLED_RESULT_SPOOLING_MEM: {
        int64_t max_spilled_result_spooling_mem;
        RETURN_IF_ERROR(ParseMemValue(value, "max spilled result spooling memory",
            &max_spilled_result_spooling_mem));
        query_options->__set_max_spilled_result_spooling_mem(
            max_spilled_result_spooling_mem);
        break;
      }
      case TImpalaQueryOptions::DEFAULT_TRANSACTIONAL_TYPE: {
        TTransactionalType::type enum_type;
        RETURN_IF_ERROR(GetThriftEnum(value, "default transactional type",
            _TTransactionalType_VALUES_TO_NAMES, &enum_type));
        query_options->__set_default_transactional_type(enum_type);
        break;
      }
      case TImpalaQueryOptions::STATEMENT_EXPRESSION_LIMIT: {
        StringParser::ParseResult result;
        const int32_t statement_expression_limit =
          StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS ||
            statement_expression_limit < MIN_STATEMENT_EXPRESSION_LIMIT) {
          return Status(Substitute("Invalid statement expression limit: $0 "
              "Valid values are in [$1, $2]", value, MIN_STATEMENT_EXPRESSION_LIMIT,
              std::numeric_limits<int32_t>::max()));
        }
        query_options->__set_statement_expression_limit(statement_expression_limit);
        break;
      }
      case TImpalaQueryOptions::MAX_STATEMENT_LENGTH_BYTES: {
        int64_t max_statement_length_bytes;
        RETURN_IF_ERROR(ParseMemValue(value, "max statement length bytes",
            &max_statement_length_bytes));
        if (max_statement_length_bytes < MIN_MAX_STATEMENT_LENGTH_BYTES ||
            max_statement_length_bytes > std::numeric_limits<int32_t>::max()) {
          return Status(Substitute("Invalid maximum statement length: $0 "
              "Valid values are in [$1, $2]", max_statement_length_bytes,
              MIN_MAX_STATEMENT_LENGTH_BYTES, std::numeric_limits<int32_t>::max()));
        }
        query_options->__set_max_statement_length_bytes(max_statement_length_bytes);
        break;
      }
      case TImpalaQueryOptions::DISABLE_DATA_CACHE: {
        query_options->__set_disable_data_cache(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::DISABLE_HBASE_NUM_ROWS_ESTIMATE: {
        query_options->__set_disable_hbase_num_rows_estimate(IsTrue(value));
        break;
      }
      case TImpalaQueryOptions::FETCH_ROWS_TIMEOUT_MS: {
        StringParser::ParseResult result;
        const int64_t requested_timeout =
            StringParser::StringToInt<int64_t>(value.c_str(), value.length(), &result);
        if (result != StringParser::PARSE_SUCCESS || requested_timeout < 0) {
          return Status(
              Substitute("Invalid fetch rows timeout: '$0'. "
                         "Only non-negative numbers are allowed.", value));
        }
        query_options->__set_fetch_rows_timeout_ms(requested_timeout);
        break;
      }
      case TImpalaQueryOptions::NOW_STRING: {
        query_options->__set_now_string(value);
        break;
      }
      case TImpalaQueryOptions::PARQUET_OBJECT_STORE_SPLIT_SIZE: {
        int64_t parquet_object_store_split_size;
        RETURN_IF_ERROR(ParseMemValue(
            value, "parquet object store split size", &parquet_object_store_split_size));
        // The MIN_SYNTHETIC_BLOCK_SIZE from HdfsPartition.java. HdfsScanNode.java forces
        // the block size to be greater than or equal to this value, so reject any
        // attempt to set PARQUET_OBJECT_STORE_SPLIT_SIZE to a value lower than
        // MIN_SYNTHETIC_BLOCK_SIZE.
        int min_synthetic_block_size = 1024 * 1024;
        if (parquet_object_store_split_size < min_synthetic_block_size) {
          return Status(Substitute("Invalid parquet object store split size: '$0'. Must "
                                   "be greater than or equal to '$1'.",
              value, min_synthetic_block_size));
        }
        query_options->__set_parquet_object_store_split_size(
            parquet_object_store_split_size);
        break;
      }
      case TImpalaQueryOptions::MEM_LIMIT_EXECUTORS: {
        // Parse the mem limit spec and validate it.
        int64_t bytes_limit;
        RETURN_IF_ERROR(
            ParseMemValue(value, "query memory limit for executors", &bytes_limit));
        query_options->__set_mem_limit_executors(bytes_limit);
        break;
      }
      case TImpalaQueryOptions::BROADCAST_BYTES_LIMIT: {
        // Parse the broadcast_bytes limit and validate it
        int64_t broadcast_bytes_limit;
        RETURN_IF_ERROR(
            ParseMemValue(value, "broadcast bytes limit for join operations",
                &broadcast_bytes_limit));
        query_options->__set_broadcast_bytes_limit(broadcast_bytes_limit);
        break;
      }
      default:
        if (IsRemovedQueryOption(key)) {
          LOG(WARNING) << "Ignoring attempt to set removed query option '" << key << "'";
          return Status::OK();
        }
        // We hit this DCHECK(false) if we forgot to add the corresponding entry here
        // when we add a new query option.
        LOG(ERROR) << "Missing exec option implementation: " << key;
        DCHECK(false);
        break;
    }
    if (set_query_options_mask != NULL) {
      DCHECK_LT(option, set_query_options_mask->size());
      set_query_options_mask->set(option);
    }
  }
  return Status::OK();
}

Status impala::ParseQueryOptions(const string& options, TQueryOptions* query_options,
    QueryOptionsMask* set_query_options_mask) {
  if (options.length() == 0) return Status::OK();
  vector<string> kv_pairs;
  split(kv_pairs, options, is_any_of(","), token_compress_on);
  // Construct an error status which is used to aggregate errors encountered during
  // parsing. It is only returned if the number of error details is greater than 0.
  Status errorStatus = Status::Expected("Errors parsing query options");
  for (string& kv_string: kv_pairs) {
    trim(kv_string);
    if (kv_string.length() == 0) continue;
    vector<string> key_value;
    split(key_value, kv_string, is_any_of("="), token_compress_on);
    if (key_value.size() != 2) {
      errorStatus.MergeStatus(
          Status(Substitute("Invalid configuration option '$0'.", kv_string)));
      continue;
    }
    errorStatus.MergeStatus(SetQueryOption(key_value[0], key_value[1], query_options,
        set_query_options_mask));
  }
  if (errorStatus.msg().details().size() > 0) return errorStatus;
  return Status::OK();
}

Status impala::ValidateQueryOptions(TQueryOptions* query_options) {
  // Validate that max_result_spooling_mem <=
  // max_spilled_result_spooling_mem (a value of 0 means memory is unbounded).
  int64_t max_mem = query_options->max_result_spooling_mem;
  int64_t max_spilled_mem = query_options->max_spilled_result_spooling_mem;
  if (max_mem == 0 && max_spilled_mem != 0) {
    return Status("If max_result_spooling_mem is set to 0 (unbounded) "
                  "max_spilled_result_spooling_mem must be set to 0 (unbounded) as "
                  "well.");
  }
  if (max_spilled_mem != 0 && max_spilled_mem < max_mem) {
    return Status(Substitute("max_spilled_result_spooling_mem '$0' must be greater than "
                             "max_result_spooling_mem '$1'",
        max_spilled_mem, max_mem));
  }
  return Status::OK();
}

void impala::PopulateQueryOptionLevels(QueryOptionLevels* query_option_levels)
{
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
  {\
    (*query_option_levels)[#ENUM] = LEVEL;\
  }
#define REMOVED_QUERY_OPT_FN(NAME, ENUM)\
  {\
    (*query_option_levels)[#ENUM] = TQueryOptionLevel::REMOVED;\
  }
  QUERY_OPTS_TABLE
  QUERY_OPT_FN(support_start_over, SUPPORT_START_OVER, TQueryOptionLevel::ADVANCED)
#undef QUERY_OPT_FN
#undef REMOVED_QUERY_OPT_FN
}