lxy-9602 commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3802574393
##########
src/paimon/common/utils/read_ahead_cache_test.cpp:
##########
@@ -61,6 +64,25 @@ TestCacheEnv CreateTestFileAndCache(const std::string&
filename, const std::stri
return {path, cache, pool};
}
+// Assert that reading the range is a cache hit filling the destination with
+// the expected content.
+void AssertReadEquals(ReadAheadCache& cache, const ByteRange& range, const
std::string& expected) {
+ std::string dest(std::max<size_t>(range.length, 1), 'X');
Review Comment:
Could we move cache to the end and store it as a pointer?
##########
src/paimon/core/table/bucket_mode_test.cpp:
##########
@@ -50,13 +50,23 @@ TEST(BucketModeTest, TestResolveBucketMode) {
std::shared_ptr<TableSchema> append_schema =
CreateTableSchema(/*primary_keys=*/{});
std::shared_ptr<TableSchema> pk_schema =
CreateTableSchema(/*primary_keys=*/{"f0"});
+ // Postpone bucket only applies to primary key tables.
EXPECT_EQ(BucketMode::POSTPONE_MODE,
+ ResolveBucketMode(BucketModeDefine::POSTPONE_BUCKET, pk_schema));
+ EXPECT_EQ(BucketMode::HASH_FIXED,
ResolveBucketMode(BucketModeDefine::POSTPONE_BUCKET,
append_schema));
+
EXPECT_EQ(BucketMode::BUCKET_UNAWARE, ResolveBucketMode(-1,
append_schema));
EXPECT_EQ(BucketMode::HASH_DYNAMIC, ResolveBucketMode(-1, pk_schema));
- EXPECT_EQ(BucketMode::BUCKET_UNAWARE,
+
+ // UNAWARE_BUCKET is a bucket id, not a bucket number, so it does not mean
unaware mode here.
+ EXPECT_EQ(BucketMode::HASH_FIXED,
ResolveBucketMode(BucketModeDefine::UNAWARE_BUCKET, pk_schema));
+ EXPECT_EQ(BucketMode::HASH_FIXED,
+ ResolveBucketMode(BucketModeDefine::UNAWARE_BUCKET,
append_schema));
+
EXPECT_EQ(BucketMode::HASH_FIXED, ResolveBucketMode(4, append_schema));
+ EXPECT_EQ(BucketMode::HASH_FIXED, ResolveBucketMode(4, pk_schema));
}
Review Comment:
Do we really need a dedicated test for this error case? It feels quite
ambiguous.
##########
src/paimon/common/utils/read_ahead_cache_test.cpp:
##########
@@ -139,21 +165,284 @@ TEST(TestReadAheadCache, TestCacheEviction) {
auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}, {16, 5}});
auto& cache = *env.cache;
- ByteSlice slice;
- ASSERT_OK_AND_ASSIGN(slice, cache.Read({0, 5}));
- ASSERT_TRUE(slice.buffer);
- std::string first_read(slice.buffer->data() + slice.offset, slice.length);
- ASSERT_EQ(first_read, "abcde");
-
- // Reading another range should evict the first one due to buffer size
limit
- ASSERT_OK_AND_ASSIGN(slice, cache.Read({8, 5}));
- ASSERT_TRUE(slice.buffer);
- std::string second_read(slice.buffer->data() + slice.offset, slice.length);
- ASSERT_EQ(second_read, "ijklm");
-
- // The first range should now be a cache miss (buffer is nullptr)
- auto miss = cache.Read({0, 5});
- ASSERT_FALSE(miss.value().buffer);
+ AssertReadEquals(cache, {0, 5}, "abcde");
+
+ // Reading another range should evict the first one due to buffer size
limit.
+ AssertReadEquals(cache, {8, 5}, "ijklm");
+
+ // The first range should now be a cache miss.
+ AssertReadMiss(cache, {0, 5});
+}
+
+// Test that Read() hits and misses are recorded in the cache metrics.
+TEST(TestReadAheadCache, TestMetrics) {
+ CacheConfig config(/*buffer_size_limit=*/256 * 1024 * 1024,
/*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}});
+ auto& cache = *env.cache;
+
+ AssertReadEquals(cache, {0, 5}, "abcde");
+ // Out of any cached range: a miss.
+ AssertReadMiss(cache, {20, 3});
+
+ std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&metrics);
+ // Both Read() requests are counted, regardless of hit or miss.
+ ASSERT_OK_AND_ASSIGN(uint64_t read_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
+ ASSERT_EQ(read_bytes, 8u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hits,
metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
+ ASSERT_EQ(hit_bytes, 5u);
+ ASSERT_OK_AND_ASSIGN(uint64_t misses,
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t miss_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES));
+ ASSERT_EQ(miss_bytes, 3u);
+ // The hit prefetches both pending ranges in one window: two IO requests
+ // for 10 bytes in total; the miss issues no further fetch.
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t io_bytes,
metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 10u);
+}
+
+// Test that ReleaseBuffers() drops the cached data but keeps the hit/miss
counters
+// readable, while Reset() zeroes them as well.
+TEST(TestReadAheadCache, TestReleaseBuffersKeepsMetrics) {
+ CacheConfig config(/*buffer_size_limit=*/256 * 1024 * 1024,
/*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+ auto& cache = *env.cache;
+
+ AssertReadEquals(cache, {0, 5}, "abcde");
+
+ cache.ReleaseBuffers();
+
+ // The previously cached range is gone: the read now misses.
+ AssertReadMiss(cache, {0, 5});
+
+ std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&metrics);
+ // The read counters survive ReleaseBuffers() as well.
+ ASSERT_OK_AND_ASSIGN(uint64_t read_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
+ ASSERT_EQ(read_bytes, 10u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hits,
metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
+ ASSERT_EQ(hit_bytes, 5u);
+ ASSERT_OK_AND_ASSIGN(uint64_t misses,
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 1u);
+ // The io counters survive ReleaseBuffers() as well.
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t io_bytes,
metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 5u);
+
+ // Reset() clears the counters too.
+ cache.Reset();
+ std::shared_ptr<Metrics> reset_metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&reset_metrics);
+ ASSERT_OK_AND_ASSIGN(read_count,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 0u);
+ ASSERT_OK_AND_ASSIGN(hits,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 0u);
+ ASSERT_OK_AND_ASSIGN(misses,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 0u);
+ ASSERT_OK_AND_ASSIGN(io_count,
reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 0u);
+ ASSERT_OK_AND_ASSIGN(io_bytes,
reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 0u);
+}
+
+// Test that a failed prefetch surfaces as an error Status from Read(), not as
+// a miss: the entry exists from the moment its fetch is submitted and its
+// future carries the IO error.
+TEST(TestReadAheadCache, TestPrefetchIOErrorPropagation) {
+ CacheConfig config(/*buffer_size_limit=*/1024, /*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto io_hook = paimon::IOHook::GetInstance();
+
+ // Single entry: the prefetch is the first IO after the hook is armed.
+ {
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0,
10}});
+ paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR);
+ std::string dest(5, 'X');
+ Result<bool> result = env.cache->Read({0, 5}, dest.data());
+ ASSERT_FALSE(result.ok());
+ EXPECT_NE(std::string::npos,
+ result.status().ToString().find("io hook triggered io error
at position"));
Review Comment:
ASSERT_NOK_WITH_MSG
##########
src/paimon/common/utils/read_ahead_cache_test.cpp:
##########
@@ -139,21 +165,284 @@ TEST(TestReadAheadCache, TestCacheEviction) {
auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}, {16, 5}});
auto& cache = *env.cache;
- ByteSlice slice;
- ASSERT_OK_AND_ASSIGN(slice, cache.Read({0, 5}));
- ASSERT_TRUE(slice.buffer);
- std::string first_read(slice.buffer->data() + slice.offset, slice.length);
- ASSERT_EQ(first_read, "abcde");
-
- // Reading another range should evict the first one due to buffer size
limit
- ASSERT_OK_AND_ASSIGN(slice, cache.Read({8, 5}));
- ASSERT_TRUE(slice.buffer);
- std::string second_read(slice.buffer->data() + slice.offset, slice.length);
- ASSERT_EQ(second_read, "ijklm");
-
- // The first range should now be a cache miss (buffer is nullptr)
- auto miss = cache.Read({0, 5});
- ASSERT_FALSE(miss.value().buffer);
+ AssertReadEquals(cache, {0, 5}, "abcde");
+
+ // Reading another range should evict the first one due to buffer size
limit.
+ AssertReadEquals(cache, {8, 5}, "ijklm");
+
+ // The first range should now be a cache miss.
+ AssertReadMiss(cache, {0, 5});
+}
+
+// Test that Read() hits and misses are recorded in the cache metrics.
+TEST(TestReadAheadCache, TestMetrics) {
+ CacheConfig config(/*buffer_size_limit=*/256 * 1024 * 1024,
/*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}});
+ auto& cache = *env.cache;
+
+ AssertReadEquals(cache, {0, 5}, "abcde");
+ // Out of any cached range: a miss.
+ AssertReadMiss(cache, {20, 3});
+
+ std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&metrics);
+ // Both Read() requests are counted, regardless of hit or miss.
+ ASSERT_OK_AND_ASSIGN(uint64_t read_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
+ ASSERT_EQ(read_bytes, 8u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hits,
metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
+ ASSERT_EQ(hit_bytes, 5u);
+ ASSERT_OK_AND_ASSIGN(uint64_t misses,
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t miss_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES));
+ ASSERT_EQ(miss_bytes, 3u);
+ // The hit prefetches both pending ranges in one window: two IO requests
+ // for 10 bytes in total; the miss issues no further fetch.
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t io_bytes,
metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 10u);
+}
+
+// Test that ReleaseBuffers() drops the cached data but keeps the hit/miss
counters
+// readable, while Reset() zeroes them as well.
+TEST(TestReadAheadCache, TestReleaseBuffersKeepsMetrics) {
+ CacheConfig config(/*buffer_size_limit=*/256 * 1024 * 1024,
/*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+ auto& cache = *env.cache;
+
+ AssertReadEquals(cache, {0, 5}, "abcde");
+
+ cache.ReleaseBuffers();
+
+ // The previously cached range is gone: the read now misses.
+ AssertReadMiss(cache, {0, 5});
+
+ std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&metrics);
+ // The read counters survive ReleaseBuffers() as well.
+ ASSERT_OK_AND_ASSIGN(uint64_t read_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 2u);
+ ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
+ ASSERT_EQ(read_bytes, 10u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hits,
metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
+
metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
+ ASSERT_EQ(hit_bytes, 5u);
+ ASSERT_OK_AND_ASSIGN(uint64_t misses,
metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 1u);
+ // The io counters survive ReleaseBuffers() as well.
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 1u);
+ ASSERT_OK_AND_ASSIGN(uint64_t io_bytes,
metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 5u);
+
+ // Reset() clears the counters too.
+ cache.Reset();
+ std::shared_ptr<Metrics> reset_metrics = std::make_shared<MetricsImpl>();
+ cache.CollectMetrics(&reset_metrics);
+ ASSERT_OK_AND_ASSIGN(read_count,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
+ ASSERT_EQ(read_count, 0u);
+ ASSERT_OK_AND_ASSIGN(hits,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
+ ASSERT_EQ(hits, 0u);
+ ASSERT_OK_AND_ASSIGN(misses,
reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
+ ASSERT_EQ(misses, 0u);
+ ASSERT_OK_AND_ASSIGN(io_count,
reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 0u);
+ ASSERT_OK_AND_ASSIGN(io_bytes,
reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
+ ASSERT_EQ(io_bytes, 0u);
+}
+
+// Test that a failed prefetch surfaces as an error Status from Read(), not as
+// a miss: the entry exists from the moment its fetch is submitted and its
+// future carries the IO error.
+TEST(TestReadAheadCache, TestPrefetchIOErrorPropagation) {
+ CacheConfig config(/*buffer_size_limit=*/1024, /*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto io_hook = paimon::IOHook::GetInstance();
+
+ // Single entry: the prefetch is the first IO after the hook is armed.
+ {
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0,
10}});
+ paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR);
+ std::string dest(5, 'X');
+ Result<bool> result = env.cache->Read({0, 5}, dest.data());
+ ASSERT_FALSE(result.ok());
+ EXPECT_NE(std::string::npos,
+ result.status().ToString().find("io hook triggered io error
at position"));
+ }
+
+ // Several adjacent entries: the error of any segment aborts the read.
+ {
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0,
25}});
+ paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ io_hook->Reset(1, paimon::IOHook::Mode::RETURN_ERROR);
+ std::string dest(20, 'X');
+ Result<bool> result = env.cache->Read({0, 20}, dest.data());
+ ASSERT_FALSE(result.ok());
+ EXPECT_NE(std::string::npos,
+ result.status().ToString().find("io hook triggered io error
at position"));
+ }
+}
+
+// Test that Warmup() fetches the pending ranges up front so the first Read()
+// issues no further IO, while without Warmup() the first Read() triggers the
+// prefetch itself.
+TEST(TestReadAheadCache, TestWarmupPrefetchesBeforeFirstRead) {
+ CacheConfig config(/*buffer_size_limit=*/1024, /*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env1 = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}});
+ env1.cache->Warmup();
+ auto env2 = CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+
+ auto io_hook = paimon::IOHook::GetInstance();
+ paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ // Any new IO fails: the warmed-up reads must be served without fetching.
+ io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR);
+
+ AssertReadEquals(*env1.cache, {0, 5}, "abcde");
+ AssertReadEquals(*env1.cache, {8, 5}, "ijklm");
+
+ // Without Warmup() the first Read() starts the prefetch and sees the
error.
+ std::string dest(5, 'X');
+ ASSERT_FALSE(env2.cache->Read({0, 5}, dest.data()).ok());
Review Comment:
ASSERT_NOK
##########
src/paimon/format/parquet/file_reader_wrapper.cpp:
##########
@@ -411,6 +435,22 @@ std::vector<::arrow::io::ReadRange>
FileReaderWrapper::CollectPreBufferRanges(
return ranges;
}
+Result<std::vector<std::pair<uint64_t, uint64_t>>>
FileReaderWrapper::GetPreBufferRanges() {
+ try {
+ std::vector<::arrow::io::ReadRange> ranges =
+ DoCollectPreBufferRanges(target_column_indices_,
+ /*skip_read_range_excluded=*/false,
/*start_idx=*/0);
+ std::vector<std::pair<uint64_t, uint64_t>> pre_buffer_ranges;
+ pre_buffer_ranges.reserve(ranges.size());
+ for (const auto& range : ranges) {
+ pre_buffer_ranges.emplace_back(static_cast<uint64_t>(range.offset),
+
static_cast<uint64_t>(range.length));
+ }
+ return pre_buffer_ranges;
+ }
Review Comment:
Where could an exception be thrown here? `DoCollectPreBufferRanges`? If so,
shouldn’t this function return a bad status instead?
##########
src/paimon/format/parquet/file_reader_wrapper.cpp:
##########
@@ -376,16 +393,23 @@ Status FileReaderWrapper::PrepareForReadingLazy(
target_row_groups_ = target_row_groups;
target_column_indices_ = column_indices;
reader_initialized_ = false;
+ pending_start_idx_.reset();
return Status::OK();
}
std::vector<::arrow::io::ReadRange> FileReaderWrapper::CollectPreBufferRanges(
- const std::vector<int32_t>& column_indices) {
+ const std::vector<int32_t>& column_indices, uint64_t start_idx) {
+ return DoCollectPreBufferRanges(column_indices,
/*skip_read_range_excluded=*/true, start_idx);
+}
+
+std::vector<::arrow::io::ReadRange>
FileReaderWrapper::DoCollectPreBufferRanges(
+ const std::vector<int32_t>& column_indices, bool skip_read_range_excluded,
uint64_t start_idx) {
std::vector<::arrow::io::ReadRange> ranges;
auto file_metadata = file_reader_->parquet_reader()->metadata();
- for (const auto& trg : target_row_groups_) {
- if (trg.IsExcludedByReadRange()) continue;
+ for (uint64_t idx = start_idx; idx < target_row_groups_.size(); idx++) {
+ const auto& trg = target_row_groups_[idx];
+ if (skip_read_range_excluded && trg.IsExcludedByReadRange()) continue;
Review Comment:
{continue;}
--
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]
