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

# Functional tests running EXPLAIN statements.
#
import re

from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.skip import SkipIfLocal, SkipIfNotHdfsMinicluster, SkipIfEC
from tests.util.filesystem_utils import WAREHOUSE

# Tests the different explain levels [0-3] on a few queries.
# TODO: Clean up this test to use an explain level test dimension and appropriate
# result sub-sections for the expected explain plans.
@SkipIfEC.fix_later
class TestExplain(ImpalaTestSuite):
  # Value for the num_scanner_threads query option to ensure that the memory estimates of
  # scan nodes are consistent even when run on machines with different numbers of cores.
  NUM_SCANNER_THREADS = 1

  @classmethod
  def get_workload(self):
    return 'functional-query'

  @classmethod
  def add_test_dimensions(cls):
    super(TestExplain, cls).add_test_dimensions()
    cls.ImpalaTestMatrix.add_constraint(lambda v:\
        v.get_value('table_format').file_format == 'text' and\
        v.get_value('table_format').compression_codec == 'none' and\
        v.get_value('exec_option')['batch_size'] == 0 and\
        v.get_value('exec_option')['disable_codegen'] == False and\
        v.get_value('exec_option')['num_nodes'] != 1)

  @SkipIfNotHdfsMinicluster.plans
  def test_explain_level0(self, vector):
    vector.get_value('exec_option')['num_scanner_threads'] = self.NUM_SCANNER_THREADS
    vector.get_value('exec_option')['explain_level'] = 0
    self.run_test_case('QueryTest/explain-level0', vector)

  @SkipIfNotHdfsMinicluster.plans
  def test_explain_level1(self, vector):
    vector.get_value('exec_option')['num_scanner_threads'] = self.NUM_SCANNER_THREADS
    vector.get_value('exec_option')['explain_level'] = 1
    self.run_test_case('QueryTest/explain-level1', vector)

  @SkipIfNotHdfsMinicluster.plans
  def test_explain_level2(self, vector):
    vector.get_value('exec_option')['num_scanner_threads'] = self.NUM_SCANNER_THREADS
    vector.get_value('exec_option')['explain_level'] = 2
    self.run_test_case('QueryTest/explain-level2', vector)

  @SkipIfNotHdfsMinicluster.plans
  def test_explain_level3(self, vector):
    vector.get_value('exec_option')['num_scanner_threads'] = self.NUM_SCANNER_THREADS
    vector.get_value('exec_option')['explain_level'] = 3
    self.run_test_case('QueryTest/explain-level3', vector)

  @staticmethod
  def check_row_size_and_cardinality(query_result, expected_row_size=None,
                                     expected_cardinality=None):
    regex = re.compile('tuple-ids=.+ row-size=(\d+)B cardinality=(.*)')
    found_match = False
    for res in query_result:
      m = regex.match(res.strip())
      if m:
        found_match = True
        assert len(m.groups()) == 2
        if expected_row_size:
          assert m.groups()[0] == expected_row_size
        if expected_cardinality:
          assert m.groups()[1] == expected_cardinality
    assert found_match, query_result

  def test_explain_validate_cardinality_estimates(self, vector, unique_database):
    # Tests that the cardinality estimates are correct for partitioned tables.
    # TODO Cardinality estimation tests should eventually be part of the planner tests.
    # TODO Remove this test
    db_name = 'functional'
    tbl_name = 'alltypes'

    def check_cardinality(query_result, expected_cardinality):
      self.check_row_size_and_cardinality(
          query_result, expected_cardinality=expected_cardinality)

    # All partitions are filtered out, cardinality should be 0.
    result = self.execute_query("explain select * from %s.%s where year = 1900" % (
        db_name, tbl_name), query_options={'explain_level':3})
    check_cardinality(result.data, '0')

    # Half of the partitions are filtered out, cardinality should be 3650.
    result = self.execute_query("explain select * from %s.%s where year = 2010" % (
        db_name, tbl_name), query_options={'explain_level':3})
    check_cardinality(result.data, '3.65K')

    # None of the partitions are filtered out, cardinality should be 7300.
    result = self.execute_query("explain select * from %s.%s" % (db_name, tbl_name),
        query_options={'explain_level':3})
    check_cardinality(result.data, '7.30K')

    # Create a partitioned table with a mixed set of available stats,
    mixed_tbl = unique_database + ".t"
    self.execute_query(
      "create table %s (c int) partitioned by (p int)" % mixed_tbl)
    self.execute_query(
      "insert into table %s partition (p) values(1,1),(2,2),(3,3)" % mixed_tbl)
    # Set the number of rows at the table level.
    self.execute_query(
      "alter table %s set tblproperties('numRows'='100')" % mixed_tbl)
    # Should fall back to table-level cardinality when partitions lack stats.
    result = self.execute_query("explain select * from %s" % mixed_tbl,
        query_options={'explain_level':3})
    check_cardinality(result.data, '100')
    # Should fall back to table-level cardinality, even for a subset of partitions,
    result = self.execute_query("explain select * from %s where p = 1" % mixed_tbl,
        query_options={'explain_level':3})
    check_cardinality(result.data, '100')
    # Set the number of rows for a single partition.
    self.execute_query(
      "alter table %s partition(p=1) set tblproperties('numRows'='50')" % mixed_tbl)
    # Use partition stats when availabe. Partitions without stats are ignored.
    result = self.execute_query("explain select * from %s" % mixed_tbl,
        query_options={'explain_level':3})
    check_cardinality(result.data, '50')
    # Fall back to table-level stats when no selected partitions have stats.
    result = self.execute_query("explain select * from %s where p = 2" % mixed_tbl,
        query_options={'explain_level':3})
    check_cardinality(result.data, '100')

  def test_explain_row_size_estimates(self, vector, unique_database):
    """ Tests that EXPLAIN returns the expected row sizes with and without stats.

    Planner tests is probably a more logical place for this, but covering string avg_size
    handling end-to-end seemed easier here.

    Note that row sizes do not include the null indicator bytes, so actual tuple sizes
    are a bit larger. """
    def check_row_size(query_result, expected_row_size):
      self.check_row_size_and_cardinality(
          query_result, expected_row_size=expected_row_size)

    def execute_explain(query):
      return self.execute_query("explain " + query, query_options={'explain_level': 3})

    FQ_TBL_NAME = unique_database + ".t"
    self.execute_query("create table %s (i int, s string)" % FQ_TBL_NAME)
    # Fill the table with data that leads to avg_size of 4 for 's'.
    self.execute_query("insert into %s values (1, '123'), (2, '12345')" % FQ_TBL_NAME)

    # Always use slot size for fixed sized types.
    result = execute_explain("select i from %s" % FQ_TBL_NAME)
    check_row_size(result.data, '4')

    # If there are no stats, use slot size for variable length types.
    result = execute_explain("select s from %s" % FQ_TBL_NAME)
    check_row_size(result.data, "12")

    self.execute_query("compute stats %s" % FQ_TBL_NAME)

    # Always use slot size for fixed sized types.
    result = execute_explain("select i from %s" % FQ_TBL_NAME)
    check_row_size(result.data, '4')

    # If there are no stats, use slot size  + avg_size for variable length types.
    result = execute_explain("select s from %s" % FQ_TBL_NAME)
    check_row_size(result.data, "16")


class TestExplainEmptyPartition(ImpalaTestSuite):
  TEST_DB_NAME = "imp_1708"

  def setup_method(self, method):
    self.cleanup_db(self.TEST_DB_NAME)
    self.execute_query("create database if not exists {0} location '{1}/{0}.db'"
        .format(self.TEST_DB_NAME, WAREHOUSE))

  def teardown_method(self, method):
    self.cleanup_db(self.TEST_DB_NAME)

  @SkipIfLocal.hdfs_client
  def test_non_empty_partition_0_rows(self):
    """Regression test for IMPALA-1708: if a partition has 0 rows but > 0 files after
    COMPUTE STATS, don't warn the user about missing stats. The files are probably
    corrupted, or used for something else."""
    self.client.execute("SET EXPLAIN_LEVEL=3")
    self.client.execute("CREATE TABLE %s.empty_partition (col int) "
                        "partitioned by (p int)" % self.TEST_DB_NAME)
    self.client.execute(
      "ALTER TABLE %s.empty_partition ADD PARTITION (p=NULL)" % self.TEST_DB_NAME)
    # Put an empty file in the partition so we have > 0 files, but 0 rows
    self.filesystem_client.create_file(
        "test-warehouse/%s.db/empty_partition/p=__HIVE_DEFAULT_PARTITION__/empty" %
        self.TEST_DB_NAME, "")
    self.client.execute("REFRESH %s.empty_partition" % self.TEST_DB_NAME)
    self.client.execute("COMPUTE STATS %s.empty_partition" % self.TEST_DB_NAME)
    assert "NULL\t0\t1" in str(
        self.client.execute("SHOW PARTITIONS %s.empty_partition" % self.TEST_DB_NAME))
    assert "missing relevant table and/or column statistics" not in str(
        self.client.execute(
            "EXPLAIN SELECT * FROM %s.empty_partition" % self.TEST_DB_NAME))

    # Now add a partition with some data (so it gets selected into the scan), to check
    # that its lack of stats is correctly identified
    self.client.execute(
      "ALTER TABLE %s.empty_partition ADD PARTITION (p=1)" % self.TEST_DB_NAME)
    self.filesystem_client.create_file("test-warehouse/%s.db/empty_partition/p=1/rows" %
                                 self.TEST_DB_NAME, "1")
    self.client.execute("REFRESH %s.empty_partition" % self.TEST_DB_NAME)
    explain_result = str(
      self.client.execute("EXPLAIN SELECT * FROM %s.empty_partition" % self.TEST_DB_NAME))
    assert "missing relevant table and/or column statistics" in explain_result
    # Also test IMPALA-1530 - adding the number of partitions missing stats
    assert "partitions: 1/2 " in explain_result