Copilot commented on code in PR #875:
URL: https://github.com/apache/iceberg-cpp/pull/875#discussion_r4481733364
##########
src/iceberg/arrow/arrow_io.cc:
##########
@@ -423,29 +424,39 @@ class ArrowInputFile : public InputFile {
std::string_view location() const override { return location_; }
Result<int64_t> Size() const override {
+ std::lock_guard lock(mutex_);
if (file_size_.has_value()) {
return *file_size_;
}
::arrow::fs::FileInfo file_info(path_, ::arrow::fs::FileType::File);
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, fs_->OpenInputFile(file_info));
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto size, input->GetSize());
+ ICEBERG_ARROW_RETURN_NOT_OK(input->Close());
+ file_size_ = size;
Review Comment:
If `input->GetSize()` fails, the function returns early without closing
`input`, which can leak file descriptors/handles on the error path. Close
`input` regardless of `GetSize()` success (e.g., capture the status from
`GetSize()`, then `Close()`, and return the original failure augmented with any
close failure).
##########
src/iceberg/util/cache_internal.h:
##########
@@ -76,41 +99,80 @@ class LruCache {
return &list_it->value;
}
+ // Insert or update the value for `key`, mark it as most recently used, and
evict least
+ // recently used entries until the cache is within its maximum weight again.
+ //
+ // Returns whether a new entry was added, and a pointer to the cached value.
A value
+ // heavier than the maximum weight is not cached: any existing entry for
`key` is
+ // removed, other entries are kept, and {false, nullptr} is returned.
template <typename K, typename V>
std::pair<bool, Value*> Replace(K&& key, V&& value) {
// Try to insert temporary iterator
auto [it, inserted] = map_.emplace(std::forward<K>(key), ListIt{});
+ ListIt list_it;
if (inserted) {
// Inserted => push item at front of the list, and update iterator
items_.emplace_front(&it->first, std::forward<V>(value));
- it->second = items_.begin();
- // Did we exceed the cache capacity? If so, remove least recently used
item
- if (static_cast<int32_t>(items_.size()) > capacity_) {
- const bool erased = map_.erase(*items_.back().key);
- ICEBERG_DCHECK(erased, "Failed to erase least recently used cache
item");
- static_cast<void>(erased);
- items_.pop_back();
- }
- return {true, &it->second->value};
+ list_it = items_.begin();
+ it->second = list_it;
+ } else {
+ // Already exists => move item at front of the list, and update value
+ list_it = it->second;
+ items_.splice(items_.begin(), items_, list_it);
+ list_it->value = std::forward<V>(value);
+ total_weight_ -= list_it->weight;
+ list_it->weight = 0;
}
- // Already exists => move item at front of the list, and update value
- auto list_it = it->second;
- items_.splice(items_.begin(), items_, list_it);
- list_it->value = std::forward<V>(value);
- return {false, &list_it->value};
+ const int64_t weight = weigher_(it->first, list_it->value);
+ ICEBERG_DCHECK(weight >= 0, "Cache entry weight must be non-negative");
+ if (weight > max_weight_) {
+ items_.erase(list_it);
+ map_.erase(it);
+ return {false, nullptr};
+ }
+
+ // The item is at the front of the list and not yet counted in
total_weight_, and
+ // `max_weight_ - weight` is non-negative, so eviction stops before
reaching it.
+ while (total_weight_ > max_weight_ - weight) {
+ EvictLeastRecentlyUsed();
+ }
+ list_it->weight = weight;
+ total_weight_ += weight;
+ return {inserted, &list_it->value};
}
private:
struct Item {
// Pointer to the key inside the unordered_map
const Key* key;
Value value;
+ int64_t weight = 0;
};
using List = std::list<Item>;
using ListIt = typename List::iterator;
- const int32_t capacity_;
+ void EvictLeastRecentlyUsed() {
+ ICEBERG_DCHECK(!items_.empty(), "Cannot evict from an empty cache");
+ const Item& item = items_.back();
+ total_weight_ -= item.weight;
+ const bool erased = map_.erase(*item.key);
+ ICEBERG_DCHECK(erased, "Failed to erase least recently used cache item");
+ static_cast<void>(erased);
+ items_.pop_back();
+ }
+
+ void ReserveForUnitWeight() {
+ // Only with unit weights does the maximum weight bound the number of
entries. The
+ // map size can temporarily exceed it by one, see Replace().
+ if constexpr (std::is_same_v<Weigher, UnitWeigher>) {
+ map_.reserve(static_cast<size_t>(max_weight_) + 1);
+ }
+ }
Review Comment:
`ReserveForUnitWeight()` uses `std::is_same_v`, which requires
`<type_traits>` to be included by this header. Add `#include <type_traits>` to
avoid relying on transitive includes (this is a build break on toolchains that
don't indirectly include it).
##########
src/iceberg/file_io.cc:
##########
@@ -19,15 +19,98 @@
#include "iceberg/file_io.h"
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
#include <limits>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <unordered_map>
#include <utility>
+#include "iceberg/util/cache_internal.h"
#include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
namespace iceberg {
+namespace internal {
+
+using MetadataContent = std::shared_ptr<const std::string>;
+
+// Weighs cached metadata content by its size in bytes.
+struct MetadataContentWeigher {
+ int64_t operator()(const std::string& /*location*/,
+ const MetadataContent& content) const {
+ return static_cast<int64_t>(content->size());
+ }
+};
+
+// The metadata content cache behind a FileIO. The mutex guards every member.
`options`
+// is set by the first ConfigureMetadataCache call, and `lru` exists only when
that
+// configuration enabled caching.
+struct MetadataCacheState {
+ std::mutex mutex;
+ std::optional<MetadataCacheOptions> options;
+ std::optional<LruCache<std::string, MetadataContent,
MetadataContentWeigher>> lru;
+};
+
+} // namespace internal
+
namespace {
+using internal::MetadataCacheState;
+using internal::MetadataContent;
+
+constexpr auto kMaxCacheWeight =
static_cast<size_t>(std::numeric_limits<int64_t>::max());
+
+template <typename T>
+Result<T> ParseNumberProperty(
+ const std::unordered_map<std::string, std::string>& properties,
std::string_view key,
+ T default_value) {
Review Comment:
`kMaxCacheWeight` is stored as `size_t` but represents an `int64_t`
boundary; on 32-bit platforms this truncates and makes the subsequent 'exceeds
int64_t max' checks/message inaccurate (and may reject values far below
`int64_t::max`). Consider expressing this limit as `uint64_t` or `int64_t` (and
comparing via safe casts) so the validation aligns with the error message and
the `LruCache`'s `int64_t` weight type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
