This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 3b260b82f5d branch-4.1: [fix](cloud) bind the packed slice location
lifetime to its writer #67347 (#67973)
3b260b82f5d is described below
commit 3b260b82f5d8659f70ef25a905697ae2e6d7c9f0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 15 11:59:48 2026 +0800
branch-4.1: [fix](cloud) bind the packed slice location lifetime to its
writer #67347 (#67973)
Cherry-picked from #67347
Co-authored-by: Xin Liao <[email protected]>
---
be/src/cloud/cloud_rowset_writer.cpp | 5 +-
be/src/io/fs/packed_file_manager.cpp | 101 +++++----
be/src/io/fs/packed_file_manager.h | 77 ++++++-
be/src/io/fs/packed_file_writer.cpp | 20 +-
be/src/io/fs/packed_file_writer.h | 7 +-
be/test/io/fs/packed_file_manager_test.cpp | 328 +++++++++++++++++++++++------
6 files changed, 415 insertions(+), 123 deletions(-)
diff --git a/be/src/cloud/cloud_rowset_writer.cpp
b/be/src/cloud/cloud_rowset_writer.cpp
index 47cc3865c1b..62f3a74657b 100644
--- a/be/src/cloud/cloud_rowset_writer.cpp
+++ b/be/src/cloud/cloud_rowset_writer.cpp
@@ -208,10 +208,11 @@ Status
CloudRowsetWriter::_collect_packed_slice_location(io::FileWriter* file_wr
return Status::OK();
}
- // Get packed slice location directly from PackedFileManager
+ // Ask the writer, which holds a reference to its own slice location.
Looking it up by
+ // path in PackedFileManager would race with the retention based cleanup
of the index.
io::PackedSliceLocation index;
RETURN_IF_ERROR(
-
io::PackedFileManager::instance()->get_packed_slice_location(file_path,
&index));
+
static_cast<io::PackedFileWriter*>(file_writer)->get_packed_slice_location(&index));
if (index.packed_file_path.empty()) {
return Status::OK(); // File not in packed file, skip
}
diff --git a/be/src/io/fs/packed_file_manager.cpp
b/be/src/io/fs/packed_file_manager.cpp
index 76f0fb7b359..ccbc64f975f 100644
--- a/be/src/io/fs/packed_file_manager.cpp
+++ b/be/src/io/fs/packed_file_manager.cpp
@@ -337,7 +337,9 @@ Status PackedFileManager::ensure_file_system(const
std::string& resource_id,
}
Status PackedFileManager::append_small_file(const std::string& path, const
Slice& data,
- const PackedAppendContext& info) {
+ const PackedAppendContext& info,
+ PackedSliceHandlePtr* handle) {
+ *handle = nullptr;
// Check if file is too large to be merged
if (data.get_size() > config::small_file_threshold_bytes) {
return Status::OK(); // Skip merging for large files
@@ -387,7 +389,9 @@ Status PackedFileManager::append_small_file(const
std::string& path, const Slice
location.resource_id = info.resource_id;
location.txn_id = info.txn_id;
- active_state->slice_locations[path] = location;
+ auto slice_handle =
std::make_shared<PackedSliceHandle>(std::move(location));
+ active_state->slice_locations[path] = slice_handle;
+ active_state->appended_slices.push_back(slice_handle);
active_state->current_offset += data.get_size();
active_state->total_size += data.get_size();
@@ -409,9 +413,10 @@ Status PackedFileManager::append_small_file(const
std::string& path, const Slice
// Update global index
{
std::lock_guard<std::mutex> global_lock(_global_index_mutex);
- _global_slice_locations[path] = location;
+ _global_slice_locations[path] = slice_handle;
}
+ *handle = std::move(slice_handle);
return Status::OK();
}
@@ -431,17 +436,20 @@ Status
PackedFileManager::wait_for_packed_file_upload(PackedFileContext* packed_
return Status::OK();
}
-Status PackedFileManager::wait_upload_done(const std::string& path) {
- std::string packed_file_path;
- {
- std::lock_guard<std::mutex> global_lock(_global_index_mutex);
- auto it = _global_slice_locations.find(path);
- if (it == _global_slice_locations.end()) {
- return Status::InternalError("File not found in global index: " +
path);
- }
- packed_file_path = it->second.packed_file_path;
+Status PackedFileManager::wait_upload_done(const PackedSliceHandlePtr& handle)
{
+ if (handle == nullptr) {
+ return Status::InternalError("Missing packed slice handle");
}
+ // The upload already finished, so there is no need to look up the
PackedFileContext,
+ // which is only kept for a limited time after the upload.
+ auto upload_state = handle->upload_state();
+ if (upload_state == PackedSliceUploadState::UPLOADED) {
+ return Status::OK();
+ }
+ const bool upload_failed = upload_state == PackedSliceUploadState::FAILED;
+ const std::string& packed_file_path = handle->packed_file_path();
+
// Find the packed file in uploaded files first - if already uploaded, no
need to wait
std::shared_ptr<PackedFileContext> managed_packed_file;
std::shared_ptr<PackedFileContext> failed_packed_file;
@@ -498,8 +506,12 @@ Status PackedFileManager::wait_upload_done(const
std::string& path) {
}
if (!packed_file_ptr) {
+ if (upload_failed) {
+ // The context carrying the original error has already been
recycled
+ return Status::InternalError("Packed file upload failed: " +
packed_file_path);
+ }
// Packed file not found in any location, this is unexpected
- return Status::InternalError("Packed file not found for path: " +
path);
+ return Status::InternalError("Packed file not found: " +
packed_file_path);
}
Status wait_status = wait_for_packed_file_upload(packed_file_ptr);
@@ -515,10 +527,19 @@ Status PackedFileManager::get_packed_slice_location(const
std::string& path,
return Status::NotFound("File not found in global packed index: {}",
path);
}
- *location = it->second;
+ *location = it->second->location();
return Status::OK();
}
+void PackedFileManager::mark_slices_upload_result(const PackedFileContext&
packed_file,
+ PackedSliceUploadState
state) {
+ // Not slice_locations: a path written twice into this packed file is only
in there once,
+ // and the writer of the shadowed slice is waiting for the result too
+ for (const auto& handle : packed_file.appended_slices) {
+ handle->set_upload_result(state, packed_file.total_size);
+ }
+}
+
void PackedFileManager::start_background_manager() {
if (_background_thread) {
return; // Already started
@@ -733,18 +754,11 @@ void PackedFileManager::process_uploading_packed_files() {
slices_stream << "; ";
}
first_slice = false;
- slices_stream << small_file_path << "(txn=" << index.txn_id
- << ", offset=" << index.offset << ", size=" <<
index.size << ")";
-
- // Update packed_file_size in global index
- {
- std::lock_guard<std::mutex> global_lock(_global_index_mutex);
- auto it = _global_slice_locations.find(small_file_path);
- if (it != _global_slice_locations.end()) {
- it->second.packed_file_size = packed_file->total_size;
- }
- }
+ auto location = index->location();
+ slices_stream << small_file_path << "(txn=" << location.txn_id
+ << ", offset=" << location.offset << ", size=" <<
location.size << ")";
}
+ mark_slices_upload_result(*packed_file,
PackedSliceUploadState::UPLOADED);
LOG(INFO) << "Packed file " << packed_file->packed_file_path
<< " uploaded; slices=" <<
packed_file->slice_locations.size()
<< ", total_bytes=" << packed_file->total_size << ",
slice_detail=["
@@ -770,6 +784,7 @@ void PackedFileManager::process_uploading_packed_files() {
const Status& status) {
LOG(WARNING) << "Failed to upload packed file: " <<
packed_file->packed_file_path
<< ", error: " << status.to_string();
+ mark_slices_upload_result(*packed_file,
PackedSliceUploadState::FAILED);
{
std::lock_guard<std::mutex> upload_lock(packed_file->upload_mutex);
packed_file->state = PackedFileState::FAILED;
@@ -797,19 +812,20 @@ void PackedFileManager::process_uploading_packed_files() {
packed_file_info.set_resource_id(packed_file->resource_id);
for (const auto& [small_file_path, index] :
packed_file->slice_locations) {
+ auto location = index->location();
auto* small_file = packed_file_info.add_slices();
small_file->set_path(small_file_path);
- small_file->set_offset(index.offset);
- small_file->set_size(index.size);
+ small_file->set_offset(location.offset);
+ small_file->set_size(location.size);
small_file->set_deleted(false);
- if (index.tablet_id != 0) {
- small_file->set_tablet_id(index.tablet_id);
+ if (location.tablet_id != 0) {
+ small_file->set_tablet_id(location.tablet_id);
}
- if (!index.rowset_id.empty()) {
- small_file->set_rowset_id(index.rowset_id);
+ if (!location.rowset_id.empty()) {
+ small_file->set_rowset_id(location.rowset_id);
}
- if (index.txn_id != 0) {
- small_file->set_txn_id(index.txn_id);
+ if (location.txn_id != 0) {
+ small_file->set_txn_id(location.txn_id);
}
}
@@ -842,8 +858,9 @@ void PackedFileManager::process_uploading_packed_files() {
oss << ", ";
}
first = false;
- oss << "[" << small_file_path << ", offset=" << index.offset
- << ", size=" << index.size << "]";
+ auto location = index->location();
+ oss << "[" << small_file_path << ", offset=" << location.offset
+ << ", size=" << location.size << "]";
}
VLOG_DEBUG << oss.str();
} else {
@@ -940,9 +957,17 @@ void PackedFileManager::cleanup_expired_data() {
std::lock_guard<std::mutex> global_lock(_global_index_mutex);
auto it = _global_slice_locations.begin();
while (it != _global_slice_locations.end()) {
- const auto& index = it->second;
- if (index.create_time > 0 &&
- current_time - index.create_time >
config::uploaded_file_retention_seconds) {
+ // A writer or a packed file context still holds this slice, so
whoever holds it
+ // may still read the entry back. Age says nothing about that: a
load can run for
+ // much longer than the retention time. Only entries the index
alone holds are
+ // stale, and the reference is dropped for us when the last owner
goes away.
+ if (it->second.use_count() > 1) {
+ ++it;
+ continue;
+ }
+ const auto create_time = it->second->create_time();
+ if (create_time > 0 &&
+ current_time - create_time >
config::uploaded_file_retention_seconds) {
it = _global_slice_locations.erase(it);
} else {
++it;
diff --git a/be/src/io/fs/packed_file_manager.h
b/be/src/io/fs/packed_file_manager.h
index 13b5a39bfc0..ab79d5bc7e4 100644
--- a/be/src/io/fs/packed_file_manager.h
+++ b/be/src/io/fs/packed_file_manager.h
@@ -53,6 +53,53 @@ struct PackedSliceLocation {
int64_t packed_file_size = -1; // Total size of the packed file, -1 means
not set
};
+// Upload state of the packed file a slice belongs to
+enum class PackedSliceUploadState : uint8_t {
+ PENDING = 0,
+ UPLOADED,
+ FAILED,
+};
+
+// A slice of a packed file, shared by PackedFileManager and the
PackedFileWriter that
+// produced it. The writer holds its handle for as long as it lives, so it can
wait for the
+// upload and read the location back at the end of the load, whatever the
manager recycled
+// from its by-path index in the meantime.
+class PackedSliceHandle {
+public:
+ explicit PackedSliceHandle(PackedSliceLocation location) :
_location(std::move(location)) {}
+
+ const std::string& packed_file_path() const { return
_location.packed_file_path; }
+
+ int64_t create_time() const { return _location.create_time; }
+
+ PackedSliceUploadState upload_state() const {
+ return _upload_state.load(std::memory_order_acquire);
+ }
+
+ // Called once the packed file this slice belongs to reaches a terminal
state
+ void set_upload_result(PackedSliceUploadState state, int64_t
packed_file_size) {
+ if (state == PackedSliceUploadState::UPLOADED) {
+ _packed_file_size.store(packed_file_size,
std::memory_order_relaxed);
+ }
+ _upload_state.store(state, std::memory_order_release);
+ }
+
+ PackedSliceLocation location() const {
+ PackedSliceLocation location = _location;
+ if (upload_state() == PackedSliceUploadState::UPLOADED) {
+ location.packed_file_size =
_packed_file_size.load(std::memory_order_relaxed);
+ }
+ return location;
+ }
+
+private:
+ const PackedSliceLocation _location; // Immutable once the slice has been
appended
+ std::atomic<int64_t> _packed_file_size {-1};
+ std::atomic<PackedSliceUploadState> _upload_state
{PackedSliceUploadState::PENDING};
+};
+
+using PackedSliceHandlePtr = std::shared_ptr<PackedSliceHandle>;
+
struct PackedAppendContext {
std::string resource_id;
int64_t tablet_id = 0;
@@ -72,14 +119,17 @@ public:
// Initialize manager state; file system will be resolved lazily
Status init();
- // Write a small file to the current packed file
+ // Write a small file to the current packed file. On success `handle`
receives a handle
+ // to the new slice, or nullptr if `data` was too large to be packed.
Status append_small_file(const std::string& path, const Slice& data,
- const PackedAppendContext& info);
+ const PackedAppendContext& info,
PackedSliceHandlePtr* handle);
- // Block until the small file's packed file is uploaded to S3
- Status wait_upload_done(const std::string& path);
+ // Block until the packed file holding `handle` is uploaded to S3
+ Status wait_upload_done(const PackedSliceHandlePtr& handle);
- // Get packed file index information for a small file
+ // Look a slice location up by small file path, for readers that have no
handle to the
+ // slice. The entry lives as long as anything else holds the slice, so
this only fails
+ // for a file whose writer and packed file context are both long gone.
Status get_packed_slice_location(const std::string& path,
PackedSliceLocation* location);
// Start the background management thread
@@ -120,6 +170,10 @@ private:
// Clean up expired data
void cleanup_expired_data();
+ // Record the terminal upload state of `packed_file` on the slices it
contains
+ void mark_slices_upload_result(const PackedFileContext& packed_file,
+ PackedSliceUploadState state);
+
// Internal structure to track packed file state
enum class PackedFileState {
INIT, // Initial state, no files written yet
@@ -133,7 +187,11 @@ private:
struct PackedFileContext {
std::string packed_file_path;
std::unique_ptr<FileWriter> writer;
- std::unordered_map<std::string, PackedSliceLocation> slice_locations;
+ std::unordered_map<std::string, PackedSliceHandlePtr> slice_locations;
+ // Every slice appended to this packed file. `slice_locations` is
keyed by path, so
+ // writing one path twice into the same packed file only leaves the
last handle
+ // there, while the upload result still has to reach both.
+ std::vector<PackedSliceHandlePtr> appended_slices;
int64_t current_offset = 0;
int64_t total_size = 0;
int64_t create_time;
@@ -179,8 +237,11 @@ private:
std::unordered_map<std::string, std::shared_ptr<PackedFileContext>>
_uploaded_packed_files;
std::mutex _packed_files_mutex;
- // Global index mapping small file path to packed file index
- std::unordered_map<std::string, PackedSliceLocation>
_global_slice_locations;
+ // Global index mapping small file path to packed file index, for readers
that have no
+ // handle to the slice, such as PackedFileSystem::open_file_impl() reading
a segment back
+ // before its rowset meta exists. An entry is only recycled once nothing
else holds the
+ // slice, so it outlives every writer and packed file context that could
still read it.
+ std::unordered_map<std::string, PackedSliceHandlePtr>
_global_slice_locations;
std::mutex _global_index_mutex;
#ifdef BE_TEST
diff --git a/be/src/io/fs/packed_file_writer.cpp
b/be/src/io/fs/packed_file_writer.cpp
index 45f49e0e03f..7d97f0acfcd 100644
--- a/be/src/io/fs/packed_file_writer.cpp
+++ b/be/src/io/fs/packed_file_writer.cpp
@@ -176,8 +176,8 @@ Status PackedFileWriter::_close_sync() {
Status PackedFileWriter::_wait_packed_upload() {
DCHECK(!_is_direct_write);
// Only wait if we have data that was sent to packed manager
- if (_bytes_appended > 0 && _packed_file_manager != nullptr) {
- return _packed_file_manager->wait_upload_done(_file_path);
+ if (_bytes_appended > 0 && _packed_file_manager != nullptr &&
_packed_slice_handle != nullptr) {
+ return _packed_file_manager->wait_upload_done(_packed_slice_handle);
}
return Status::OK();
}
@@ -217,7 +217,17 @@ Status PackedFileWriter::_send_to_packed_manager() {
}
Slice data_slice(_buffer.data(), _buffer.size());
- RETURN_IF_ERROR(_packed_file_manager->append_small_file(_file_path,
data_slice, _append_info));
+ RETURN_IF_ERROR(_packed_file_manager->append_small_file(_file_path,
data_slice, _append_info,
+
&_packed_slice_handle));
+ if (_packed_slice_handle == nullptr) {
+ // append_small_file() skips data larger than
`small_file_threshold_bytes` without
+ // writing it anywhere. appendv() switches to direct write before the
buffer can
+ // reach that size, unless the threshold was lowered in between.
Report the failure
+ // instead of closing successfully on a file that does not exist.
+ return Status::InternalError(
+ "Packed file did not accept the buffered data for {}, size={},
threshold={}",
+ _file_path, data_slice.get_size(),
config::small_file_threshold_bytes);
+ }
_release_buffer();
return Status::OK();
}
@@ -225,11 +235,11 @@ Status PackedFileWriter::_send_to_packed_manager() {
Status PackedFileWriter::get_packed_slice_location(PackedSliceLocation*
location) const {
DCHECK(_state == State::CLOSED)
<< " file_path: " << _file_path << " bytes_appended: " <<
_bytes_appended;
- if (_is_direct_write) {
+ if (_is_direct_write || _packed_slice_handle == nullptr) {
*location = PackedSliceLocation {};
return Status::OK();
}
-
RETURN_IF_ERROR(_packed_file_manager->get_packed_slice_location(_file_path,
location));
+ *location = _packed_slice_handle->location();
LOG(INFO) << "get_packed_slice_location: " << _file_path
<< " packed_path: " << location->packed_file_path << " " <<
location->offset << " "
<< location->size;
diff --git a/be/src/io/fs/packed_file_writer.h
b/be/src/io/fs/packed_file_writer.h
index d2c06621138..a53d412e667 100644
--- a/be/src/io/fs/packed_file_writer.h
+++ b/be/src/io/fs/packed_file_writer.h
@@ -89,9 +89,10 @@ private:
std::string _buffer;
size_t _bytes_appended = 0;
State _state = State::OPENED;
- bool _is_direct_write = false; // Whether to use
direct write mode
- PackedFileManager* _packed_file_manager = nullptr; // Packed file manager
instance
- mutable PackedSliceLocation _packed_slice_location; // Packed slice info
(mutable for lazy init)
+ bool _is_direct_write = false; // Whether to use
direct write mode
+ PackedFileManager* _packed_file_manager = nullptr; // Packed file manager
instance
+ // Handle to this file's slice, shared with PackedFileManager
+ PackedSliceHandlePtr _packed_slice_handle;
PackedAppendContext _append_info;
std::optional<std::chrono::steady_clock::time_point>
_first_append_timestamp;
bool _close_latency_recorded = false;
diff --git a/be/test/io/fs/packed_file_manager_test.cpp
b/be/test/io/fs/packed_file_manager_test.cpp
index 93548f8c955..c481be20c1a 100644
--- a/be/test/io/fs/packed_file_manager_test.cpp
+++ b/be/test/io/fs/packed_file_manager_test.cpp
@@ -264,6 +264,14 @@ protected:
_old_enable_file_cache_write_from_s3_file_writer;
}
+ // Most tests do not care about the handle they get back
+ Status append_small_file(const std::string& path, const Slice& data,
+ const PackedAppendContext& info,
+ PackedSliceHandlePtr* handle = nullptr) {
+ PackedSliceHandlePtr unused;
+ return manager->append_small_file(path, data, info, handle != nullptr
? handle : &unused);
+ }
+
PackedAppendContext default_append_info() const {
PackedAppendContext info;
info.resource_id = _resource_id;
@@ -308,19 +316,19 @@ TEST_F(PackedFileManagerTest, AppendSmallFileSuccess) {
Slice slice(payload);
auto info = default_append_info();
- Status st = manager->append_small_file("s/path", slice, info);
+ Status st = append_small_file("s/path", slice, info);
EXPECT_TRUE(st.ok());
EXPECT_EQ(writer->append_calls(), 1);
EXPECT_EQ(writer->bytes_appended(), payload.size());
auto it = manager->global_slice_locations_for_test().find("s/path");
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, payload.size());
- EXPECT_EQ(it->second.offset, 0);
- EXPECT_EQ(it->second.tablet_id, info.tablet_id);
- EXPECT_EQ(it->second.rowset_id, info.rowset_id);
- EXPECT_EQ(it->second.resource_id, info.resource_id);
- EXPECT_EQ(it->second.txn_id, info.txn_id);
+ EXPECT_EQ(it->second->location().size, payload.size());
+ EXPECT_EQ(it->second->location().offset, 0);
+ EXPECT_EQ(it->second->location().tablet_id, info.tablet_id);
+ EXPECT_EQ(it->second->location().rowset_id, info.rowset_id);
+ EXPECT_EQ(it->second->location().resource_id, info.resource_id);
+ EXPECT_EQ(it->second->location().txn_id, info.txn_id);
}
TEST_F(PackedFileManagerTest, DisableFileCacheWriteFromS3FileWriter) {
@@ -334,7 +342,7 @@ TEST_F(PackedFileManagerTest,
DisableFileCacheWriteFromS3FileWriter) {
auto info = default_append_info();
ASSERT_TRUE(info.write_file_cache);
- EXPECT_TRUE(manager->append_small_file("s/no_cache", slice, info).ok());
+ EXPECT_TRUE(append_small_file("s/no_cache", slice, info).ok());
EXPECT_EQ(writer->append_calls(), 1);
EXPECT_EQ(writer->bytes_appended(), payload.size());
}
@@ -345,7 +353,7 @@ TEST_F(PackedFileManagerTest, AppendFailsWithoutTxnId) {
auto info = default_append_info();
info.txn_id = 0;
- Status st = manager->append_small_file("missing_txn", slice, info);
+ Status st = append_small_file("missing_txn", slice, info);
EXPECT_EQ(st.code(), doris::ErrorCode::INVALID_ARGUMENT);
EXPECT_EQ(manager->global_slice_locations_for_test().count("missing_txn"),
0);
}
@@ -358,7 +366,7 @@ TEST_F(PackedFileManagerTest, AppendLargeFileSkipped) {
std::string payload = "0123456789";
Slice slice(payload);
auto info = default_append_info();
- Status st = manager->append_small_file("large", slice, info);
+ Status st = append_small_file("large", slice, info);
EXPECT_TRUE(st.ok());
EXPECT_EQ(writer->append_calls(), 0);
EXPECT_EQ(manager->global_slice_locations_for_test().count("large"), 0);
@@ -372,7 +380,7 @@ TEST_F(PackedFileManagerTest, AppendWriterFailure) {
std::string payload = "data";
Slice slice(payload);
auto info = default_append_info();
- Status st = manager->append_small_file("broken", slice, info);
+ Status st = append_small_file("broken", slice, info);
EXPECT_FALSE(st.ok());
EXPECT_EQ(manager->global_slice_locations_for_test().count("broken"), 0);
}
@@ -385,10 +393,10 @@ TEST_F(PackedFileManagerTest,
AppendTriggersRotationWhenThresholdReached) {
Slice slice2(payload2);
auto info = default_append_info();
- EXPECT_TRUE(manager->append_small_file("file1", slice1, info).ok());
+ EXPECT_TRUE(append_small_file("file1", slice1, info).ok());
size_t create_before = file_system->created_paths().size();
- EXPECT_TRUE(manager->append_small_file("file2", slice2, info).ok());
+ EXPECT_TRUE(append_small_file("file2", slice2, info).ok());
EXPECT_EQ(file_system->created_paths().size(), create_before + 1);
EXPECT_EQ(manager->uploading_packed_files_for_test().size(), 1);
}
@@ -401,11 +409,11 @@ TEST_F(PackedFileManagerTest,
AppendTriggersRotationWhenFileCountThresholdReache
Slice slice(payload);
auto info = default_append_info();
- EXPECT_TRUE(manager->append_small_file("file1", slice, info).ok());
+ EXPECT_TRUE(append_small_file("file1", slice, info).ok());
EXPECT_TRUE(manager->uploading_packed_files_for_test().empty());
size_t created_before = file_system->created_paths().size();
- EXPECT_TRUE(manager->append_small_file("file2", slice, info).ok());
+ EXPECT_TRUE(append_small_file("file2", slice, info).ok());
EXPECT_EQ(manager->uploading_packed_files_for_test().size(), 1);
EXPECT_EQ(file_system->created_paths().size(), created_before + 1);
@@ -442,7 +450,7 @@ TEST_F(PackedFileManagerTest,
GetMergeFileIndexReturnsStoredValue) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- EXPECT_TRUE(manager->append_small_file("stored", slice, info).ok());
+ EXPECT_TRUE(append_small_file("stored", slice, info).ok());
PackedSliceLocation index;
EXPECT_TRUE(manager->get_packed_slice_location("stored", &index).ok());
EXPECT_EQ(index.size, payload.size());
@@ -456,7 +464,9 @@ TEST_F(PackedFileManagerTest,
WaitWriteDoneReturnsOkWhenAlreadyUploaded) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("path", slice, info).ok());
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("path", slice, info, &handle).ok());
+ ASSERT_NE(handle, nullptr);
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -470,14 +480,15 @@ TEST_F(PackedFileManagerTest,
WaitWriteDoneReturnsOkWhenAlreadyUploaded) {
manager->uploading_packed_files_for_test().clear();
uploading->upload_cv.notify_all();
- EXPECT_TRUE(manager->wait_upload_done("path").ok());
+ EXPECT_TRUE(manager->wait_upload_done(handle).ok());
}
TEST_F(PackedFileManagerTest, WaitWriteDoneReturnsErrorWhenFailed) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("bad", slice, info).ok());
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("bad", slice, info, &handle).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
auto uploading =
manager->uploading_packed_files_for_test().begin()->second;
@@ -491,15 +502,13 @@ TEST_F(PackedFileManagerTest,
WaitWriteDoneReturnsErrorWhenFailed) {
manager->uploading_packed_files_for_test().clear();
uploading->upload_cv.notify_all();
- auto status = manager->wait_upload_done("bad");
+ auto status = manager->wait_upload_done(handle);
EXPECT_FALSE(status.ok());
EXPECT_NE(status.to_string().find("meta failed"), std::string::npos);
}
-TEST_F(PackedFileManagerTest, WaitWriteDoneReturnsErrorWhenPathMissing) {
- auto indices = manager->global_slice_locations_for_test().find("ghost");
- ASSERT_EQ(indices, manager->global_slice_locations_for_test().end());
- auto status = manager->wait_upload_done("ghost");
+TEST_F(PackedFileManagerTest, WaitWriteDoneReturnsErrorWhenLocationMissing) {
+ auto status = manager->wait_upload_done(nullptr);
EXPECT_FALSE(status.ok());
}
@@ -507,7 +516,8 @@ TEST_F(PackedFileManagerTest,
WaitWriteDoneBlocksUntilUploadCompletes) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("waiting", slice, info).ok());
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("waiting", slice, info, &handle).ok());
auto current_state =
manager->current_packed_files_for_test()[_resource_id].get();
ASSERT_NE(current_state, nullptr);
@@ -520,7 +530,7 @@ TEST_F(PackedFileManagerTest,
WaitWriteDoneBlocksUntilUploadCompletes) {
std::future<void> ready_future = ready.get_future();
std::thread waiter([&] {
ready.set_value();
- auto status = manager->wait_upload_done("waiting");
+ auto status = manager->wait_upload_done(handle);
EXPECT_TRUE(status.ok());
});
@@ -560,7 +570,7 @@ TEST_F(PackedFileManagerTest,
ProcessUploadingFilesSetsFailedWhenMetaUpdateFails
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("meta_fail", slice, info).ok());
+ ASSERT_TRUE(append_small_file("meta_fail", slice, info).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -577,7 +587,7 @@ TEST_F(PackedFileManagerTest,
ProcessUploadingFilesCompletesAsyncUpload) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("async_success", slice, info).ok());
+ ASSERT_TRUE(append_small_file("async_success", slice, info).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -603,7 +613,7 @@ TEST_F(PackedFileManagerTest,
ProcessUploadingFilesSetsFailedWhenAsyncCloseFails
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("async_fail", slice, info).ok());
+ ASSERT_TRUE(append_small_file("async_fail", slice, info).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -630,7 +640,7 @@ TEST_F(PackedFileManagerTest,
ProcessUploadingFilesPollsAsyncCloseWithoutBlockin
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("async_poll_fail", slice,
info).ok());
+ ASSERT_TRUE(append_small_file("async_poll_fail", slice, info).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -664,7 +674,7 @@ TEST_F(PackedFileManagerTest,
AppendPackedFileInfoToFileTail) {
std::string payload = "abc";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("trailer_path", slice, info).ok());
+ ASSERT_TRUE(append_small_file("trailer_path", slice, info).ok());
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
@@ -716,6 +726,190 @@ TEST_F(PackedFileManagerTest,
CleanupExpiredDataRemovesOldEntries) {
EXPECT_TRUE(manager->uploaded_packed_files_for_test().empty());
}
+TEST_F(PackedFileManagerTest, CleanupKeepsIndexEntryWhileSliceIsStillHeld) {
+ config::uploaded_file_retention_seconds = -1;
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("long_load", slice, info, &handle).ok());
+ ASSERT_NE(handle, nullptr);
+
+ // Destroy the packed file context, so `handle` is the only holder besides
the index. It
+ // stands in for the writer that produced the slice: a load can run for
far longer than
+ // the retention time, and the location must still be there when it reads
the file back.
+ manager->current_packed_files_for_test().clear();
+
+ manager->cleanup_expired_data();
+ EXPECT_EQ(manager->global_slice_locations_for_test().count("long_load"),
1);
+
+ PackedSliceLocation location;
+ EXPECT_TRUE(manager->get_packed_slice_location("long_load",
&location).ok());
+ EXPECT_EQ(location.size, payload.size());
+}
+
+TEST_F(PackedFileManagerTest,
CleanupRemovesIndexEntryOnceNothingHoldsTheSlice) {
+ config::uploaded_file_retention_seconds = -1;
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ ASSERT_TRUE(append_small_file("short_load", slice, info).ok());
+ // Destroy the packed file context, so the index holds the only remaining
reference, the
+ // way it does once the writer is gone and the packed file has been
recycled
+ manager->current_packed_files_for_test().clear();
+
+ manager->cleanup_expired_data();
+ EXPECT_EQ(manager->global_slice_locations_for_test().count("short_load"),
0);
+}
+
+TEST_F(PackedFileManagerTest, WaitUploadDoneSurvivesRecycledContext) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("long_load", slice, info, &handle).ok());
+
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+ ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
+
+ auto uploading =
manager->uploading_packed_files_for_test().begin()->second;
+ auto* writer = dynamic_cast<MockFileWriter*>(uploading->writer.get());
+ ASSERT_NE(writer, nullptr);
+ uploading->state = PackedFileManager::PackedFileState::UPLOADING;
+ ASSERT_TRUE(writer->close(true).ok());
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+ ASSERT_EQ(manager->uploaded_packed_files_for_test().size(), 1);
+ EXPECT_EQ(handle->upload_state(), PackedSliceUploadState::UPLOADED);
+ EXPECT_EQ(handle->location().packed_file_size, uploading->total_size);
+
+ // A load that runs far longer than the retention time: the packed file
context is
+ // recycled while the writer that produced the slice is still open. A
negative retention
+ // expires everything, including entries created in this very second.
+ config::uploaded_file_retention_seconds = -1;
+ manager->uploaded_packed_files_for_test().begin()->second->upload_time = 1;
+ uploading.reset(); // the test's own reference, so recycling really drops
the context
+ manager->cleanup_expired_data();
+ ASSERT_TRUE(manager->uploaded_packed_files_for_test().empty());
+ // `handle` stands in for the still open writer, so the index entry is not
stale
+ EXPECT_EQ(manager->global_slice_locations_for_test().count("long_load"),
1);
+
+ EXPECT_TRUE(manager->wait_upload_done(handle).ok());
+}
+
+// The handle carries the upload result on its own, so it does not need the
index either
+TEST_F(PackedFileManagerTest, WaitUploadDoneSurvivesRemovedIndexEntry) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("long_load", slice, info, &handle).ok());
+
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+
+ auto uploading =
manager->uploading_packed_files_for_test().begin()->second;
+ auto* writer = dynamic_cast<MockFileWriter*>(uploading->writer.get());
+ ASSERT_NE(writer, nullptr);
+ uploading->state = PackedFileManager::PackedFileState::UPLOADING;
+ ASSERT_TRUE(writer->close(true).ok());
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+ ASSERT_EQ(handle->upload_state(), PackedSliceUploadState::UPLOADED);
+
+ uploading.reset();
+ manager->global_slice_locations_for_test().clear();
+ manager->uploaded_packed_files_for_test().clear();
+
+ EXPECT_TRUE(manager->wait_upload_done(handle).ok());
+ EXPECT_EQ(handle->location().size, payload.size());
+}
+
+TEST_F(PackedFileManagerTest,
WaitUploadDoneReportsFailureAfterContextRecycled) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr handle;
+ ASSERT_TRUE(append_small_file("long_load_fail", slice, info,
&handle).ok());
+
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+
+ auto uploading =
manager->uploading_packed_files_for_test().begin()->second;
+ auto* writer = dynamic_cast<MockFileWriter*>(uploading->writer.get());
+ ASSERT_NE(writer, nullptr);
+ uploading->state = PackedFileManager::PackedFileState::UPLOADING;
+ writer->set_close_status(Status::IOError("async close fail"));
+ ASSERT_TRUE(writer->close(true).ok());
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+ ASSERT_EQ(manager->uploaded_packed_files_for_test().size(), 1);
+ EXPECT_EQ(handle->upload_state(), PackedSliceUploadState::FAILED);
+
+ config::uploaded_file_retention_seconds = 0;
+ manager->uploaded_packed_files_for_test().begin()->second->upload_time = 1;
+ manager->cleanup_expired_data();
+ ASSERT_TRUE(manager->uploaded_packed_files_for_test().empty());
+
+ auto status = manager->wait_upload_done(handle);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Packed file upload failed"),
std::string::npos);
+}
+
+// Same path twice with no rotation in between: both slices live in one packed
file, and
+// slice_locations only keeps the last of them
+TEST_F(PackedFileManagerTest,
RewritingTheSamePathWithinOnePackedFileNotifiesBothSlices) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr shadowed_handle;
+ PackedSliceHandlePtr fresh_handle;
+ ASSERT_TRUE(append_small_file("reused", slice, info,
&shadowed_handle).ok());
+ ASSERT_TRUE(append_small_file("reused", slice, info, &fresh_handle).ok());
+ ASSERT_NE(shadowed_handle, fresh_handle);
+
+ auto* state = manager->current_packed_files_for_test()[_resource_id].get();
+ ASSERT_NE(state, nullptr);
+ EXPECT_EQ(state->slice_locations.size(), 1);
+ EXPECT_EQ(state->appended_slices.size(), 2);
+
+
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+ auto uploading =
manager->uploading_packed_files_for_test().begin()->second;
+ auto* writer = dynamic_cast<MockFileWriter*>(uploading->writer.get());
+ ASSERT_NE(writer, nullptr);
+ uploading->state = PackedFileManager::PackedFileState::UPLOADING;
+ ASSERT_TRUE(writer->close(true).ok());
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+
+ // The shadowed slice has a writer waiting on it too
+ EXPECT_EQ(shadowed_handle->upload_state(),
PackedSliceUploadState::UPLOADED);
+ EXPECT_EQ(fresh_handle->upload_state(), PackedSliceUploadState::UPLOADED);
+ EXPECT_EQ(shadowed_handle->location().packed_file_size,
uploading->total_size);
+ EXPECT_TRUE(manager->wait_upload_done(shadowed_handle).ok());
+}
+
+TEST_F(PackedFileManagerTest,
RewritingTheSamePathKeepsBothLocationsIndependent) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ PackedSliceHandlePtr stale_handle;
+ ASSERT_TRUE(append_small_file("reused", slice, info, &stale_handle).ok());
+
ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+ auto stale = manager->uploading_packed_files_for_test().begin()->second;
+
+ // The same path is written again, e.g. a retried publish rewriting the
same segment
+ PackedSliceHandlePtr fresh_handle;
+ ASSERT_TRUE(append_small_file("reused", slice, info, &fresh_handle).ok());
+ ASSERT_NE(stale_handle, fresh_handle);
+
+ auto* writer = dynamic_cast<MockFileWriter*>(stale->writer.get());
+ ASSERT_NE(writer, nullptr);
+ stale->state = PackedFileManager::PackedFileState::UPLOADING;
+ ASSERT_TRUE(writer->close(true).ok());
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+
+ // Finishing the older packed file says nothing about the newer write
+ EXPECT_EQ(stale_handle->upload_state(), PackedSliceUploadState::UPLOADED);
+ EXPECT_EQ(fresh_handle->upload_state(), PackedSliceUploadState::PENDING);
+}
+
TEST_F(PackedFileManagerTest, MergeFileBvarMetricsUpdated) {
manager->reset_packed_file_bvars_for_test();
auto state = std::make_shared<PackedFileManager::PackedFileContext>();
@@ -724,8 +918,8 @@ TEST_F(PackedFileManagerTest, MergeFileBvarMetricsUpdated) {
idx1.size = 100;
PackedSliceLocation idx2;
idx2.size = 200;
- state->slice_locations["a"] = idx1;
- state->slice_locations["b"] = idx2;
+ state->slice_locations["a"] = std::make_shared<PackedSliceHandle>(idx1);
+ state->slice_locations["b"] = std::make_shared<PackedSliceHandle>(idx2);
manager->record_packed_file_metrics_for_test(state.get());
@@ -763,15 +957,15 @@ TEST_F(PackedFileManagerTest,
MultipleSmallFilesTriggerMergeFile) {
std::string payload(file_size, 'a' + (i % 26)); // 50 bytes per file
Slice slice(payload);
- Status st = manager->append_small_file(path, slice, info);
+ Status st = append_small_file(path, slice, info);
EXPECT_TRUE(st.ok()) << "Failed to append file " << i << ": " <<
st.msg();
// Verify file is in global index map
auto it = manager->global_slice_locations_for_test().find(path);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, file_size);
- EXPECT_EQ(it->second.tablet_id, info.tablet_id);
- EXPECT_EQ(it->second.rowset_id, info.rowset_id);
+ EXPECT_EQ(it->second->location().size, file_size);
+ EXPECT_EQ(it->second->location().tablet_id, info.tablet_id);
+ EXPECT_EQ(it->second->location().rowset_id, info.rowset_id);
}
// Verify all files are in global index map
@@ -784,9 +978,9 @@ TEST_F(PackedFileManagerTest,
MultipleSmallFilesTriggerMergeFile) {
auto it = manager->global_slice_locations_for_test().find(path);
if (it != manager->global_slice_locations_for_test().end()) {
total_files_in_index++;
- total_size_in_index += it->second.size;
- EXPECT_LT(it->second.size, config::small_file_threshold_bytes)
- << "File " << path << " size " << it->second.size
+ total_size_in_index += it->second->location().size;
+ EXPECT_LT(it->second->location().size,
config::small_file_threshold_bytes)
+ << "File " << path << " size " <<
it->second->location().size
<< " should be less than threshold " <<
config::small_file_threshold_bytes;
}
}
@@ -819,7 +1013,7 @@ TEST_F(PackedFileManagerTest,
MultipleSmallFilesTriggerMergeFile) {
Slice slice_trigger(payload_trigger);
size_t create_before = file_system->created_paths().size();
- Status st = manager->append_small_file(path_trigger, slice_trigger, info);
+ Status st = append_small_file(path_trigger, slice_trigger, info);
EXPECT_TRUE(st.ok());
// Should create new merge file when threshold is reached
@@ -853,13 +1047,13 @@ TEST_F(PackedFileManagerTest,
FilesNearThresholdBoundaryMergeFile) {
std::string payload1(99, 'x'); // 99 bytes, just below threshold
Slice slice1(payload1);
- Status st = manager->append_small_file(path1, slice1, info);
+ Status st = append_small_file(path1, slice1, info);
EXPECT_TRUE(st.ok());
auto it = manager->global_slice_locations_for_test().find(path1);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, 99);
- EXPECT_LT(it->second.size, config::small_file_threshold_bytes);
+ EXPECT_EQ(it->second->location().size, 99);
+ EXPECT_LT(it->second->location().size,
config::small_file_threshold_bytes);
}
// Files exactly at threshold (should be merged, uses > not >=, so 100 <=
100 means merged)
@@ -868,14 +1062,14 @@ TEST_F(PackedFileManagerTest,
FilesNearThresholdBoundaryMergeFile) {
std::string payload2(100, 'y'); // 100 bytes, exactly at threshold
Slice slice2(payload2);
- Status st = manager->append_small_file(path2, slice2, info);
+ Status st = append_small_file(path2, slice2, info);
EXPECT_TRUE(st.ok());
// File at threshold should be merged (check uses >, so 100 is not >
100, so it's merged)
auto it = manager->global_slice_locations_for_test().find(path2);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, 100);
- EXPECT_LE(it->second.size, config::small_file_threshold_bytes);
+ EXPECT_EQ(it->second->location().size, 100);
+ EXPECT_LE(it->second->location().size,
config::small_file_threshold_bytes);
}
// Files at threshold + 1 (should NOT be merged)
@@ -885,7 +1079,7 @@ TEST_F(PackedFileManagerTest,
FilesNearThresholdBoundaryMergeFile) {
Slice slice3(payload3);
size_t append_calls_before = writer->append_calls();
- Status st = manager->append_small_file(path3, slice3, info);
+ Status st = append_small_file(path3, slice3, info);
EXPECT_TRUE(st.ok());
// File above threshold should not be merged
@@ -904,14 +1098,14 @@ TEST_F(PackedFileManagerTest,
FilesNearThresholdBoundaryMergeFile) {
std::string payload(file_size, 'a' + (i % 26));
Slice slice(payload);
- Status st = manager->append_small_file(path, slice, info);
+ Status st = append_small_file(path, slice, info);
EXPECT_TRUE(st.ok()) << "Failed to append file " << i << ": " <<
st.msg();
// Verify file is merged
auto it = manager->global_slice_locations_for_test().find(path);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, file_size);
- EXPECT_LT(it->second.size, config::small_file_threshold_bytes);
+ EXPECT_EQ(it->second->location().size, file_size);
+ EXPECT_LT(it->second->location().size,
config::small_file_threshold_bytes);
}
// Verify merge file state
@@ -938,15 +1132,15 @@ TEST_F(PackedFileManagerTest,
FilesNearThresholdBoundaryMergeFile) {
// Verify all merged files are at or below threshold (uses >, so ==
threshold is merged)
for (const auto& [path, index] : current_state->slice_locations) {
- EXPECT_LE(index.size, config::small_file_threshold_bytes)
- << "File " << path << " size " << index.size
+ EXPECT_LE(index->location().size, config::small_file_threshold_bytes)
+ << "File " << path << " size " << index->location().size
<< " should be less than or equal to threshold "
<< config::small_file_threshold_bytes;
}
for (const auto& [path, state] :
manager->uploading_packed_files_for_test()) {
for (const auto& [file_path, index] : state->slice_locations) {
- EXPECT_LE(index.size, config::small_file_threshold_bytes)
- << "File " << file_path << " size " << index.size
+ EXPECT_LE(index->location().size,
config::small_file_threshold_bytes)
+ << "File " << file_path << " size " <<
index->location().size
<< " should be less than or equal to threshold "
<< config::small_file_threshold_bytes;
}
@@ -991,7 +1185,7 @@ TEST_F(PackedFileManagerTest, TimeoutTriggersDirectUpload)
{
std::string payload = "small_file_data";
Slice slice(payload);
auto info = default_append_info();
- ASSERT_TRUE(manager->append_small_file("small_file_1", slice, info).ok());
+ ASSERT_TRUE(append_small_file("small_file_1", slice, info).ok());
// Get the merge file path before timeout
PackedSliceLocation index_before;
@@ -1085,14 +1279,14 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
std::string payload(file_size, 'a' + (i % 26));
Slice slice(payload);
- Status st = manager->append_small_file(path, slice, info);
+ Status st = append_small_file(path, slice, info);
EXPECT_TRUE(st.ok()) << "Failed to append file " << i << ": " <<
st.msg();
// Verify file is in global index map
auto it = manager->global_slice_locations_for_test().find(path);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, file_size);
- EXPECT_LT(it->second.size, config::small_file_threshold_bytes);
+ EXPECT_EQ(it->second->location().size, file_size);
+ EXPECT_LT(it->second->location().size,
config::small_file_threshold_bytes);
}
// Phase 2: Modify threshold to 200 bytes (larger than before)
@@ -1106,14 +1300,14 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
std::string payload(file_size, 'x' + (i % 26));
Slice slice(payload);
- Status st = manager->append_small_file(path, slice, info);
+ Status st = append_small_file(path, slice, info);
EXPECT_TRUE(st.ok()) << "Failed to append file phase2_" << i << ": "
<< st.msg();
// Verify file is in global index map
auto it = manager->global_slice_locations_for_test().find(path);
ASSERT_NE(it, manager->global_slice_locations_for_test().end());
- EXPECT_EQ(it->second.size, file_size);
- EXPECT_LT(it->second.size, config::small_file_threshold_bytes);
+ EXPECT_EQ(it->second->location().size, file_size);
+ EXPECT_LT(it->second->location().size,
config::small_file_threshold_bytes);
}
// Phase 4: Modify threshold to 30 bytes (smaller than file size)
@@ -1126,7 +1320,7 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
std::string payload(file_size, 'z');
Slice slice(payload);
- Status st = manager->append_small_file(path, slice, info);
+ Status st = append_small_file(path, slice, info);
EXPECT_TRUE(st.ok());
// Verify file is NOT in global index map (should be skipped)
@@ -1162,7 +1356,7 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
EXPECT_NE(it, manager->global_slice_locations_for_test().end())
<< "Phase 1 file " << i << " should be in global index map";
if (it != manager->global_slice_locations_for_test().end()) {
- EXPECT_EQ(it->second.size, file_size);
+ EXPECT_EQ(it->second->location().size, file_size);
}
}
@@ -1172,7 +1366,7 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
EXPECT_NE(it, manager->global_slice_locations_for_test().end())
<< "Phase 2 file " << i << " should be in global index map";
if (it != manager->global_slice_locations_for_test().end()) {
- EXPECT_EQ(it->second.size, file_size);
+ EXPECT_EQ(it->second->location().size, file_size);
}
}
@@ -1189,7 +1383,7 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
Slice slice_trigger(payload_trigger);
size_t create_before = file_system->created_paths().size();
- Status st = manager->append_small_file(path_trigger, slice_trigger,
info);
+ Status st = append_small_file(path_trigger, slice_trigger, info);
EXPECT_TRUE(st.ok());
// Should create new merge file when threshold is reached (if not
already)
@@ -1200,7 +1394,7 @@ TEST_F(PackedFileManagerTest,
ModifyThresholdDuringContinuousImport) {
EXPECT_NE(trigger_it, manager->global_slice_locations_for_test().end())
<< "Trigger file should be in global index map after threshold
change";
if (trigger_it != manager->global_slice_locations_for_test().end()) {
- EXPECT_EQ(trigger_it->second.size, trigger_file_size);
+ EXPECT_EQ(trigger_it->second->location().size, trigger_file_size);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
