Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion be/test/cloud/cloud_tablet_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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<RecordingObjStorageClient>();
Expand All @@ -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<RecordingObjStorageClient>();
Expand Down
14 changes: 13 additions & 1 deletion be/test/io/cache/block_file_cache_downloader_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -38,14 +38,21 @@ 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
config::file_cache_downloader_thread_num_min = 2;
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,
Expand Down Expand Up @@ -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
Expand Down
10 changes: 9 additions & 1 deletion be/test/io/fs/hdfs_file_system_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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<io::HdfsFileWriter>("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;
Expand Down
15 changes: 12 additions & 3 deletions be/test/io/fs/packed_file_manager_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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;

Expand All @@ -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<MockFileSystem>();
manager = std::make_unique<PackedFileManager>();
manager->_file_systems[_resource_id] = file_system;
Expand All @@ -252,14 +256,19 @@ 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;
config::packed_file_time_threshold_ms = _old_time_threshold;
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;
}
Expand All @@ -281,6 +290,7 @@ class PackedFileManagerTest : public testing::Test {
return info;
}

test::ScopedFileCache _file_cache;
std::shared_ptr<MockFileSystem> file_system;
std::unique_ptr<PackedFileManager> manager;

Expand All @@ -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;
Expand Down
53 changes: 40 additions & 13 deletions be/test/io/fs/s3_file_writer_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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<test::ScopedFileCache>();
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);
Expand All @@ -300,22 +309,41 @@ class S3FileWriterTest : public testing::Test {
s3_fs = res.value();

std::unique_ptr<ThreadPool> _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<test::ScopedFileCache> _file_cache;
inline static std::unique_ptr<ThreadPool> _previous_upload_pool;
inline static int32_t _previous_disk_resource_limit_percent = 0;
};

TEST_F(S3FileWriterTest, DisableFileCacheWriteFromS3FileWriter) {
Expand Down Expand Up @@ -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
Expand Down
63 changes: 63 additions & 0 deletions be/test/testutil/scoped_file_cache.h
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading