text: # 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.
  
  import getpass
  import grp
  import pwd
  import pytest
  import re
  
  from tests.common.impala_test_suite import ImpalaTestSuite
  from tests.common.parametrize import UniqueDatabase
  from tests.common.skip import (SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon,
      SkipIfLocal, SkipIfDockerizedCluster)
  from tests.util.filesystem_utils import WAREHOUSE, get_fs_path, IS_S3
  
  @SkipIfLocal.hdfs_client
  class TestInsertBehaviour(ImpalaTestSuite):
    """Tests for INSERT behaviour that isn't covered by checking query results"""
    @classmethod
    def get_workload(self):
      return 'functional-query'
  
    TEST_DB_NAME = "insert_empty_result_db"
  
    def setup_method(self, method):
      # cleanup and create a fresh test database
      if method.__name__ == "test_insert_select_with_empty_resultset":
        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):
      if method.__name__ == "test_insert_select_with_empty_resultset":
        self.cleanup_db(self.TEST_DB_NAME)
  
    @SkipIfADLS.eventually_consistent
    @pytest.mark.execute_serially
    def test_insert_removes_staging_files(self):
      TBL_NAME = "insert_overwrite_nopart"
      insert_staging_dir = ("test-warehouse/functional.db/%s/"
          "_impala_insert_staging" % TBL_NAME)
      self.filesystem_client.delete_file_dir(insert_staging_dir, recursive=True)
      self.client.execute("INSERT OVERWRITE functional.%s "
                          "SELECT int_col FROM functional.tinyinttable" % TBL_NAME)
      ls = self.filesystem_client.ls(insert_staging_dir)
      assert len(ls) == 0
  
    @pytest.mark.execute_serially
    def test_insert_preserves_hidden_files(self):
      """Test that INSERT OVERWRITE preserves hidden files in the root table directory"""
      TBL_NAME = "insert_overwrite_nopart"
      table_dir = "test-warehouse/functional.db/%s/" % TBL_NAME
      hidden_file_locations = [".hidden", "_hidden"]
      dir_locations = ["dir", ".hidden_dir"]
  
      for dir_ in dir_locations:
        self.filesystem_client.make_dir(table_dir + dir_)
  
      # We do this here because the above 'make_dir' call doesn't make a directory for S3.
      for dir_ in dir_locations:
        self.filesystem_client.create_file(
            table_dir + dir_ + '/' + hidden_file_locations[0] , '', overwrite=True)
  
      for file_ in hidden_file_locations:
        self.filesystem_client.create_file(table_dir + file_, '', overwrite=True)
  
      self.client.execute("INSERT OVERWRITE functional.%s"
                          " SELECT int_col FROM functional.tinyinttable" % TBL_NAME)
  
      for file_ in hidden_file_locations:
        assert self.filesystem_client.exists(table_dir + file_), "Hidden file {0} was " \
            "unexpectedly deleted by INSERT OVERWRITE".format(table_dir + file_)
  
      for dir_ in dir_locations:
        assert self.filesystem_client.exists(table_dir + dir_), "Directory {0} was " \
            "unexpectedly deleted by INSERT OVERWRITE".format(table_dir + dir_)
  
    def test_insert_ascii_nulls(self, unique_database):
      TBL_NAME = '`{0}`.`null_insert`'.format(unique_database)
      self.execute_query_expect_success(self.client, "DROP TABLE IF EXISTS %s" % TBL_NAME)
      self.execute_query_expect_success(self.client, "create table %s as select '\0' s"
          % TBL_NAME)
      result = self.execute_query_expect_success(self.client,
          "SELECT LENGTH(s) FROM %s" % TBL_NAME)
      assert int(result.get_data()) == 1
  
    @UniqueDatabase.parametrize(name_prefix='test_insert_alter_partition_location_db')
    def test_insert_alter_partition_location(self, unique_database):
      """Test that inserts after changing the location of a partition work correctly,
      including the creation of a non-existant partition dir"""
      part_dir = "tmp/{0}".format(unique_database)
      qualified_part_dir = get_fs_path('/' + part_dir)
      table_name = "`{0}`.`insert_alter_partition_location`".format(unique_database)
  
      self.execute_query_expect_success(self.client, "DROP TABLE IF EXISTS %s" % table_name)
      self.filesystem_client.delete_file_dir(part_dir, recursive=True)
  
      self.execute_query_expect_success(
          self.client,
          "CREATE TABLE  %s (c int) PARTITIONED BY (p int)" % table_name)
      self.execute_query_expect_success(
          self.client,
          "ALTER TABLE %s ADD PARTITION(p=1)" % table_name)
      self.execute_query_expect_success(
          self.client,
          "ALTER TABLE %s PARTITION(p=1) SET LOCATION '%s'" % (table_name,
                                                               qualified_part_dir))
      self.execute_query_expect_success(
          self.client,
          "INSERT OVERWRITE %s PARTITION(p=1) VALUES(1)" % table_name)
  
      result = self.execute_query_expect_success(
          self.client,
          "SELECT COUNT(*) FROM %s" % table_name)
      assert int(result.get_data()) == 1
  
      # Should have created the partition dir, which should contain exactly one file (not in
      # a subdirectory)
      assert len(self.filesystem_client.ls(part_dir)) == 1
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    @pytest.mark.xfail(run=False, reason="Fails intermittently on test clusters")
    @pytest.mark.execute_serially
    def test_insert_inherit_acls(self):
      """Check that ACLs are inherited when we create new partitions"""
  
      ROOT_ACL = "default:group:dummy_group:rwx"
      TEST_ACL = "default:group:impala_test_users:r-x"
  
      def check_has_acls(part, acl=TEST_ACL):
        path = "test-warehouse/functional.db/insert_inherit_acls/" + part
        result = self.hdfs_client.getacl(path)
        assert acl in result['AclStatus']['entries']
  
      # Add a spurious ACL to functional.db directory
      self.hdfs_client.setacl("test-warehouse/functional.db", ROOT_ACL)
  
      self.execute_query_expect_success(self.client, "DROP TABLE IF EXISTS"
                                        " functional.insert_inherit_acls")
      self.execute_query_expect_success(self.client, "CREATE TABLE "
                                        "functional.insert_inherit_acls (col int)"
                                        " PARTITIONED BY (p1 int, p2 int, p3 int)")
  
      # Check that table creation inherited the ACL
      check_has_acls("", ROOT_ACL)
  
      self.execute_query_expect_success(self.client, "ALTER TABLE "
                                        "functional.insert_inherit_acls ADD PARTITION"
                                        "(p1=1, p2=1, p3=1)")
  
      check_has_acls("p1=1", ROOT_ACL)
      check_has_acls("p1=1/p2=1", ROOT_ACL)
      check_has_acls("p1=1/p2=1/p3=1", ROOT_ACL)
  
      self.hdfs_client.setacl(
          "test-warehouse/functional.db/insert_inherit_acls/p1=1/", TEST_ACL)
  
      self.execute_query_expect_success(self.client, "INSERT INTO "
                                        "functional.insert_inherit_acls "
                                        "PARTITION(p1=1, p2=2, p3=2) VALUES(1)")
  
      check_has_acls("p1=1/p2=2/")
      check_has_acls("p1=1/p2=2/p3=2")
  
      # Check that SETACL didn't cascade down to children (which is more to do with HDFS
      # than Impala, but worth asserting here)
      check_has_acls("p1=1/p2=1", ROOT_ACL)
      check_has_acls("p1=1/p2=1/p3=1", ROOT_ACL)
  
      # Change ACLs on p1=1,p2=2 and create a new leaf at p3=30
      self.hdfs_client.setacl(
          "test-warehouse/functional.db/insert_inherit_acls/p1=1/p2=2/",
          "default:group:new_leaf_group:-w-")
  
      self.execute_query_expect_success(self.client, "INSERT INTO "
                                        "functional.insert_inherit_acls "
                                        "PARTITION(p1=1, p2=2, p3=30) VALUES(1)")
      check_has_acls("p1=1/p2=2/p3=30", "default:group:new_leaf_group:-w-")
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    def test_insert_file_permissions(self, unique_database):
      """Test that INSERT correctly respects file permission (minimum ACLs)"""
      table = "`{0}`.`insert_acl_permissions`".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_acl_permissions".format(unique_database)
  
      insert_query = "INSERT INTO {0} VALUES(1)".format(table)
      refresh_query = "REFRESH {0}".format(table)
  
      self.execute_query_expect_success(self.client,
                                        "DROP TABLE IF EXISTS {0}".format(table))
      self.execute_query_expect_success(
          self.client, "CREATE TABLE {0} (col int)".format(table))
  
      # Check that a simple insert works
      self.execute_query_expect_success(self.client, insert_query)
  
      # Remove the permission to write and confirm that INSERTs won't work
      self.hdfs_client.setacl(table_path, "user::r-x,group::r-x,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Now add group access, still should fail (because the user will match and take
      # priority)
      self.hdfs_client.setacl(table_path, "user::r-x,group::rwx,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Now make the target directory non-writable with posix permissions, but writable with
      # ACLs (ACLs should take priority). Note: chmod affects ACLs (!) so it has to be done
      # first.
      self.hdfs_client.chmod(table_path, "000")
      self.hdfs_client.setacl(table_path, "user::rwx,group::r-x,other::r-x")
  
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_success(self.client, insert_query)
  
      # Finally, change the owner
      self.hdfs_client.chown(table_path, "another_user", "another_group")
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be unwritable because 'other' ACLs don't allow writes
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Give write perms back to 'other'
      self.hdfs_client.setacl(table_path, "user::rwx,group::r-x,other::rwx")
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be writable because 'other' ACLs allow writes
      self.execute_query_expect_success(self.client, insert_query)
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    def test_mixed_partition_permissions(self, unique_database):
      """
      Test that INSERT and LOAD DATA into explicit partitions is allowed even
      if some partitions exist without appropriate permissions.
      """
      table = "`{0}`.`insert_partition_perms`".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_partition_perms".format(unique_database)
  
      def partition_path(partition):
        return "%s/p=%s" % (table_path, partition)
  
      def insert_query(partition):
        return "INSERT INTO {0} PARTITION (p='{1}') VALUES(1)".format(table, partition)
  
      def load_data(query_fn, partition):
        path = "tmp/{0}/data_file".format(unique_database)
        self.hdfs_client.delete_file_dir(path)
        self.hdfs_client.create_file(path, "1")
        q = "LOAD DATA INPATH '/{0}' INTO TABLE {1} PARTITION (p='{2}')".format(
            path, table, partition)
        return query_fn(self.client, q)
  
      insert_dynamic_partition = \
          "INSERT INTO {0} (col, p) SELECT col, p from {0}".format(table)
  
      insert_dynamic_partition_2 = \
          "INSERT INTO {0} PARTITION(p) SELECT col, p from {0}".format(table)
  
      # TODO(todd): REFRESH does not seem to refresh the permissions of existing
      # partitions -- need to fully reload the metadata to do so.
      invalidate_query = "INVALIDATE METADATA {0}".format(table)
  
      self.execute_query_expect_success(self.client,
                                        "DROP TABLE IF EXISTS {0}".format(table))
      self.execute_query_expect_success(
          self.client, "CREATE TABLE {0} (col int) PARTITIONED BY (p string)".format(table))
  
      # Check that a simple insert works
      self.execute_query_expect_success(self.client, insert_query("initial_part"))
  
      # Add a partition with no write permissions
      self.execute_query_expect_success(
          self.client,
          "ALTER TABLE {0} ADD PARTITION (p=\"no_write\")".format(table))
      self.hdfs_client.chown(partition_path("no_write"), "another_user", "another_group")
      self.execute_query_expect_success(self.client, invalidate_query)
  
      # Check that inserts into the new partition do not work.
      err = self.execute_query_expect_failure(self.client, insert_query("no_write"))
      assert re.search(r'Impala does not have WRITE access.*p=no_write', str(err))
  
      err = load_data(self.execute_query_expect_failure, "no_write")
      assert re.search(r'Impala does not have WRITE access.*p=no_write', str(err))
  
      # Inserts into the existing partition should still work
      self.execute_query_expect_success(self.client, insert_query("initial_part"))
      load_data(self.execute_query_expect_success, "initial_part")
  
      # Inserts into newly created partitions should work.
      self.execute_query_expect_success(self.client, insert_query("added_part"))
  
      # TODO(IMPALA-7313): loading data cannot currently add partitions, so this
      # is commented out. If that behavior is changed, uncomment this.
      # load_data(self.execute_query_expect_success, "added_part_load")
  
      # Inserts into dynamic partitions should fail.
      err = self.execute_query_expect_failure(self.client, insert_dynamic_partition)
      assert re.search(r'Impala does not have WRITE access.*p=no_write', str(err))
      err = self.execute_query_expect_failure(self.client, insert_dynamic_partition_2)
      assert re.search(r'Impala does not have WRITE access.*p=no_write', str(err))
  
      # If we make the top-level table directory read-only, we should still be able to
      # insert into existing writable partitions.
      self.hdfs_client.chown(table_path, "another_user", "another_group")
      self.execute_query_expect_success(self.client, insert_query("added_part"))
      load_data(self.execute_query_expect_success, "added_part")
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    def test_readonly_table_dir(self, unique_database):
      """
      Test that, if a partitioned table has a read-only base directory,
      the frontend rejects queries that may attempt to create new
      partitions.
      """
      table = "`{0}`.`insert_partition_perms`".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_partition_perms".format(unique_database)
      insert_dynamic_partition = \
          "INSERT INTO {0} (col, p) SELECT col, p from {0}".format(table)
  
      def insert_query(partition):
        return "INSERT INTO {0} PARTITION (p='{1}') VALUES(1)".format(table, partition)
  
      self.execute_query_expect_success(self.client,
                                        "DROP TABLE IF EXISTS {0}".format(table))
      self.execute_query_expect_success(
          self.client, "CREATE TABLE {0} (col int) PARTITIONED BY (p string)".format(table))
  
      # If we make the top-level table directory read-only, we should not be able
      # to create new partitions, either explicitly, or dynamically.
      self.hdfs_client.chown(table_path, "another_user", "another_group")
      self.execute_query_expect_success(self.client, "INVALIDATE METADATA %s" % table)
      err = self.execute_query_expect_failure(self.client, insert_dynamic_partition)
      assert re.search(r'Impala does not have WRITE access.*' + table_path, str(err))
      err = self.execute_query_expect_failure(self.client, insert_query("some_new_part"))
      assert re.search(r'Impala does not have WRITE access.*' + table_path, str(err))
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    @SkipIfDockerizedCluster.insert_acls
    def test_insert_acl_permissions(self, unique_database):
      """Test that INSERT correctly respects ACLs"""
      table = "`{0}`.`insert_acl_permissions`".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_acl_permissions".format(unique_database)
  
      insert_query = "INSERT INTO %s VALUES(1)" % table
      refresh_query = "REFRESH {0}".format(table)
  
      self.execute_query_expect_success(self.client,
                                        "DROP TABLE IF EXISTS {0}".format(table))
      self.execute_query_expect_success(self.client,
                                        "CREATE TABLE {0} (col int)".format(table))
  
      # Check that a simple insert works
      self.execute_query_expect_success(self.client, insert_query)
  
      user = getpass.getuser()
      group = grp.getgrgid(pwd.getpwnam(user).pw_gid).gr_name
      # First, change the owner to someone other than user who runs impala service
      self.hdfs_client.chown(table_path, "another_user", group)
  
      # Remove the permission to write and confirm that INSERTs won't work
      self.hdfs_client.setacl(table_path,
                              "user::r-x,user:" + user + ":r-x,group::r-x,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Add the permission to write. if we're not the owner of the file, INSERTs should work
      self.hdfs_client.setacl(table_path,
                              "user::r-x,user:" + user + ":rwx,group::r-x,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_success(self.client, insert_query)
  
      # Now add group access, still should fail (because the user will match and take
      # priority)
      self.hdfs_client.setacl(table_path,
                              "user::r-x,user:" + user + ":r-x,group::rwx,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Check that the mask correctly applies to the anonymous group ACL
      self.hdfs_client.setacl(table_path, "user::r-x,group::rwx,other::rwx,mask::r--")
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be unwritable because mask applies to unnamed group and disables writing
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Check that the mask correctly applies to the named user ACL
      self.hdfs_client.setacl(table_path, "user::r-x,user:" + user +
                              ":rwx,group::r-x,other::rwx,mask::r--")
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be unwritable because mask applies to named user and disables writing
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Now make the target directory non-writable with posix permissions, but writable with
      # ACLs (ACLs should take priority). Note: chmod affects ACLs (!) so it has to be done
      # first.
      self.hdfs_client.chmod(table_path, "000")
      self.hdfs_client.setacl(table_path, "user::rwx,user:foo:rwx,group::rwx,other::r-x")
  
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_success(self.client, insert_query)
  
      # Finally, change the owner/group
      self.hdfs_client.chown(table_path, "test_user", "invalid")
  
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be unwritable because 'other' ACLs don't allow writes
      self.execute_query_expect_failure(self.client, insert_query)
  
      # Give write perms back to 'other'
      self.hdfs_client.setacl(table_path, "user::r-x,user:foo:rwx,group::r-x,other::rwx")
      self.execute_query_expect_success(self.client, refresh_query)
      # Should be writable because 'other' ACLs allow writes
      self.execute_query_expect_success(self.client, insert_query)
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    def test_load_permissions(self, unique_database):
      # We rely on test_insert_acl_permissions() to exhaustively check that ACL semantics
      # are correct. Here we just validate that LOADs can't be done when we cannot read from
      # or write to the src directory, or write to the dest directory.
  
      table = "`{0}`.`load_acl_permissions`".format(unique_database)
      table_path = "test-warehouse/{0}.db/load_acl_permissions".format(unique_database)
  
      file_path = "tmp/{0}".format(unique_database)
      file_name = "%s/impala_data_file" % file_path
  
      load_file_query = "LOAD DATA INPATH '/%s' INTO TABLE %s" % (file_name, table)
      load_dir_query = "LOAD DATA INPATH '/%s' INTO TABLE %s" % (file_path, table)
  
      refresh_query = "REFRESH {0}".format(table)
  
      self.hdfs_client.make_dir(file_path)
  
      self.hdfs_client.setacl(file_path, "user::rwx,group::rwx,other::---")
      self.execute_query_expect_success(self.client,
                                        "DROP TABLE IF EXISTS {0}".format(table))
      self.execute_query_expect_success(
          self.client, "CREATE TABLE {0} (col int)".format(table))
      self.hdfs_client.delete_file_dir(file_name)
      self.hdfs_client.create_file(file_name, "1")
  
      self.execute_query_expect_success(self.client, load_file_query)
  
      # Now remove write perms from the source directory
      self.hdfs_client.create_file(file_name, "1")
      self.hdfs_client.setacl(file_path, "user::---,group::---,other::---")
      self.hdfs_client.setacl(table_path, "user::rwx,group::r-x,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, load_file_query)
      self.execute_query_expect_failure(self.client, load_dir_query)
  
      # Remove write perms from target
      self.hdfs_client.setacl(file_path, "user::rwx,group::rwx,other::rwx")
      self.hdfs_client.setacl(table_path, "user::r-x,group::r-x,other::r-x")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, load_file_query)
      self.execute_query_expect_failure(self.client, load_dir_query)
  
      # Finally remove read perms from file itself
      self.hdfs_client.setacl(file_name, "user::-wx,group::rwx,other::rwx")
      self.hdfs_client.setacl(table_path, "user::rwx,group::rwx,other::rwx")
      self.execute_query_expect_success(self.client, refresh_query)
      self.execute_query_expect_failure(self.client, load_file_query)
      # We expect this to succeed, it's not an error if all files in the dir cannot be read
      self.execute_query_expect_success(self.client, load_dir_query)
  
    @SkipIfADLS.eventually_consistent
    @pytest.mark.execute_serially
    def test_insert_select_with_empty_resultset(self):
      """Test insert/select query won't trigger partition directory or zero size data file
      creation if the resultset of select is empty."""
      def check_path_exists(path, should_exist):
        assert self.filesystem_client.exists(path) == should_exist, "file/dir '{0}' " \
            "should {1}exist but does {2}exist.".format(
                path, '' if should_exist else 'not ', 'not ' if should_exist else '')
  
      db_path = "test-warehouse/%s.db/" % self.TEST_DB_NAME
      table_path = db_path + "test_insert_empty_result"
      partition_path = "{0}/year=2009/month=1".format(table_path)
      check_path_exists(table_path, False)
  
      table_name = self.TEST_DB_NAME + ".test_insert_empty_result"
      self.execute_query_expect_success(
          self.client,
          "CREATE TABLE %s (id INT, col INT) PARTITIONED BY (year INT, month "
          "INT)" % table_name)
      check_path_exists(table_path, True)
      check_path_exists(partition_path, False)
  
      # Run an insert/select stmt that returns an empty resultset.
      insert_query = ("INSERT INTO TABLE {0} PARTITION(year=2009, month=1)"
                      "select 1, 1 from {0} LIMIT 0".format(table_name))
      self.execute_query_expect_success(self.client, insert_query)
      # Partition directory is created
      check_path_exists(partition_path, True)
  
      # Insert one record
      insert_query_one_row = ("INSERT INTO TABLE %s PARTITION(year=2009, month=1) "
                              "values(2, 2)" % table_name)
      self.execute_query_expect_success(self.client, insert_query_one_row)
      # Partition directory should be created with one data file
      check_path_exists(partition_path, True)
      ls = (self.filesystem_client.ls(partition_path))
      assert len(ls) == 1
  
      # Run an insert/select statement that returns an empty resultset again
      self.execute_query_expect_success(self.client, insert_query)
      # No new data file should be created
      new_ls = self.filesystem_client.ls(partition_path)
      assert len(new_ls) == 1
      assert new_ls == ls
  
      # Run an insert overwrite/select that returns an empty resultset
      insert_query = ("INSERT OVERWRITE {0} PARTITION(year=2009, month=1)"
                      " select 1, 1 from  {0} LIMIT 0".format(table_name))
      self.execute_query_expect_success(self.client, insert_query)
      # Data file should be deleted
      new_ls2 = self.filesystem_client.ls(partition_path)
      assert len(new_ls2) == 0
      assert new_ls != new_ls2
  
      # Test for IMPALA-2008 insert overwrite to an empty table with empty dataset
      empty_target_tbl = "test_overwrite_with_empty_target"
      create_table = "create table %s.%s (id INT, col INT)" % (self.TEST_DB_NAME,
                                                               empty_target_tbl)
      self.execute_query_expect_success(self.client, create_table)
      insert_query = ("INSERT OVERWRITE {0}.{1} select 1, 1 from  {0}.{1} LIMIT 0"
                      .format(self.TEST_DB_NAME, empty_target_tbl))
      self.execute_query_expect_success(self.client, insert_query)
  
      # Delete target table directory, query should fail with
      # "No such file or directory" error
      target_table_path = "%s%s" % (db_path, empty_target_tbl)
      self.filesystem_client.delete_file_dir(target_table_path, recursive=True)
      self.execute_query_expect_failure(self.client, insert_query)
  
    @SkipIfS3.hdfs_acls
    @SkipIfABFS.hdfs_acls
    @SkipIfADLS.hdfs_acls
    @SkipIfIsilon.hdfs_acls
    @SkipIfDockerizedCluster.insert_acls
    def test_multiple_group_acls(self, unique_database):
      """Test that INSERT correctly respects multiple group ACLs"""
      table = "`{0}`.`insert_group_acl_permissions`".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_group_acl_permissions".format(
          unique_database)
      insert_query = "INSERT INTO %s VALUES(1)" % table
  
      self.execute_query_expect_success(self.client, "DROP TABLE IF EXISTS " + table)
      self.execute_query_expect_success(self.client, "CREATE TABLE %s (col int)" % table)
  
      user = getpass.getuser()
      test_user = "test_user"
      # Get the list of all groups of user except the user's owning group.
      owning_group = grp.getgrgid(pwd.getpwnam(user).pw_gid).gr_name
      groups = [g.gr_name for g in grp.getgrall() if user in g.gr_mem]
      if (len(groups) < 1):
        pytest.xfail(reason="Cannot run test, user belongs to only one group.")
  
      # First, change the owner to someone other than user who runs impala service
      self.hdfs_client.chown(table_path, "another_user", owning_group)
  
      # Set two group ACLs, one contains requested permission, the other doesn't.
      self.hdfs_client.setacl(
          table_path,
          "user::r-x,user:{0}:r-x,group::---,group:{1}:rwx,"
          "other::r-x".format(test_user, groups[0]))
      self.execute_query_expect_success(self.client, "REFRESH " + table)
      self.execute_query_expect_success(self.client, insert_query)
  
      # Two group ACLs but with mask to deny the permission.
      self.hdfs_client.setacl(
          table_path,
          "user::r-x,group::r--,group:{0}:rwx,mask::r-x,"
          "other::---".format(groups[0]))
      self.execute_query_expect_success(self.client, "REFRESH " + table)
      self.execute_query_expect_failure(self.client, insert_query)
  
    def test_clustered_partition_single_file(self, unique_database):
      """IMPALA-2523: Tests that clustered insert creates one file per partition, even when
      inserting over multiple row batches."""
      # On s3 this test takes about 220 seconds and we are unlikely to break it, so only run
      # it in exhaustive strategy.
      if self.exploration_strategy() != 'exhaustive' and IS_S3:
        pytest.skip("only runs in exhaustive")
      table = "{0}.insert_clustered".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_clustered".format(unique_database)
      table_location = get_fs_path("/" + table_path)
  
      create_stmt = """create table {0} like functional.alltypes""".format(table)
      self.execute_query_expect_success(self.client, create_stmt)
  
      set_location_stmt = """alter table {0} set location '{1}'""".format(
          table, table_location)
      self.execute_query_expect_success(self.client, set_location_stmt)
  
      # Setting a lower batch size will result in multiple row batches being written.
      self.execute_query_expect_success(self.client, "set batch_size=10")
  
      insert_stmt = """insert into {0} partition(year, month) /*+ clustered,shuffle */
                       select * from functional.alltypes""".format(table)
      self.execute_query_expect_success(self.client, insert_stmt)
  
      # We expect exactly one partition per year and month, since subsequent row batches of
      # a partition will be written into the same file.
      expected_partitions = \
          ["year=%s/month=%s" % (y, m) for y in [2009, 2010] for m in range(1,13)]
  
      for partition in expected_partitions:
        partition_path = "{0}/{1}".format(table_path, partition)
        files = self.filesystem_client.ls(partition_path)
        assert len(files) == 1, "%s: %s" % (partition, files)
  
    def test_clustered_partition_multiple_files(self, unique_database):
      """IMPALA-2523: Tests that clustered insert creates the right number of files per
      partition when inserting over multiple row batches."""
      # This test takes about 30 seconds and we are unlikely to break it, so only run it in
      # exhaustive strategy.
      if self.exploration_strategy() != 'exhaustive':
        pytest.skip("only runs in exhaustive")
      table = "{0}.insert_clustered".format(unique_database)
      table_path = "test-warehouse/{0}.db/insert_clustered".format(unique_database)
      table_location = get_fs_path("/" + table_path)
  
      create_stmt = """create table {0} (
                       l_orderkey BIGINT,
                       l_partkey BIGINT,
                       l_suppkey BIGINT,
                       l_linenumber INT,
                       l_quantity DECIMAL(12,2),
                       l_extendedprice DECIMAL(12,2),
                       l_discount DECIMAL(12,2),
                       l_tax DECIMAL(12,2),
                       l_linestatus STRING,
                       l_shipdate STRING,
                       l_commitdate STRING,
                       l_receiptdate STRING,
                       l_shipinstruct STRING,
                       l_shipmode STRING,
                       l_comment STRING)
                       partitioned by (l_returnflag STRING) stored as parquet
                       """.format(table)
      self.execute_query_expect_success(self.client, create_stmt)
  
      set_location_stmt = """alter table {0} set location '{1}'""".format(
          table, table_location)
      self.execute_query_expect_success(self.client, set_location_stmt)
  
      # Setting a lower parquet file size will result in multiple files being written.
      self.execute_query_expect_success(self.client, "set parquet_file_size=10485760")
  
      insert_stmt = """insert into {0} partition(l_returnflag) /*+ clustered,shuffle */
          select l_orderkey,
                 l_partkey,
                 l_suppkey,
                 l_linenumber,
                 l_quantity,
                 l_extendedprice,
                 l_discount,
                 l_tax,
                 l_linestatus,
                 l_shipdate,
                 l_commitdate,
                 l_receiptdate,
                 l_shipinstruct,
                 l_shipmode,
                 l_comment,
                 l_returnflag
                 from tpch_parquet.lineitem""".format(table)
      self.execute_query_expect_success(self.client, insert_stmt)
  
      expected_partition_files = [("l_returnflag=A", 3, 30),
                                  ("l_returnflag=N", 3, 30),
                                  ("l_returnflag=R", 3, 30)]
  
      for partition, min_files, max_files in expected_partition_files:
        partition_path = "{0}/{1}".format(table_path, partition)
        files = self.filesystem_client.ls(partition_path)
        assert min_files <= len(files) and len(files) <= max_files, \
            "%s: %s" % (partition, files)
  