This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new bafe5573 perf(read): open data file streams and build readers
concurrently (#334)
bafe5573 is described below
commit bafe557384f3e48cf11ae41cb9a1bad58e77c275
Author: Yonghao Fang <[email protected]>
AuthorDate: Mon Sep 14 19:23:00 2026 +0800
perf(read): open data file streams and build readers concurrently (#334)
---
src/paimon/CMakeLists.txt | 1 +
.../common/executor/reader_build_executor.cpp | 43 ++++++++++++
src/paimon/common/executor/reader_build_executor.h | 47 +++++++++++++
.../reader/prefetch_file_batch_reader_impl.cpp | 69 +++++++++++++++-----
.../prefetch_file_batch_reader_impl_test.cpp | 61 +++++++++++++++++
src/paimon/core/operation/abstract_split_read.cpp | 68 ++++++++++++++++---
src/paimon/core/operation/abstract_split_read.h | 8 +++
src/paimon/core/schema/schema_manager.h | 1 +
test/inte/read_inte_test.cpp | 76 ++++++++++++++++++++++
test/inte/scan_and_read_inte_test.cpp | 44 +++++++++++++
10 files changed, 393 insertions(+), 25 deletions(-)
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 0a9e1029..afea66bd 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 00000000..685119fe
--- /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 <algorithm>
+#include <mutex>
+#include <thread>
+
+#include "paimon/result.h"
+
+namespace paimon {
+
+std::shared_ptr<Executor> GetReaderBuildExecutor() {
+ static std::shared_ptr<Executor> 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 00000000..7053512c
--- /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 <cstdint>
+#include <memory>
+
+#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<Executor> 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 da7ba458..2b2c7ec8 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<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
if (enable_io_metrics) {
io_metrics = std::make_shared<PrefetchIoMetricsState>();
}
+ // 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<std::future<Result<std::unique_ptr<InputStream>>>>
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<std::unique_ptr<InputStream>> {
+ 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<Result<std::unique_ptr<InputStream>>> opened_streams =
CollectAll(open_futures);
+ std::vector<std::unique_ptr<InputStream>> 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<ReadAheadCache> cache;
if (read_ahead_cache_enabled) {
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
- fs->Open(FileStatus(data_file_path,
data_file_size)));
+ std::shared_ptr<InputStream> input_stream =
std::move(streams[next_stream++]);
if (io_metrics) {
input_stream = std::make_shared<MetricsInputStream>(input_stream,
io_metrics);
}
@@ -246,22 +281,24 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
cache = std::make_shared<ReadAheadCache>(input_stream, cache_config,
static_cast<uint64_t>(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<std::future<Result<std::unique_ptr<FileBatchReader>>>> 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<std::unique_ptr<FileBatchReader>> {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream>
input_stream,
- fs->Open(FileStatus(data_file_path,
data_file_size)));
- if (io_metrics) {
- input_stream =
-
std::make_unique<MetricsInputStream>(std::move(input_stream), io_metrics);
- }
- auto cache_input_stream =
-
std::make_shared<CacheInputStream>(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<FileBatchReader>> {
+ std::unique_ptr<InputStream> input_stream =
+ std::move(streams[next_stream + i]);
+ if (io_metrics) {
+ input_stream =
std::make_unique<MetricsInputStream>(
+ std::move(input_stream), io_metrics);
+ }
+ auto cache_input_stream =
std::make_shared<CacheInputStream>(
+ std::move(input_stream), cache);
+ return
reader_builder->Build(cache_input_stream);
+ }));
}
std::vector<std::shared_ptr<PrefetchFileBatchReader>> 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 27375241..aa0bd2e4 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 <atomic>
+#include <chrono>
+#include <condition_variable>
#include <limits>
+#include <mutex>
#include <set>
#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<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ const uint32_t arrived = ++inflight_opens_;
+ std::unique_lock<std::mutex> 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<MockInputStream>();
+ }
+
+ 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<uint32_t> inflight_opens_{0};
+ mutable std::mutex mutex_;
+ mutable std::condition_variable condition_;
+};
+
class IoReadingMockFileBatchReader : public MockFileBatchReader {
public:
IoReadingMockFileBatchReader(const std::shared_ptr<arrow::Array>& 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<BarrierFileSystem>(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> 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 f1540843..fadf5178 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 <algorithm>
#include <cassert>
#include <cstddef>
+#include <future>
#include <map>
#include <set>
#include <utility>
@@ -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<std::vector<std::unique_ptr<FileBatchReader>>>
AbstractSplitRead::CreateR
std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers;
raw_file_readers.reserve(data_files.size());
+ const uint32_t parallel_num =
+ static_cast<uint32_t>(std::min<size_t>(kReaderBuildMaxParallelNum,
data_files.size()));
+ if (parallel_num <= 1) {
+ for (const auto& file : data_files) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileBatchReader> 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<Executor> build_executor = GetReaderBuildExecutor();
+ std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> 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<ReaderBuilder> reader_builder,
- PrepareReaderBuilder(data_file_identifier,
extra_format_options));
- PAIMON_ASSIGN_OR_RAISE(
- std::unique_ptr<FileBatchReader> 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<FileBatchReader> 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<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::CreateRawFileReader(
+ const BinaryRow& partition, const std::shared_ptr<DataFileMeta>& file,
+ const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory
dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& 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<ReaderBuilder> 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<arrow::Schema>&
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 1e5f5644..a52bed66 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<std::unique_ptr<FileBatchReader>> CreateRawFileReader(
+ const BinaryRow& partition, const std::shared_ptr<DataFileMeta>& file,
+ const FieldMappingBuilder* field_mapping_builder,
DeletionVector::Factory dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& extra_format_options) const;
+
Result<std::unique_ptr<ReaderBuilder>> PrepareReaderBuilder(
const std::string& format_identifier,
const std::map<std::string, std::string>& extra_format_options) const;
diff --git a/src/paimon/core/schema/schema_manager.h
b/src/paimon/core/schema/schema_manager.h
index 7480ec5d..da81bcb8 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<std::shared_ptr<TableSchema>> ReadSchema(int64_t schema_id) const;
Result<std::optional<std::shared_ptr<TableSchema>>> Latest() const;
Result<std::unique_ptr<TableSchema>> CreateTable(
diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp
index 4b2f0f21..284cd362 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<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)));
+
+ // 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<DataField> 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::DataType> arrow_data_type =
+ DataField::ConvertDataFieldsToArrowStructType(fields_with_row_kind);
+
+ std::shared_ptr<arrow::ChunkedArray> 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<DataField> 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 ad134540..093d1385 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<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)));
+
+ // 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::ChunkedArray>(
+ 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";
