lxy-9602 commented on code in PR #334:
URL: https://github.com/apache/paimon-cpp/pull/334#discussion_r4003694979
##########
test/inte/scan_and_read_inte_test.cpp:
##########
@@ -604,6 +604,32 @@ TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot5) {
ASSERT_EQ(count, read_result->length());
}
+// One reader is built per data file of a split, and those readers are built
in parallel.
+// Snapshot 5 has a bucket holding five files, so the parallel build path is
exercised here and
+// has to return the very same rows in the very same order as the other
snapshot-5 read above.
+TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot5WithReaderBuildParallelism)
{
+ auto file_format = FileFormat();
+ std::string table_path = GetDataDir() + "/" + file_format +
"/append_09.db/append_09";
+
+ ScanContextBuilder scan_context_builder(table_path);
+ scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5");
+ ASSERT_OK_AND_ASSIGN(auto scan_context,
FinishScanContext(scan_context_builder));
+ ASSERT_OK_AND_ASSIGN(auto table_scan,
TableScan::Create(std::move(scan_context)));
+ ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan());
+ auto splits = result_plan->Splits();
+ ASSERT_EQ(3, splits.size());
+
+ ReadContextBuilder read_context_builder(table_path);
+ AddReadOptionsForPrefetch(&read_context_builder);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context,
read_context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto table_read,
TableRead::Create(std::move(read_context)));
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits));
+ ASSERT_OK_AND_ASSIGN(auto read_result,
+
ReadResultCollector::CollectResult(std::move(batch_reader)));
+ ASSERT_TRUE(read_result);
+ ASSERT_GT(read_result->length(), 0);
Review Comment:
I don’t see how `ASSERT_GT(read_result->length(), 0);` matches the comment
saying it “has to return the very same rows in the very same order as the other
snapshot-5 read above.” At a minimum, shouldn’t we compare the actual query
results?
##########
test/inte/read_inte_test.cpp:
##########
@@ -2818,6 +2818,53 @@ TEST_P(ReadInteTest, TestAppendReadWithSchemaEvolution) {
}
}
+// Building a reader for a file written under an older schema goes through
SchemaManager, and the
+// readers of one split are built concurrently. Every split here mixes both
schema ids, so the
+// parallel build has to resolve each file against the right schema and return
the right rows.
+TEST_P(ReadInteTest,
TestAppendReadWithSchemaEvolutionWithReaderBuildParallelism) {
+ auto param = GetParam();
+ std::string path = paimon::test::GetDataDir() + "/" + param.file_format +
+
"/append_table_with_alter_table.db/append_table_with_alter_table/";
+
+ std::vector<std::string> file_list_0;
+ std::vector<std::string> file_list_1;
+ if (param.file_format == "orc") {
+ file_list_0 = {"data-2190cec3-ce87-4175-8d19-9268becf4440-0.orc",
+ "data-b34cd128-03e3-4e70-ba9c-5dec2183849c-0.orc"};
+ file_list_1 = {"data-13824b84-8572-4a20-b712-c0475d1828b4-0.orc",
+ "data-492ed5ab-4740-4e93-8a0a-79a6893b1770-0.orc"};
+ } else if (param.file_format == "parquet") {
+ file_list_0 = {"data-512651de-64b5-4a10-8068-65403aaccdb8-0.parquet",
+ "data-1aaec161-5365-426f-b33d-3cd99a3908f2-0.parquet"};
+ file_list_1 = {"data-11b12094-192f-4ad8-92a8-ae8cba5e25ef-0.parquet",
+ "data-9dfb749f-0509-4db2-ae7b-1e4448b32165-0.parquet"};
+ }
+
+ DataSplitsSchema input_data_splits = {
+ {path + "key0=0/key1=1/bucket-0", BinaryRowGenerator::GenerateRow({0,
1}, pool_.get()),
+ file_list_0,
+ /*schema ids*/ {0, 1}},
+ {path + "key0=1/key1=1/bucket-0", BinaryRowGenerator::GenerateRow({1,
1}, pool_.get()),
+ file_list_1,
+ /*schema ids*/ {1, 0}}};
+ auto data_splits = CreateDataSplits(input_data_splits, /*snapshot_id=*/2);
+
+ ReadContextBuilder context_builder(path);
+ context_builder.SetReadAheadCacheEnabled(param.read_ahead_cache_enabled);
+ context_builder.AddOption(Options::FILE_FORMAT, param.file_format)
+ .AddOption("read.batch-size", "2");
+ context_builder.EnablePrefetch(param.enable_prefetch)
+ .AddOption("test.enable-adaptive-prefetch-strategy",
+ param.enable_adaptive_prefetch_strategy);
+ ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto table_read,
TableRead::Create(std::move(read_context)));
+ ASSERT_OK_AND_ASSIGN(auto batch_reader,
table_read->CreateReader(data_splits));
+ ASSERT_OK_AND_ASSIGN(auto result_array,
+
ReadResultCollector::CollectResult(std::move(batch_reader)));
+ ASSERT_TRUE(result_array);
+ ASSERT_GT(result_array->length(), 0);
Review Comment:
Please compare the actual results to verify that the read is correct.
##########
src/paimon/core/operation/abstract_split_read.cpp:
##########
@@ -64,6 +68,29 @@ class FileStorePathFactory;
class MemoryPool;
class Predicate;
+namespace {
+
+/// Set while a thread is building one data file reader. Building a reader may
reach
+/// `CreateRawFileReaders` again, and that nested call has to stay serial: it
would
+/// otherwise submit tasks to the reader build pool and block a worker of that
same
+/// pool waiting for them.
+thread_local bool building_reader = false;
+
+class BuildingReaderGuard {
+ public:
+ BuildingReaderGuard() {
+ building_reader = true;
+ }
+ ~BuildingReaderGuard() {
+ building_reader = false;
+ }
+
+ BuildingReaderGuard(const BuildingReaderGuard&) = delete;
+ BuildingReaderGuard& operator=(const BuildingReaderGuard&) = delete;
+};
+
+} // namespace
+
Review Comment:
It seems that `BuildingReaderGuard` was introduced to handle recursive
`CreateRawFileReaders` calls, but I don’t clearly see where such recursion
happens in the current code. Could you clarify why this `thread_local` variable
is needed?
--
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]
