Repository: incubator-impala Updated Branches: refs/heads/master 88f1f86b3 -> eea4ad7ca
IMPALA-5386: Fix ReopenCachedHdfsFileHandle failure case This fixes three issues with the file handle cache. The first issue is that ReopenCachedHdfsFileHandle can destroy the passed in file handle without removing the reference to it. The old file handle then refers to a piece of memory that is not a handle in the cache, so future use of the handle fails with an assert. The fix is to always overwrite the reference to the file handle when it has been destroyed. The second issue is that query_test/test_hdfs_fd_caching.py should run on anything that supports the hdfs commandline and tolerate query failure. It's logic is not specific to file handle caching, so it has been renamed to query_test/test_hdfs_file_mods.py. Finally, custom_cluster/test_hdfs_fd_caching.py should not be running on remote files (S3, ADLS, Isilon, remote clusters). The file handle cache semantics won't apply on those platforms. Change-Id: Iee982fa5e964f6c8969b2eb7e5f3eca89e793b3a Reviewed-on: http://gerrit.cloudera.org:8080/7020 Reviewed-by: Joe McDonnell <[email protected]> Reviewed-by: Dan Hecht <[email protected]> Tested-by: Impala Public Jenkins Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/5f9f704b Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/5f9f704b Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/5f9f704b Branch: refs/heads/master Commit: 5f9f704bde6816c8469e31b1d655f3df5732dac4 Parents: 88f1f86 Author: Joe McDonnell <[email protected]> Authored: Tue May 30 12:59:37 2017 -0700 Committer: Impala Public Jenkins <[email protected]> Committed: Fri Jun 9 01:45:37 2017 +0000 ---------------------------------------------------------------------- .../runtime/disk-io-mgr-handle-cache.inline.h | 1 + be/src/runtime/disk-io-mgr.cc | 8 +- tests/common/skip.py | 3 + tests/custom_cluster/test_hdfs_fd_caching.py | 5 +- tests/query_test/test_hdfs_fd_caching.py | 124 ------------------ tests/query_test/test_hdfs_file_mods.py | 129 +++++++++++++++++++ 6 files changed, 140 insertions(+), 130 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/be/src/runtime/disk-io-mgr-handle-cache.inline.h ---------------------------------------------------------------------- diff --git a/be/src/runtime/disk-io-mgr-handle-cache.inline.h b/be/src/runtime/disk-io-mgr-handle-cache.inline.h index 8d3a9e0..ed8ed63 100644 --- a/be/src/runtime/disk-io-mgr-handle-cache.inline.h +++ b/be/src/runtime/disk-io-mgr-handle-cache.inline.h @@ -106,6 +106,7 @@ HdfsFileHandle* FileHandleCache<NUM_PARTITIONS>::GetFileHandle( template <size_t NUM_PARTITIONS> void FileHandleCache<NUM_PARTITIONS>::ReleaseFileHandle(std::string* fname, HdfsFileHandle* fh, bool destroy_handle) { + DCHECK(fh != nullptr); // Hash the key and get appropriate partition int index = HashUtil::Hash(fname->data(), fname->size(), 0) % NUM_PARTITIONS; FileHandleCachePartition& p = cache_partitions_[index]; http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/be/src/runtime/disk-io-mgr.cc ---------------------------------------------------------------------- diff --git a/be/src/runtime/disk-io-mgr.cc b/be/src/runtime/disk-io-mgr.cc index 01dea0e..89a8863 100644 --- a/be/src/runtime/disk-io-mgr.cc +++ b/be/src/runtime/disk-io-mgr.cc @@ -1279,9 +1279,11 @@ Status DiskIoMgr::ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fnam int64_t mtime, HdfsFileHandle** fid) { bool dummy; file_handle_cache_.ReleaseFileHandle(fname, *fid, true); - HdfsFileHandle* fh = file_handle_cache_.GetFileHandle(fs, fname, mtime, true, + // The old handle has been destroyed, so *fid must be overwritten before returning. + *fid = file_handle_cache_.GetFileHandle(fs, fname, mtime, true, &dummy); - if (!fh) return Status(GetHdfsErrorMsg("Failed to open HDFS file ", fname->data())); - *fid = fh; + if (*fid == nullptr) { + return Status(GetHdfsErrorMsg("Failed to open HDFS file ", fname->data())); + } return Status::OK(); } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/tests/common/skip.py ---------------------------------------------------------------------- diff --git a/tests/common/skip.py b/tests/common/skip.py index 642dea4..77ccc1e 100644 --- a/tests/common/skip.py +++ b/tests/common/skip.py @@ -85,6 +85,9 @@ class SkipIf: not_hdfs = pytest.mark.skipif(not IS_HDFS, reason="HDFS Filesystem needed") no_secondary_fs = pytest.mark.skipif(not SECONDARY_FILESYSTEM, reason="Secondary filesystem needed") + no_file_handle_caching = pytest.mark.skipif(IS_S3 or IS_ADLS or IS_ISILON or \ + pytest.config.option.testing_remote_cluster, + reason="File handle caching needed") class SkipIfIsilon: caching = pytest.mark.skipif(IS_ISILON, reason="SET CACHED not implemented for Isilon") http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/tests/custom_cluster/test_hdfs_fd_caching.py ---------------------------------------------------------------------- diff --git a/tests/custom_cluster/test_hdfs_fd_caching.py b/tests/custom_cluster/test_hdfs_fd_caching.py index bcf50e6..b1e8f30 100644 --- a/tests/custom_cluster/test_hdfs_fd_caching.py +++ b/tests/custom_cluster/test_hdfs_fd_caching.py @@ -18,10 +18,9 @@ import pytest from tests.common.custom_cluster_test_suite import CustomClusterTestSuite -from tests.common.skip import SkipIfS3, SkipIfADLS +from tests.common.skip import SkipIf [email protected] [email protected] [email protected]_file_handle_caching class TestHdfsFdCaching(CustomClusterTestSuite): """Tests that if HDFS file handle caching is enabled, file handles are actually cached and the associated metrics return valid results. In addition, tests that the upper bound http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/tests/query_test/test_hdfs_fd_caching.py ---------------------------------------------------------------------- diff --git a/tests/query_test/test_hdfs_fd_caching.py b/tests/query_test/test_hdfs_fd_caching.py deleted file mode 100644 index fa03b24..0000000 --- a/tests/query_test/test_hdfs_fd_caching.py +++ /dev/null @@ -1,124 +0,0 @@ -# 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 pytest - -from tests.common.impala_cluster import ImpalaCluster -from tests.common.impala_test_suite import ImpalaTestSuite -from tests.common.skip import SkipIfS3, SkipIfADLS -from tests.common.test_vector import ImpalaTestDimension -from subprocess import call -from tests.util.filesystem_utils import FILESYSTEM_PREFIX - -# Modifications to test with the file handle cache -MODIFICATION_TYPES=["delete_files", "delete_directory", "move_file", "append"] - [email protected] [email protected] -class TestHdfsFdCaching(ImpalaTestSuite): - """ - This test suite tests the behavior of HDFS file descriptor caching, including - potential error cases such as file deletes. - """ - - @classmethod - def file_format_constraint(cls, v): - return v.get_value('table_format').file_format in ["text"] - - @classmethod - def add_test_dimensions(cls): - super(TestHdfsFdCaching, cls).add_test_dimensions() - cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('modification_type',\ - *MODIFICATION_TYPES)) - cls.ImpalaTestMatrix.add_constraint(cls.file_format_constraint) - - @classmethod - def get_workload(cls): - return 'functional-query' - - def setup_ext_table(self, vector, unique_database, new_table_location): - # Use HDFS commands to clone the table's files at the hdfs level - old_table_location = "{0}/test-warehouse/tinytable".format(FILESYSTEM_PREFIX) - call(["hdfs", "dfs", "-mkdir", new_table_location]) - call(["hdfs", "dfs", "-cp", old_table_location + "/*", new_table_location]) - - # Create an external table with the new files (tinytable has two string columns) - create_table = "create external table {0}.t1 (a string, b string) "\ - + "row format delimited fields terminated by \',\' location \'{1}\'" - self.client.execute(create_table.format(unique_database, new_table_location)) - - @pytest.mark.execute_serially - def test_file_modifications(self, vector, unique_database): - """Tests file modifications on a file that is cached in the file handle cache.""" - - new_table_location = "{0}/test-warehouse/{1}".format(FILESYSTEM_PREFIX,\ - unique_database) - self.setup_ext_table(vector, unique_database, new_table_location) - - # Query the table (puts file handle in the cache) - count_query = "select count(*) from {0}.t1".format(unique_database) - original_result = self.execute_query_expect_success(self.client, count_query) - assert(original_result.data[0] == '3') - - # Do the modification based on the test settings - modification_type = vector.get_value('modification_type') - if (modification_type == 'delete_files'): - # Delete the data file (not the directory) - call(["hdfs", "dfs", "-rm", "-skipTrash", new_table_location + "/*"]) - elif (modification_type == 'delete_directory'): - # Delete the whole directory (including data file) - call(["hdfs", "dfs", "-rm", "-r", "-skipTrash", new_table_location]) - elif (modification_type == 'move_file'): - # Move the file underneath the directory - call(["hdfs", "dfs", "-mv", new_table_location + "/data.csv", \ - new_table_location + "/data.csv.moved"]) - elif (modification_type == 'append'): - # Append a copy of the hdfs file to itself (duplicating all entries) - local_tmp_location = "/tmp/{0}.data.csv".format(unique_database) - call(["hdfs", "dfs", "-copyToLocal", new_table_location + "/data.csv", \ - local_tmp_location]) - call(["hdfs", "dfs", "-appendToFile", local_tmp_location, \ - new_table_location + "/data.csv"]) - call(["rm", local_tmp_location]) - else: - assert(false) - - # The query might fail, but nothing should crash. - self.execute_query(count_query) - - # Invalidate metadata - invalidate_metadata_sql = "invalidate metadata {0}.t1".format(unique_database) - self.execute_query_expect_success(self.client, invalidate_metadata_sql) - - # Verify that nothing crashes and the query should succeed - new_result = self.execute_query_expect_success(self.client, count_query) - if (modification_type == 'move_file'): - assert(new_result.data[0] == '3') - elif (modification_type == 'delete_files' or \ - modification_type == 'delete_directory'): - assert(new_result.data[0] == '0') - elif (modification_type == 'append'): - assert(new_result.data[0] == '6') - - # Drop table - drop_table_sql = "drop table {0}.t1".format(unique_database) - self.execute_query_expect_success(self.client, drop_table_sql) - - # Cleanup directory (which may already be gone) - call(["hdfs", "dfs", "-rm", "-r", "-skipTrash", new_table_location]) - http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5f9f704b/tests/query_test/test_hdfs_file_mods.py ---------------------------------------------------------------------- diff --git a/tests/query_test/test_hdfs_file_mods.py b/tests/query_test/test_hdfs_file_mods.py new file mode 100644 index 0000000..30698e1 --- /dev/null +++ b/tests/query_test/test_hdfs_file_mods.py @@ -0,0 +1,129 @@ +# 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 pytest + +from tests.beeswax.impala_beeswax import ImpalaBeeswaxException +from tests.common.impala_cluster import ImpalaCluster +from tests.common.impala_test_suite import ImpalaTestSuite +from tests.common.skip import SkipIfLocal +from tests.common.test_vector import ImpalaTestDimension +from subprocess import call +from tests.util.filesystem_utils import FILESYSTEM_PREFIX + +# Modifications to test +MODIFICATION_TYPES=["delete_files", "delete_directory", "move_file", "append"] + [email protected]_client +class TestHdfsFileMods(ImpalaTestSuite): + """ + This test suite tests that modifications to HDFS files don't crash Impala. + In particular, this interacts with existing functionality such as the + file handle cache and IO retries. + """ + + @classmethod + def file_format_constraint(cls, v): + return v.get_value('table_format').file_format in ["text"] + + @classmethod + def add_test_dimensions(cls): + super(TestHdfsFileMods, cls).add_test_dimensions() + cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('modification_type',\ + *MODIFICATION_TYPES)) + cls.ImpalaTestMatrix.add_constraint(cls.file_format_constraint) + + @classmethod + def get_workload(cls): + return 'functional-query' + + def setup_ext_table(self, vector, unique_database, new_table_location): + # Use HDFS commands to clone the table's files at the hdfs level + old_table_location = "{0}/test-warehouse/tinytable".format(FILESYSTEM_PREFIX) + call(["hdfs", "dfs", "-mkdir", new_table_location]) + call(["hdfs", "dfs", "-cp", old_table_location + "/*", new_table_location]) + + # Create an external table with the new files (tinytable has two string columns) + create_table = "create external table {0}.t1 (a string, b string) "\ + + "row format delimited fields terminated by \',\' location \'{1}\'" + self.client.execute(create_table.format(unique_database, new_table_location)) + + @pytest.mark.execute_serially + def test_file_modifications(self, vector, unique_database): + """Tests file modifications on an external table.""" + + new_table_location = "{0}/test-warehouse/{1}".format(FILESYSTEM_PREFIX,\ + unique_database) + self.setup_ext_table(vector, unique_database, new_table_location) + + # Query the table. If file handle caching is enabled, this will fill the cache. + count_query = "select count(*) from {0}.t1".format(unique_database) + original_result = self.execute_query_expect_success(self.client, count_query) + assert(original_result.data[0] == '3') + + # Do the modification based on the test settings + modification_type = vector.get_value('modification_type') + if (modification_type == 'delete_files'): + # Delete the data file (not the directory) + call(["hdfs", "dfs", "-rm", "-skipTrash", new_table_location + "/*"]) + elif (modification_type == 'delete_directory'): + # Delete the whole directory (including data file) + call(["hdfs", "dfs", "-rm", "-r", "-skipTrash", new_table_location]) + elif (modification_type == 'move_file'): + # Move the file underneath the directory + call(["hdfs", "dfs", "-mv", new_table_location + "/data.csv", \ + new_table_location + "/data.csv.moved"]) + elif (modification_type == 'append'): + # Append a copy of the hdfs file to itself (duplicating all entries) + local_tmp_location = "/tmp/{0}.data.csv".format(unique_database) + call(["hdfs", "dfs", "-copyToLocal", new_table_location + "/data.csv", \ + local_tmp_location]) + call(["hdfs", "dfs", "-appendToFile", local_tmp_location, \ + new_table_location + "/data.csv"]) + call(["rm", local_tmp_location]) + else: + assert(false) + + # The query might fail, but nothing should crash. + try: + self.execute_query(count_query) + except ImpalaBeeswaxException as e: + pass + + # Invalidate metadata + invalidate_metadata_sql = "invalidate metadata {0}.t1".format(unique_database) + self.execute_query_expect_success(self.client, invalidate_metadata_sql) + + # Verify that nothing crashes and the query should succeed + new_result = self.execute_query_expect_success(self.client, count_query) + if (modification_type == 'move_file'): + assert(new_result.data[0] == '3') + elif (modification_type == 'delete_files' or \ + modification_type == 'delete_directory'): + assert(new_result.data[0] == '0') + elif (modification_type == 'append'): + # Allow either the old count or the new count to tolerate delayed consistency. + assert(new_result.data[0] == '6' or new_result.data[0] == '3') + + # Drop table + drop_table_sql = "drop table {0}.t1".format(unique_database) + self.execute_query_expect_success(self.client, drop_table_sql) + + # Cleanup directory (which may already be gone) + call(["hdfs", "dfs", "-rm", "-r", "-skipTrash", new_table_location]) +
