[fix](cloud) bind the packed slice location lifetime to its writer

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
     }

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;

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 @@ class PackedFileManager {
     // 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 @@ class PackedFileManager {
     // 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 @@ class PackedFileManager {
     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 @@ class PackedFileManager {
     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

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,19 +217,29 @@ 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();
 }
 
 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;

be/src/io/fs/packed_file_writer.h

@@ -89,9 +89,10 @@ class PackedFileWriter final : public FileWriter {
     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;