diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index a25397b98..876d600bd 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -63,6 +63,7 @@ set(PAIMON_COMMON_SRCS common/data/variant/variant_type_utils.cpp common/defs.cpp common/executor/executor.cpp + common/executor/reader_build_executor.cpp common/factories/singleton.cpp common/factories/io_hook.cpp common/factories/factory_creator.cpp diff --git a/src/paimon/common/executor/reader_build_executor.cpp b/src/paimon/common/executor/reader_build_executor.cpp new file mode 100644 index 000000000..685119fe3 --- /dev/null +++ b/src/paimon/common/executor/reader_build_executor.cpp @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/common/executor/reader_build_executor.h" + +#include +#include +#include + +#include "paimon/result.h" + +namespace paimon { + +std::shared_ptr GetReaderBuildExecutor() { + static std::shared_ptr executor; + static std::once_flag once; + std::call_once(once, []() { + const uint32_t hardware_threads = std::thread::hardware_concurrency(); + const uint32_t upper_bound = hardware_threads > 0 ? hardware_threads : 1; + const uint32_t threads = std::clamp(kReaderBuildMaxParallelNum, 1u, upper_bound); + // CreateDefaultExecutor only rejects a zero thread count, which the clamp rules out. + executor = CreateDefaultExecutor(threads).value(); + }); + return executor; +} + +} // namespace paimon diff --git a/src/paimon/common/executor/reader_build_executor.h b/src/paimon/common/executor/reader_build_executor.h new file mode 100644 index 000000000..7053512c4 --- /dev/null +++ b/src/paimon/common/executor/reader_build_executor.h @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include + +#include "paimon/executor.h" + +namespace paimon { + +/// Number of data file readers built in parallel within one split, and the size of the dedicated +/// reader build pool (clamped to hardware concurrency). Building a reader mostly waits on remote +/// I/O, so a small fixed pool overlaps those waits without unbounded thread growth. +inline constexpr uint32_t kReaderBuildMaxParallelNum = 4; + +/// Returns the process-wide thread pool dedicated to building data file readers. +/// +/// This pool is kept separate from the executor of a read context on purpose. That executor is +/// handed to the readers this pool builds, and they use it for async prefetch once reading +/// starts; running the per-file builds on a dedicated pool keeps the blocking build phase from +/// contending with it, and keeps the pool a single reused instance instead of a fresh one per +/// split. +/// +/// The pool is created once, on first use, sized to `kReaderBuildMaxParallelNum` clamped to +/// hardware concurrency. Callers that build readers serially never call this and therefore never +/// pay for any thread. +std::shared_ptr GetReaderBuildExecutor(); + +} // namespace paimon diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp index da7ba4588..2b2c7ec8b 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp @@ -38,6 +38,7 @@ #include "paimon/common/utils/scope_guard.h" #include "paimon/format/reader_builder.h" #include "paimon/fs/file_system.h" +#include "paimon/memory/memory_pool.h" namespace arrow { class Schema; @@ -233,10 +234,44 @@ Result> PrefetchFileBatchReaderImpl if (enable_io_metrics) { io_metrics = std::make_shared(); } + // Opening a file mostly waits on remote I/O, so the streams are all opened concurrently + // instead of opening the cache's stream first and paying its round trip before the readers' + // begin. Both waves run on the read context's `executor`, which is sized to the prefetch + // parallelism, so no separate pool is spun up and torn down per file and the open/build + // concurrency follows the configured parallelism instead of a fixed default. That reuse is + // safe: neither wave's tasks wait on one another or on `executor`, and Create never runs on + // an `executor` worker, so the CollectAll calls below only ever wait on tasks that progress. + // The build has to be a second wave rather than folded into the first: each build task wraps + // its stream in the cache, and a task waiting on the cache would block a worker while the + // cache's own open is still queued behind it. Wave one only opens, so its tasks never wait on + // one another. + const uint32_t open_count = + read_ahead_cache_enabled ? prefetch_max_parallel_num + 1 : prefetch_max_parallel_num; + std::vector>>> open_futures; + open_futures.reserve(open_count); + for (uint32_t i = 0; i < open_count; i++) { + open_futures.push_back( + Via(executor.get(), + [&fs, &data_file_path, data_file_size]() -> Result> { + return fs->Open(FileStatus(data_file_path, data_file_size)); + })); + } + // The tasks only reference locals of this frame, which stay alive because both CollectAll + // calls below drain every future before returning. + std::vector>> opened_streams = CollectAll(open_futures); + std::vector> streams; + streams.reserve(opened_streams.size()); + for (auto& opened_stream : opened_streams) { + if (!opened_stream.ok()) { + return opened_stream.status(); + } + streams.push_back(std::move(opened_stream).value()); + } + + size_t next_stream = 0; std::shared_ptr cache; if (read_ahead_cache_enabled) { - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input_stream, - fs->Open(FileStatus(data_file_path, data_file_size))); + std::shared_ptr input_stream = std::move(streams[next_stream++]); if (io_metrics) { input_stream = std::make_shared(input_stream, io_metrics); } @@ -246,22 +281,24 @@ Result> PrefetchFileBatchReaderImpl cache = std::make_shared(input_stream, cache_config, static_cast(data_file_size), pool); } + // Wave two builds the readers, which reads each file's footer and so still needs the cache + // above to exist. The builds are concurrent, like the opens were. std::vector>>> futures; + futures.reserve(prefetch_max_parallel_num); for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) { - futures.push_back(Via( - executor.get(), - [&fs, &data_file_path, data_file_size, &reader_builder, &cache, - io_metrics]() -> Result> { - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr input_stream, - fs->Open(FileStatus(data_file_path, data_file_size))); - if (io_metrics) { - input_stream = - std::make_unique(std::move(input_stream), io_metrics); - } - auto cache_input_stream = - std::make_shared(std::move(input_stream), cache); - return reader_builder->Build(cache_input_stream); - })); + futures.push_back(Via(executor.get(), + [&reader_builder, &cache, &streams, i, next_stream, + io_metrics]() -> Result> { + std::unique_ptr input_stream = + std::move(streams[next_stream + i]); + if (io_metrics) { + input_stream = std::make_unique( + std::move(input_stream), io_metrics); + } + auto cache_input_stream = std::make_shared( + std::move(input_stream), cache); + return reader_builder->Build(cache_input_stream); + })); } std::vector> readers; for (auto& file_batch_reader : CollectAll(futures)) { diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp index 273752413..aa0bd2e4e 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp @@ -19,7 +19,10 @@ #include "paimon/common/reader/prefetch_file_batch_reader_impl.h" #include +#include +#include #include +#include #include #include "arrow/compute/api.h" @@ -130,6 +133,37 @@ class FailingFileSystem : public MockFileSystem { } }; +/// Blocks every Open() until `opens_needed` of them are in flight at once, so a caller that +/// opens its streams one at a time never reaches the barrier and fails instead of passing +/// silently. The wait is bounded so that failure is reported rather than hanging the test. +class BarrierFileSystem : public MockFileSystem { + public: + explicit BarrierFileSystem(uint32_t opens_needed) : opens_needed_(opens_needed) {} + + Result> Open(const std::string& path) const override { + const uint32_t arrived = ++inflight_opens_; + std::unique_lock lock(mutex_); + if (!condition_.wait_for(lock, kBarrierTimeout, + [this] { return inflight_opens_ >= opens_needed_; })) { + return Status::IOError(fmt::format( + "only {} of {} opens were in flight at once, so they are not concurrent", arrived, + opens_needed_)); + } + condition_.notify_all(); + return std::make_unique(); + } + + private: + // Generous so a loaded machine does not turn this into a flaky test; a regression makes the + // opens serial, which no wait ever satisfies. + static constexpr std::chrono::seconds kBarrierTimeout{60}; + + const uint32_t opens_needed_; + mutable std::atomic inflight_opens_{0}; + mutable std::mutex mutex_; + mutable std::condition_variable condition_; +}; + class IoReadingMockFileBatchReader : public MockFileBatchReader { public: IoReadingMockFileBatchReader(const std::shared_ptr& data, @@ -1377,4 +1411,31 @@ TEST_P(PrefetchFileBatchReaderImplTest, TestWarmupLevelDecoded) { ASSERT_GT(produced_batches, 0); } +// The cache's stream used to be opened before the readers' streams were dispatched, which +// serialized one remote round trip ahead of all of them. They are opened together now, and this +// pins that down: the barrier needs every one of them in flight at once, so opening the cache's +// stream first would leave the others waiting on a round trip that has not been started. +TEST_F(PrefetchFileBatchReaderImplTest, TestCreateOpensAllStreamsConcurrently) { + auto data_array = PrepareArray(101); + const int32_t batch_size = 10; + const int32_t prefetch_max_parallel_num = 3; + // One stream for the cache plus one per reader. + const uint32_t opens_needed = prefetch_max_parallel_num + 1; + auto barrier_fs = std::make_shared(opens_needed); + MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size); + // Unlike executor_ above, this one has a thread per open: with fewer threads than opens the + // barrier could not be reached however concurrent the dispatch is. + ASSERT_OK_AND_ASSIGN(std::shared_ptr executor, CreateDefaultExecutor(opens_needed)); + + ASSERT_OK_AND_ASSIGN( + auto reader, + PrefetchFileBatchReaderImpl::Create( + /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, barrier_fs, + prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2, + /*enable_adaptive_prefetch_strategy=*/false, executor, + /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/true, CacheConfig(), + /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_))); + ASSERT_NE(reader, nullptr); +} + } // namespace paimon::test diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index f15408433..fadf51781 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -18,8 +18,10 @@ #include "paimon/core/operation/abstract_split_read.h" +#include #include #include +#include #include #include #include @@ -33,6 +35,8 @@ #include "paimon/common/data/shredding/shredding_file_reader.h" #include "paimon/common/data/variant/variant_shredding_read_plan_factory.h" #include "paimon/common/data/variant/variant_type_utils.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/executor/reader_build_executor.h" #include "paimon/common/reader/data_file_reader_factory.h" #include "paimon/common/reader/delegating_prefetch_reader.h" #include "paimon/common/reader/late_materializing_reader_builder.h" @@ -93,23 +97,69 @@ Result>> AbstractSplitRead::CreateR std::vector> raw_file_readers; raw_file_readers.reserve(data_files.size()); + const uint32_t parallel_num = + static_cast(std::min(kReaderBuildMaxParallelNum, data_files.size())); + if (parallel_num <= 1) { + for (const auto& file : data_files) { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr file_reader, + CreateRawFileReader(partition, file, field_mapping_builder.get(), dv_factory, + row_ranges, data_file_path_factory, extra_format_options)); + if (file_reader) { + raw_file_readers.push_back(std::move(file_reader)); + } + } + return std::move(raw_file_readers); + } + + // Opening a file, reading its footer and building the reader is mostly waiting on remote + // I/O, so files are built concurrently. The tasks only reference locals of this frame, + // which stay alive because CollectAll below drains every future before returning. + std::shared_ptr build_executor = GetReaderBuildExecutor(); + std::vector>>> futures; + futures.reserve(data_files.size()); for (const auto& file : data_files) { - auto data_file_path = data_file_path_factory->ToPath(file); - PAIMON_ASSIGN_OR_RAISE(std::string data_file_identifier, file->FileFormat()); - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader_builder, - PrepareReaderBuilder(data_file_identifier, extra_format_options)); - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr file_reader, - CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder), - field_mapping_builder.get(), dv_factory, row_ranges, - data_file_path_factory)); + futures.push_back(Via( + build_executor.get(), [this, file, &partition, &field_mapping_builder, &dv_factory, + &row_ranges, &data_file_path_factory, &extra_format_options]() { + return CreateRawFileReader(partition, file, field_mapping_builder.get(), dv_factory, + row_ranges, data_file_path_factory, + extra_format_options); + })); + } + // Readers keep the file order of the split, and CollectAll preserves the submit order. + Status first_error; + for (auto& built : CollectAll(futures)) { + if (!built.ok()) { + if (first_error.ok()) { + first_error = built.status(); + } + continue; + } + std::unique_ptr file_reader = std::move(built).value(); if (file_reader) { raw_file_readers.push_back(std::move(file_reader)); } } + PAIMON_RETURN_NOT_OK(first_error); return std::move(raw_file_readers); } +Result> AbstractSplitRead::CreateRawFileReader( + const BinaryRow& partition, const std::shared_ptr& file, + const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, + const std::optional>& row_ranges, + const std::shared_ptr& data_file_path_factory, + const std::map& extra_format_options) const { + auto data_file_path = data_file_path_factory->ToPath(file); + PAIMON_ASSIGN_OR_RAISE(std::string data_file_identifier, file->FileFormat()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader_builder, + PrepareReaderBuilder(data_file_identifier, extra_format_options)); + return CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder), + field_mapping_builder, dv_factory, row_ranges, + data_file_path_factory); +} + bool AbstractSplitRead::NeedCompleteRowTrackingFields( bool row_tracking_enabled, const std::shared_ptr& read_schema) { if (row_tracking_enabled && diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h index 1e5f56441..a52bed66b 100644 --- a/src/paimon/core/operation/abstract_split_read.h +++ b/src/paimon/core/operation/abstract_split_read.h @@ -107,6 +107,14 @@ class AbstractSplitRead : public SplitRead { /// that the reader builder and the file reader cannot be built from different answers. DataFileReadOptions BuildDataFileReadOptions() const; + // return nullptr if data file is skipped by index or dv + Result> CreateRawFileReader( + const BinaryRow& partition, const std::shared_ptr& file, + const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, + const std::optional>& row_ranges, + const std::shared_ptr& data_file_path_factory, + const std::map& extra_format_options) const; + Result> PrepareReaderBuilder( const std::string& format_identifier, const std::map& extra_format_options) const; diff --git a/src/paimon/core/schema/schema_manager.h b/src/paimon/core/schema/schema_manager.h index 7480ec5da..da81bcb81 100644 --- a/src/paimon/core/schema/schema_manager.h +++ b/src/paimon/core/schema/schema_manager.h @@ -47,6 +47,7 @@ class SchemaManager { const std::string& branch); /// Read schema for schema id. Find schema in cache first. + /// Safe to call concurrently. Result> ReadSchema(int64_t schema_id) const; Result>> Latest() const; Result> CreateTable( diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 4b2f0f21e..284cd362f 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -2818,6 +2818,82 @@ 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 file_list_0; + std::vector 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))); + + // The concurrent build resolves each file against its own schema id, so it has to return + // exactly the rows the sequential schema-evolution read above returns. + std::vector read_fields = {DataField(0, arrow::field("key0", arrow::int32())), + DataField(1, arrow::field("key1", arrow::int32())), + DataField(6, arrow::field("k", arrow::int32())), + DataField(3, arrow::field("c", arrow::int32())), + DataField(7, arrow::field("d", arrow::int32())), + DataField(5, arrow::field("a", arrow::int32())), + DataField(8, arrow::field("e", arrow::int32()))}; + auto fields_with_row_kind = read_fields; + fields_with_row_kind.insert(fields_with_row_kind.begin(), SpecialFields::ValueKind()); + std::shared_ptr arrow_data_type = + DataField::ConvertDataFieldsToArrowStructType(fields_with_row_kind); + + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow_data_type, {R"([ + [0, 0, 1, 16, 13, null, 15, null], + [0, 0, 1, 26, 23, null, 25, null], + [0, 0, 1, 36, 33, null, 35, null], + [0, 0, 1, 66, 63, 517, 65, 618], + [0, 0, 1, 76, 73, 527, 75, 628], + [0, 0, 1, 86, 83, 537, 85, 638], + [0, 1, 1, 96, 93, 547, 95, 648], + [0, 1, 1, 106, 103, 557, 105, 658], + [0, 1, 1, 46, 43, null, 45, null], + [0, 1, 1, 56, 53, null, 55, null] + ])"}, + &expected_array); + ASSERT_TRUE(array_status.ok()); + ASSERT_TRUE(result_array->Equals(*expected_array)); +} + TEST_P(ReadInteTest, TestAppendReadWithSchemaEvolutionWithPredicateFilter) { std::vector read_fields = {DataField(5, arrow::field("a", arrow::int32())), DataField(6, arrow::field("k", arrow::int32())), diff --git a/test/inte/scan_and_read_inte_test.cpp b/test/inte/scan_and_read_inte_test.cpp index ad1345405..093d13852 100644 --- a/test/inte/scan_and_read_inte_test.cpp +++ b/test/inte/scan_and_read_inte_test.cpp @@ -604,6 +604,50 @@ 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 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))); + + // The parallel build has to return exactly the rows of the sequential snapshot-5 read above, + // in the very same order. + auto expected = std::make_shared( + arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ +[0, "Alice", 10, 1, 11.1], +[0, "Bob", 10, 0, 12.1], +[0, "Emily", 10, 0, 13.1], +[0, "Tony", 10, 0, 14.1], +[0, "Emily", 10, 0, 15.1], +[0, "Bob", 10, 0, 12.1], +[0, "Alex", 10, 0, 16.1], +[0, "David", 10, 0, 17.1], +[0, "Lily", 10, 0, 17.1], +[0, "Lucy", 20, 1, 14.1], +[0, "Paul", 20, 1, null] + ])") + .ValueOrDie()); + ASSERT_TRUE(expected); + ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); +} + TEST_P(ScanAndReadInteTest, TestWithAppendSnapshotWithStreamWithDefaultMode) { auto file_format = FileFormat(); std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09";