lxy-9602 commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3877717989
##########
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;
Review Comment:
Consider use `StringUtils::StartsWith`
##########
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:
Could we keep the tests in this PR focused on the basic streaming PK
functionality and core scenarios, and move exception-related cases into a
separate PR? The current fault-injection style is a bit different from the rest
of the codebase, so reviewing it separately would make it easier to understand
and evaluate. Also, for some classes like `SplitBatchReader`, it doesn’t seem
like the current code path would actually hit the simulated issues, so it might
make more sense to add that kind of simulation when the scenario really arises.
--
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]
