This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a7c0a7843a0 [fix](cloud) bind the packed slice location lifetime to
its writer (#67347)
a7c0a7843a0 is described below
commit a7c0a7843a0a363b9f298c4a35d5914061ad0784
Author: Xin Liao <[email protected]>
AuthorDate: Tue Sep 15 01:06:44 2026 +0800
[fix](cloud) bind the packed slice location lifetime to its writer (#67347)
Problem Summary:
A small file is handed to `PackedFileManager` when the segment is
flushed, but its slice
location is read again only when the rowset is closed —
`wait_upload_done()` from
`finish_close()`, and `get_packed_slice_location()` when the rowset meta
is built. The gap
between the two is the duration of the whole load, not the lifetime of
the packed file.
Both reads went through `_global_slice_locations`, which
`cleanup_expired_data()` recycles
purely on the age of the small file (`uploaded_file_retention_seconds`,
1800s by default).
Any load running longer than that lost the mapping and failed with:
```
[INTERNAL_ERROR]File not found in global index:
data/<tablet_id>/<segment>_0.idx
```
The packed file itself had been uploaded successfully — only the
in-memory mapping was gone.
We hit this in production with a broker load that ran for 38 minutes.
Two related points:
`_uploaded_packed_files` is recycled on the same config, so a waiter
could also fail with
`Packed file not found for path: ...`; and this is not specific to
inverted index files,
since segments smaller than `small_file_threshold_bytes` take the same
async-close /
late-wait path.
### What is changed and how it works?
A slice location should live as long as whoever may still read it, not
until a wall clock
deadline.
1. `append_small_file()` returns the location as a `std::shared_ptr` and
`PackedFileWriter`
keeps it for as long as it lives. `wait_upload_done()` and the writer's
`get_packed_slice_location()` take that reference, and
`CloudRowsetWriter` asks the writer
instead of the manager, so neither can fail because the index was
recycled. The terminal
upload state (UPLOADED / FAILED) is recorded on the slices, so a waiter
does not need the
`PackedFileContext` either.
2. The index itself gets the same property.
`PackedFileSystem::open_file_impl()` falls back
to `_global_slice_locations` for a file whose location has not reached
the rowset meta
yet, which is every read during the load — `SegmentIndexFileCacheLoader`
does exactly that
when it preloads a segment's index and footer ranges into file cache
right after close.
Since a location is now a `shared_ptr`, the cleanup can tell whether an
entry is still
needed: entries some writer or packed file context still holds are
skipped, and the age
check applies only to those the index alone holds. The last owner going
away drops the
reference, so an entry outlives everything that could still read it.
3. `PackedFileContext::slice_locations` is keyed by the small file path,
so one path written
twice into the same packed file left only the second handle in there and
the first never
reached a terminal state. Every appended handle is now kept alongside
that map and
notified from it. No caller does this today —
`CloudTablet::create_transient_rowset_writer()` sets `allow_packed_file
= false`, so the
MOW publish retry that writes one segment path twice never produces
packed files — but
the interface allowed it and failed quietly.
Behavior note: an index entry is no longer guaranteed to be dropped
after
`uploaded_file_retention_seconds`. It now stays for as long as a writer
or a packed file
context holds the slice, so a load producing N packed small files keeps
N entries resident
until it finishes. That is the point of the fix — those entries were
being dropped while
still in use — but it does change how much the map holds during a long
load.
Behavior for short loads is unchanged.
### Release note
Fix `File not found in global index` failures for loads that run longer
than
`uploaded_file_retention_seconds` when packed file is enabled.
---
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 ca39096c9cf..9753ca3dd99 100644
--- a/be/src/cloud/cloud_rowset_writer.cpp
+++ b/be/src/cloud/cloud_rowset_writer.cpp
@@ -223,10 +223,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 9245387379a..9cc76eb803e 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;
@@ -73,14 +120,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
@@ -121,6 +171,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
@@ -134,7 +188,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;
@@ -180,8 +238,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]
