diff --git a/be/test/cloud/cloud_tablet_test.cpp b/be/test/cloud/cloud_tablet_test.cpp index a5869d2ca24583..732d939984dd89 100644 --- a/be/test/cloud/cloud_tablet_test.cpp +++ b/be/test/cloud/cloud_tablet_test.cpp @@ -1130,6 +1130,7 @@ class CloudTabletSyncMetaTest : public testing::Test { CloudTabletSyncMetaTest() : _engine(CloudStorageEngine(EngineOptions {})) {} void SetUp() override { + _old_enable_file_cache = config::enable_file_cache; config::enable_file_cache = true; // Use incrementing tablet_id to create unique schema cache keys for each test @@ -1161,7 +1162,7 @@ class CloudTabletSyncMetaTest : public testing::Test { _current_tablet_id = unique_tablet_id; } - void TearDown() override { config::enable_file_cache = false; } + void TearDown() override { config::enable_file_cache = _old_enable_file_cache; } protected: // Helper to create a unique TabletMeta for mock responses @@ -1194,6 +1195,9 @@ class CloudTabletSyncMetaTest : public testing::Test { CloudStorageEngine _engine; int64_t _current_tablet_id; static inline int _test_counter = 0; + +private: + bool _old_enable_file_cache = false; }; // Test sync_meta syncs disable_auto_compaction from false to true diff --git a/be/test/exec/sink/writer/vhive_partition_writer_report_lifecycle_test.cpp b/be/test/exec/sink/writer/vhive_partition_writer_report_lifecycle_test.cpp index 69463007c4c57a..96003b7b03a1a2 100644 --- a/be/test/exec/sink/writer/vhive_partition_writer_report_lifecycle_test.cpp +++ b/be/test/exec/sink/writer/vhive_partition_writer_report_lifecycle_test.cpp @@ -30,6 +30,7 @@ #include "io/fs/s3_file_writer.h" #include "runtime/exec_env.h" #include "testutil/mock/mock_runtime_state.h" +#include "testutil/scoped_file_cache.h" namespace doris { namespace { @@ -174,8 +175,19 @@ ReportStatusRequest report_request(RuntimeState* state, bool done) { } // namespace -TEST(VHivePartitionWriterReportLifecycleTest, - PeriodicReportDefersMetadataAndFinalReportTransfersUploadIdentity) { +class VHivePartitionWriterReportLifecycleTest : public testing::Test { +protected: + void SetUp() override { + auto st = _file_cache.init(); + ASSERT_TRUE(st.ok()) << st; + } + +private: + test::ScopedFileCache _file_cache; +}; + +TEST_F(VHivePartitionWriterReportLifecycleTest, + PeriodicReportDefersMetadataAndFinalReportTransfersUploadIdentity) { MockRuntimeState state; VExprContextSPtrs output_exprs; auto client = std::make_shared(); @@ -201,7 +213,7 @@ TEST(VHivePartitionWriterReportLifecycleTest, EXPECT_EQ("upload-id", pending_upload.upload_id); } -TEST(VHivePartitionWriterReportLifecycleTest, AzureFinalReportCarriesExactBlockIdentity) { +TEST_F(VHivePartitionWriterReportLifecycleTest, AzureFinalReportCarriesExactBlockIdentity) { MockRuntimeState state; VExprContextSPtrs output_exprs; auto client = std::make_shared(); diff --git a/be/test/io/cache/block_file_cache_downloader_test.cpp b/be/test/io/cache/block_file_cache_downloader_test.cpp index c74e9f0e9b9505..84e250bc98b96b 100644 --- a/be/test/io/cache/block_file_cache_downloader_test.cpp +++ b/be/test/io/cache/block_file_cache_downloader_test.cpp @@ -38,6 +38,9 @@ class FileCacheBlockDownloaderTest : public testing::Test { FileCacheBlockDownloaderTest() : _engine(CloudStorageEngine(EngineOptions {})) {} void SetUp() override { + _old_enable_file_cache = config::enable_file_cache; + _old_thread_num_min = config::file_cache_downloader_thread_num_min; + _old_thread_num_max = config::file_cache_downloader_thread_num_max; // Enable file cache for testing config::enable_file_cache = true; // Set reasonable thread pool size for testing @@ -45,7 +48,11 @@ class FileCacheBlockDownloaderTest : public testing::Test { config::file_cache_downloader_thread_num_max = 4; } - void TearDown() override { config::enable_file_cache = false; } + void TearDown() override { + config::enable_file_cache = _old_enable_file_cache; + config::file_cache_downloader_thread_num_min = _old_thread_num_min; + config::file_cache_downloader_thread_num_max = _old_thread_num_max; + } // Helper to wait for inflight tasks to complete with timeout bool wait_for_task_done(FileCacheBlockDownloader& downloader, int64_t tablet_id, @@ -104,6 +111,11 @@ class FileCacheBlockDownloaderTest : public testing::Test { protected: CloudStorageEngine _engine; + +private: + bool _old_enable_file_cache = false; + int32_t _old_thread_num_min = 0; + int32_t _old_thread_num_max = 0; }; // Test that inflight count is correctly decremented when tablet is not found diff --git a/be/test/io/fs/hdfs_file_system_test.cpp b/be/test/io/fs/hdfs_file_system_test.cpp index 859b728fcdd1a9..e5823e8716c49d 100644 --- a/be/test/io/fs/hdfs_file_system_test.cpp +++ b/be/test/io/fs/hdfs_file_system_test.cpp @@ -30,6 +30,8 @@ #include "io/fs/file_writer.h" #include "io/fs/hdfs_file_writer.h" #include "io/fs/local_file_system.h" +#include "testutil/scoped_file_cache.h" +#include "util/defer_op.h" namespace doris { @@ -105,12 +107,18 @@ TEST(HdfsFileSystemTest, Write) { }, &guard4); + test::ScopedFileCache file_cache; + st = file_cache.init(); + ASSERT_TRUE(st.ok()) << st; + + const auto old_batch_size = config::hdfs_write_batch_buffer_size_mb; + Defer restore_batch_size {[&] { config::hdfs_write_batch_buffer_size_mb = old_batch_size; }}; config::hdfs_write_batch_buffer_size_mb = 1; - // TODO(plat1ko): Test write file cache io::FileWriterOptions opts {.write_file_cache = true}; auto hdfs_file_writer = std::make_unique("mock_hdfs_file", nullptr, nullptr, "fs_name", &opts); + ASSERT_NE(hdfs_file_writer->cache_builder(), nullptr); std::string content_1000(1000, 'a'); constexpr size_t MB = 1024 * 1024; diff --git a/be/test/io/fs/packed_file_manager_test.cpp b/be/test/io/fs/packed_file_manager_test.cpp index c481be20c1ac39..de56ac6aa1f2fb 100644 --- a/be/test/io/fs/packed_file_manager_test.cpp +++ b/be/test/io/fs/packed_file_manager_test.cpp @@ -40,8 +40,11 @@ #include "io/fs/file_writer.h" #include "io/fs/packed_file_trailer.h" #include "io/fs/path.h" +#include "runtime/exec_env.h" +#include "testutil/scoped_file_cache.h" #include "util/coding.h" #include "util/slice.h" +#include "util/threadpool.h" namespace doris::io { @@ -225,7 +228,6 @@ class PackedFileManagerTest : public testing::Test { _old_small_file_count_threshold = config::packed_file_small_file_count_threshold; _old_deploy_mode = config::deploy_mode; _old_cloud_id = config::cloud_unique_id; - _old_enable_file_cache = config::enable_file_cache; _old_enable_file_cache_write_from_s3_file_writer = config::enable_file_cache_write_from_s3_file_writer; @@ -238,6 +240,8 @@ class PackedFileManagerTest : public testing::Test { config::cloud_unique_id.clear(); config::enable_file_cache_write_from_s3_file_writer = true; + auto st = _file_cache.init(); + ASSERT_TRUE(st.ok()) << st; file_system = std::make_shared(); manager = std::make_unique(); manager->_file_systems[_resource_id] = file_system; @@ -252,6 +256,12 @@ class PackedFileManagerTest : public testing::Test { manager.reset(); file_system.reset(); + // Standalone tests write the cache synchronously; a preceding fixture may provide a + // pool. Drain its cache writes before restoring configuration or destroying our cache. + if (auto* pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool()) { + pool->wait(); + } + config::packed_file_size_threshold_bytes = _old_merge_threshold; config::small_file_threshold_bytes = _old_small_threshold; config::uploaded_file_retention_seconds = _old_retention; @@ -259,7 +269,6 @@ class PackedFileManagerTest : public testing::Test { config::packed_file_small_file_count_threshold = _old_small_file_count_threshold; config::deploy_mode = _old_deploy_mode; config::cloud_unique_id = _old_cloud_id; - config::enable_file_cache = _old_enable_file_cache; config::enable_file_cache_write_from_s3_file_writer = _old_enable_file_cache_write_from_s3_file_writer; } @@ -281,6 +290,7 @@ class PackedFileManagerTest : public testing::Test { return info; } + test::ScopedFileCache _file_cache; std::shared_ptr file_system; std::unique_ptr manager; @@ -292,7 +302,6 @@ class PackedFileManagerTest : public testing::Test { int64_t _old_small_file_count_threshold = 0; std::string _old_deploy_mode; std::string _old_cloud_id; - bool _old_enable_file_cache = false; bool _old_enable_file_cache_write_from_s3_file_writer = true; std::string _resource_id = "test_resource"; int64_t _tablet_id = 12345; diff --git a/be/test/io/fs/s3_file_writer_test.cpp b/be/test/io/fs/s3_file_writer_test.cpp index 757edf585b0860..41826def55b89b 100644 --- a/be/test/io/fs/s3_file_writer_test.cpp +++ b/be/test/io/fs/s3_file_writer_test.cpp @@ -62,6 +62,7 @@ #include "storage/index/index_file_writer.h" #include "storage/index/index_writer.h" #include "testutil/mock/obj_storage_client_test_stub.h" +#include "testutil/scoped_file_cache.h" #include "util/slice.h" #include "util/threadpool.h" #include "util/uuid_generator.h" @@ -279,9 +280,17 @@ static auto test_mock_callbacks = std::array { class S3FileWriterTest : public testing::Test { public: static void SetUpTestSuite() { + auto* env = ExecEnv::GetInstance(); + _previous_upload_pool = std::move(env->_s3_file_upload_thread_pool); + _previous_disk_resource_limit_percent = + config::file_cache_enter_disk_resource_limit_mode_percent; + config::file_cache_enter_disk_resource_limit_mode_percent = 99; + _file_cache = std::make_unique(); + auto st = _file_cache->init(); + ASSERT_TRUE(st.ok()) << st; + auto sp = SyncPoint::get_instance(); sp->enable_processing(); - config::file_cache_enter_disk_resource_limit_mode_percent = 99; std::for_each(test_mock_callbacks.begin(), test_mock_callbacks.end(), [sp](const MockCallback& mockcallback) { sp->set_call_back(mockcallback.point_name, mockcallback.callback); @@ -300,22 +309,41 @@ class S3FileWriterTest : public testing::Test { s3_fs = res.value(); std::unique_ptr _pool; - std::ignore = ThreadPoolBuilder("s3_upload_file_thread_pool") - .set_min_threads(5) - .set_max_threads(10) - .build(&_pool); - ExecEnv::GetInstance()->_s3_file_upload_thread_pool = std::move(_pool); + st = ThreadPoolBuilder("s3_upload_file_thread_pool") + .set_min_threads(5) + .set_max_threads(10) + .build(&_pool); + ASSERT_TRUE(st.ok()) << st; + env->_s3_file_upload_thread_pool = std::move(_pool); } static void TearDownTestSuite() { + auto* env = ExecEnv::GetInstance(); + // Setup can fail before creating the pool. Otherwise finish queued uploads before + // removing their callbacks or releasing the cache they use. + if (auto* pool = env->s3_file_upload_thread_pool()) { + pool->wait(); + } + env->_s3_file_upload_thread_pool.reset(); auto sp = SyncPoint::get_instance(); std::for_each(test_mock_callbacks.begin(), test_mock_callbacks.end(), [sp](const MockCallback& mockcallback) { sp->clear_call_back(mockcallback.point_name); }); sp->disable_processing(); - ExecEnv::GetInstance()->_s3_file_upload_thread_pool.reset(); + // Release the fixture's clients before Aws::ShutdownAPI at process exit. + s3_fs.reset(); + mock_client.reset(); + _file_cache.reset(); + config::file_cache_enter_disk_resource_limit_mode_percent = + _previous_disk_resource_limit_percent; + env->_s3_file_upload_thread_pool = std::move(_previous_upload_pool); } + +private: + inline static std::unique_ptr _file_cache; + inline static std::unique_ptr _previous_upload_pool; + inline static int32_t _previous_disk_resource_limit_percent = 0; }; TEST_F(S3FileWriterTest, DisableFileCacheWriteFromS3FileWriter) { @@ -1417,14 +1445,13 @@ std::string get_s3_path(std::string_view path) { // put object // create_multi_parts_upload + upload_part + complete_parts TEST_F(S3FileWriterTest, write_buffer_boundary) { - // diable file cache to avoid write to cache bool enable_file_cache = config::enable_file_cache; + int64_t write_buffer_size = config::s3_write_buffer_size; + Defer restore_config {[&]() { + config::enable_file_cache = enable_file_cache; + config::s3_write_buffer_size = write_buffer_size; + }}; config::enable_file_cache = false; - Defer defer {[&]() { config::enable_file_cache = enable_file_cache; }}; - - auto sp = SyncPoint::get_instance(); - sp->enable_processing(); - sp->clear_all_call_backs(); // s3_file_writer is the interface to write to s3 // mock_client is a SimpleMockObjStorageClient for testing, it holds the data in memory diff --git a/be/test/testutil/scoped_file_cache.h b/be/test/testutil/scoped_file_cache.h new file mode 100644 index 00000000000000..a7fe53cde2a115 --- /dev/null +++ b/be/test/testutil/scoped_file_cache.h @@ -0,0 +1,63 @@ +// 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. + +#pragma once + +#include "common/config.h" +#include "common/status.h" +#include "io/cache/block_file_cache_factory.h" +#include "io/cache/file_cache_common.h" +#include "runtime/exec_env.h" + +namespace doris::test { + +// Provide a populated cache even when the test runs alone or follows a fixture that disabled caching. +// Callers must release cache users and finish asynchronous writes before destroying this scope. +class ScopedFileCache { +public: + ScopedFileCache() + : _previous_factory(ExecEnv::GetInstance()->file_cache_factory()), + _previous_enable_file_cache(config::enable_file_cache) {} + + ~ScopedFileCache() { + ExecEnv::GetInstance()->set_file_cache_factory(_previous_factory); + config::enable_file_cache = _previous_enable_file_cache; + } + + ScopedFileCache(const ScopedFileCache&) = delete; + ScopedFileCache& operator=(const ScopedFileCache&) = delete; + + Status init() { + // Standalone runs have no factory, and earlier fixtures can leave an empty one. + // Reuse a populated factory: memory caches share the same path and metric names. + if (_previous_factory == nullptr || _previous_factory->get_cache_instance_size() == 0) { + auto settings = io::get_file_cache_settings(16 * 1024 * 1024, 0); + settings.storage = "memory"; + RETURN_IF_ERROR(_factory.create_file_cache("memory", settings)); + ExecEnv::GetInstance()->set_file_cache_factory(&_factory); + } + config::enable_file_cache = true; + return Status::OK(); + } + +private: + io::FileCacheFactory* _previous_factory; + bool _previous_enable_file_cache; + io::FileCacheFactory _factory; +}; + +} // namespace doris::test diff --git a/be/test/testutil/scoped_file_cache_test.cpp b/be/test/testutil/scoped_file_cache_test.cpp new file mode 100644 index 00000000000000..a91a0418035d7e --- /dev/null +++ b/be/test/testutil/scoped_file_cache_test.cpp @@ -0,0 +1,80 @@ +// 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 "testutil/scoped_file_cache.h" + +#include + +#include "util/defer_op.h" + +namespace doris::test { + +TEST(ScopedFileCacheTest, InitializesWithoutAnExistingCacheAndRestoresState) { + auto* env = ExecEnv::GetInstance(); + auto* original_factory = env->file_cache_factory(); + bool original_enable_file_cache = config::enable_file_cache; + io::FileCacheFactory empty_factory; + Defer restore {[&] { + env->set_file_cache_factory(original_factory); + config::enable_file_cache = original_enable_file_cache; + }}; + + // A standalone run has no factory; another fixture can leave an empty factory instead. + for (auto* previous_factory : {static_cast(nullptr), &empty_factory}) { + for (bool enabled : {false, true}) { + env->set_file_cache_factory(previous_factory); + config::enable_file_cache = enabled; + { + ScopedFileCache cache; + auto st = cache.init(); + ASSERT_TRUE(st.ok()) << st; + EXPECT_TRUE(config::enable_file_cache); + auto* factory = env->file_cache_factory(); + ASSERT_NE(factory, nullptr); + ASSERT_EQ(factory->get_cache_instance_size(), 1); + EXPECT_NE(factory->get_by_path(io::BlockFileCache::hash("test.dat")), nullptr); + } + EXPECT_EQ(env->file_cache_factory(), previous_factory); + EXPECT_EQ(config::enable_file_cache, enabled); + EXPECT_EQ(empty_factory.get_cache_instance_size(), 0); + } + } +} + +TEST(ScopedFileCacheTest, PreservesAnExistingPopulatedCache) { + ScopedFileCache previous_cache; + auto st = previous_cache.init(); + ASSERT_TRUE(st.ok()) << st; + auto* previous_factory = ExecEnv::GetInstance()->file_cache_factory(); + auto key = io::BlockFileCache::hash("test.dat"); + auto* previous_instance = previous_factory->get_by_path(key); + + { + ScopedFileCache cache; + st = cache.init(); + ASSERT_TRUE(st.ok()) << st; + auto* factory = ExecEnv::GetInstance()->file_cache_factory(); + EXPECT_EQ(factory, previous_factory); + EXPECT_EQ(factory->get_by_path(key), previous_instance); + } + + EXPECT_EQ(ExecEnv::GetInstance()->file_cache_factory(), previous_factory); + EXPECT_EQ(previous_factory->get_by_path(key), previous_instance); + EXPECT_TRUE(config::enable_file_cache); +} + +} // namespace doris::test diff --git a/run-be-ut.sh b/run-be-ut.sh index 78e24ef9bdc59e..1571e2006a6c52 100755 --- a/run-be-ut.sh +++ b/run-be-ut.sh @@ -412,6 +412,9 @@ CONF_DIR="${DORIS_TEST_BINARY_DIR}/conf" rm -rf "${CONF_DIR}" mkdir "${CONF_DIR}" cp "${DORIS_HOME}/conf/be.conf" "${CONF_DIR}"/ +# doris_be_test does not initialize ExecEnv's file cache. Fixtures that exercise +# caching install a populated factory and enable it for the duration of the test. +printf '\nenable_file_cache = false\n' >> "${CONF_DIR}/be.conf" export TERM="xterm" export UDF_RUNTIME_DIR="${DORIS_TEST_BINARY_DIR}/lib/udf-runtime"