lxy-9602 commented on code in PR #453:
URL: https://github.com/apache/paimon-cpp/pull/453#discussion_r4357274542
##########
src/paimon/core/operation/data_evolution_split_read_test.cpp:
##########
@@ -751,6 +751,95 @@ TEST_F(DataEvolutionSplitReadTest,
TestPlanIndependentlyRolledVectorReadRanges)
ASSERT_EQ(vector1->row_count, 4);
}
+TEST_F(DataEvolutionSplitReadTest, TestPlanSequentialVectorRead) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 3, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 103, 4, 2);
+ auto vector2 = CreateDataFileMeta("vector2.vector.vortex", 107, 3, 3);
+ for (const auto& file : {vector0, vector1, vector2}) {
+ file->write_cols = std::vector<std::string>{"embedding"};
+ }
+ auto field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 2} :
std::vector<int32_t>{1};
+ };
+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(
+ {vector2, main, vector0, vector1}, field_ids));
+ ASSERT_EQ(bunches.size(), 2);
+ ASSERT_EQ(bunches[0]->Files(),
std::vector<std::shared_ptr<DataFileMeta>>{main});
+ ASSERT_EQ(bunches[1]->Files(),
+ (std::vector<std::shared_ptr<DataFileMeta>>{vector0, vector1,
vector2}));
+ ASSERT_EQ(bunches[0]->RowCount(), 10);
+ ASSERT_EQ(bunches[1]->RowCount(), 10);
+ ASSERT_EQ(vector1->first_row_id, 103);
+ ASSERT_EQ(vector1->row_count, 4);
+}
+
+TEST_F(DataEvolutionSplitReadTest,
TestSequentialVectorReadFallsBackForUnsupportedLayouts) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 4, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 104, 6, 1);
+ vector0->write_cols = vector1->write_cols =
std::vector<std::string>{"embedding"};
+ auto field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 2} :
std::vector<int32_t>{1};
+ };
+ auto check_fallback = [&](const
std::vector<std::shared_ptr<DataFileMeta>>& files) {
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(files,
field_ids));
+ ASSERT_TRUE(bunches.empty());
+ };
+ check_fallback({vector0, vector1});
+ check_fallback({main, main, vector0, vector1});
+ check_fallback({main, vector0, vector1, CreateBlobFile("payload", 100, 10,
1, {{"payload"}})});
+ check_fallback({main, vector1});
+ vector1->first_row_id = 103;
+ check_fallback({main, vector0, vector1});
+ vector1->first_row_id = 105;
+ check_fallback({main, vector0, vector1});
+ vector1->first_row_id = 104;
+ vector1->row_count = 7;
+ check_fallback({main, vector0, vector1});
+ vector1->row_count = 6;
+ vector1->schema_id = 1;
+ check_fallback({main, vector0, vector1});
+ vector1->schema_id = 0;
+ vector1->file_name = "vector1.vector.parquet";
+ check_fallback({main, vector0, vector1});
+ vector1->file_name = "vector1.vector.vortex";
+ vector1->write_cols = std::vector<std::string>{"another_embedding"};
+ check_fallback({main, vector0, vector1});
+ vector1->write_cols = vector0->write_cols =
+ std::vector<std::string>{"embedding", "another_embedding"};
+ check_fallback({main, vector0, vector1});
+
+ vector1->write_cols = vector0->write_cols =
std::vector<std::string>{"embedding"};
+ auto inline_field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 1, 2} :
std::vector<int32_t>{1};
+ };
+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(
+ {main, vector0, vector1}, inline_field_ids));
+ ASSERT_TRUE(bunches.empty());
+}
+
+TEST_F(DataEvolutionSplitReadTest,
TestSequentialVectorReadPropagatesMetadataErrors) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector = CreateDataFileMeta("vector.vector.vortex", 100, 10, 1);
+ vector->write_cols = std::vector<std::string>{"embedding"};
+ auto failed_schema = [](const std::shared_ptr<DataFileMeta>&) ->
Result<std::vector<int32_t>> {
+ return Status::IOError("schema unavailable");
+ };
+ auto result =
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead({main, vector},
failed_schema);
+ ASSERT_FALSE(result.ok());
+ ASSERT_TRUE(result.status().IsIOError());
+ ASSERT_NE(result.status().ToString().find("schema unavailable"),
std::string::npos);
Review Comment:
ASSERT_NOK_WITH_MSG
##########
test/inte/data_evolution_table_test.cpp:
##########
@@ -57,6 +61,36 @@
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
+namespace {
+class CountingLocalFileSystem : public LocalFileSystem {
+ public:
+ explicit CountingLocalFileSystem(const std::string& failed_path = "")
+ : failed_path_(failed_path) {}
+
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ {
+ std::scoped_lock lock(mutex_);
+ ++open_counts_[path];
+ }
+ if (path == failed_path_) {
+ return Status::IOError("injected open failure: " + path);
+ }
+ return LocalFileSystem::Open(path);
+ }
+
+ int64_t OpenCount(const std::string& path) const {
+ std::scoped_lock lock(mutex_);
+ auto iter = open_counts_.find(path);
+ return iter == open_counts_.end() ? 0 : iter->second;
+ }
+
+ private:
+ mutable std::mutex mutex_;
+ mutable std::map<std::string, int64_t> open_counts_;
+ std::string failed_path_;
+};
+} // namespace
+
Review Comment:
In the integration test, testing “lazy open” and the exact file-open count
seems a bit too fine-grained. We may just verify that the query results are
correct.
--
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]
