HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3878648800
##########
test/inte/realtime_write_inte_test.cpp:
##########
@@ -71,6 +81,592 @@
#include "paimon/write_context.h"
namespace paimon::test {
+namespace {
+
+bool HasSuffix(const std::string& value, const std::string& suffix) {
+ return value.size() >= suffix.size() &&
+ value.compare(value.size() - suffix.size(), suffix.size(), suffix)
== 0;
+}
+
+Result<std::set<std::string>> ListPhysicalArtifacts(const
std::shared_ptr<FileSystem>& file_system,
+ const std::string& root) {
+ std::set<std::string> artifacts;
+ std::vector<std::string> directories = {root};
+ while (!directories.empty()) {
+ std::string directory = std::move(directories.back());
+ directories.pop_back();
+ std::vector<BasicFileStatus> statuses;
+ PAIMON_RETURN_NOT_OK(file_system->ListDir(directory, &statuses));
+ for (const BasicFileStatus& status : statuses) {
+ if (status.IsDir()) {
+ directories.push_back(status.GetPath());
+ } else if (HasSuffix(status.GetPath(), ".orc") ||
+ HasSuffix(status.GetPath(), ".index") ||
+ HasSuffix(status.GetPath(), ".channel")) {
+ artifacts.insert(status.GetPath());
+ }
+ }
+ }
+ return artifacts;
+}
+
+class FailAllocationMemoryPool final : public MemoryPool {
+ public:
+ explicit FailAllocationMemoryPool(const std::shared_ptr<MemoryPool>&
delegate)
+ : delegate_(delegate) {}
+
+ void FailAfterAllocations(int64_t successful_allocations) {
+ allocations_before_failure_.store(successful_allocations,
std::memory_order_release);
+ }
+
+ void* Malloc(uint64_t size, uint64_t alignment = 0) override {
+ if (ShouldFail()) {
+ throw std::bad_alloc();
+ }
+ return delegate_->Malloc(size, alignment);
+ }
+
+ void* Realloc(void* p, size_t old_size, size_t new_size, uint64_t
alignment = 0) override {
+ if (ShouldFail()) {
+ throw std::bad_alloc();
+ }
+ return delegate_->Realloc(p, old_size, new_size, alignment);
+ }
+
+ void Free(void* p, uint64_t size) override {
+ delegate_->Free(p, size);
+ }
+
+ void Free(void* p, uint64_t size, uint64_t alignment) override {
+ delegate_->Free(p, size, alignment);
+ }
+
+ uint64_t CurrentUsage() const override {
+ return delegate_->CurrentUsage();
+ }
+
+ uint64_t MaxMemoryUsage() const override {
+ return delegate_->MaxMemoryUsage();
+ }
+
+ private:
+ bool ShouldFail() {
+ int64_t remaining =
allocations_before_failure_.load(std::memory_order_acquire);
+ while (remaining >= 0) {
+ if (allocations_before_failure_.compare_exchange_weak(remaining,
remaining - 1,
+
std::memory_order_acq_rel)) {
+ return remaining == 0;
+ }
+ }
+ return false;
+ }
+
+ std::shared_ptr<MemoryPool> delegate_;
+ std::atomic<int64_t> allocations_before_failure_{-1};
+};
+
Review Comment:
Agreed, and I narrowed the test scope in
`813c6444e39dba4076018a34126a3e3b3dfc176c`. The custom fault-injection helpers,
`SplitBatchReader`, and the related exception and malformed-reader cases were
removed, leaving the basic streaming PK and core lifecycle coverage in this PR.
Those exception paths are not covered here anymore; they can be added
separately once we have representative production scenarios for them.
--
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]
