diff --git a/docs/source/user_guide/read.rst b/docs/source/user_guide/read.rst index adebaecb..dbf48c63 100644 --- a/docs/source/user_guide/read.rst +++ b/docs/source/user_guide/read.rst @@ -40,6 +40,24 @@ the two layers can evolve their functionality and optimize code relatively indep cross-language task scheduling and interaction (e.g., Java and C++), substantially reducing engineering maintenance costs across the two language ecosystems. +Late Materialization +-------------------- + +In the raw-file read path, late materialization can reduce payload-column decoding when a +predicate selects only a small number of rows. The reader first loads the columns required by the +predicate (the *probe* columns), evaluates the predicate, and then reads the remaining projected +columns (the *payload* columns) only for matching row ranges. + +The optimization is disabled by default. Enable it with +``read.late-materialization.enabled=true``. It is applied only when predicate filtering is enabled, +and the probe and payload projections are both non-empty. The selected file-local row IDs are +pushed directly to each file reader, so row tracking and global row IDs are not required. +Otherwise, the reader uses the normal single-pass path. + +``read.late-materialization.max-match-rows`` limits the number of matching rows accumulated for a +split and defaults to ``1024``. If the limit is exceeded, the reader falls back to the normal path. +The probe work already completed before the fallback is retained in the reader metrics. + Schema Evolution ----------------------- diff --git a/include/paimon/defs.h b/include/paimon/defs.h index cf01a1ba..7f8c072a 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -202,6 +202,15 @@ struct PAIMON_EXPORT Options { /// The default value is 1024. static const char READ_BATCH_SIZE[]; + /// "read.late-materialization.enabled" - Whether to read predicate columns first and then + /// fetch payload columns only for matching rows. Default value is false. + static const char READ_LATE_MATERIALIZATION_ENABLED[]; + + /// "read.late-materialization.max-match-rows" - Maximum number of matching rows per split + /// allowed for late materialization. If the predicate matches more rows, the reader falls back + /// to the normal read path. Default value is 1024. + static const char READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS[]; + /// "write.batch-size" - Write batch size for any file format if it supports. /// The default value is 1024. static const char WRITE_BATCH_SIZE[]; diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 514c1354..b8aa4588 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -133,6 +133,7 @@ set(PAIMON_COMMON_SRCS common/predicate/starts_with.cpp common/reader/batch_reader.cpp common/reader/concat_batch_reader.cpp + common/reader/late_materialization_batch_reader.cpp common/reader/predicate_batch_reader.cpp common/reader/prefetch_file_batch_reader_impl.cpp common/reader/reader_utils.cpp @@ -584,6 +585,7 @@ if(PAIMON_BUILD_TESTS) common/predicate/predicate_utils_test.cpp common/predicate/predicate_validator_test.cpp common/reader/concat_batch_reader_test.cpp + common/reader/late_materialization_batch_reader_test.cpp common/reader/predicate_batch_reader_test.cpp common/reader/prefetch_file_batch_reader_impl_test.cpp common/reader/reader_utils_test.cpp diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 53f18c3f..e2785024 100644 --- a/src/paimon/common/defs.cpp +++ b/src/paimon/common/defs.cpp @@ -60,6 +60,9 @@ const char Options::SCAN_MODE[] = "scan.mode"; const char Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[] = "scan.manifest-entry-cache.max-snapshots"; const char Options::READ_BATCH_SIZE[] = "read.batch-size"; +const char Options::READ_LATE_MATERIALIZATION_ENABLED[] = "read.late-materialization.enabled"; +const char Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS[] = + "read.late-materialization.max-match-rows"; const char Options::WRITE_BATCH_SIZE[] = "write.batch-size"; const char Options::WRITE_BUFFER_SIZE[] = "write-buffer-size"; const char Options::WRITE_BUFFER_SPILLABLE[] = "write-buffer-spillable"; diff --git a/src/paimon/common/reader/concat_batch_reader.cpp b/src/paimon/common/reader/concat_batch_reader.cpp index c36886c5..5bd1f2d5 100644 --- a/src/paimon/common/reader/concat_batch_reader.cpp +++ b/src/paimon/common/reader/concat_batch_reader.cpp @@ -30,8 +30,12 @@ namespace paimon { class MemoryPool; ConcatBatchReader::ConcatBatchReader(std::vector>&& readers, - const std::shared_ptr& pool) - : arrow_pool_(GetArrowPool(pool)), readers_(std::move(readers)), current_(0) {} + const std::shared_ptr& pool, + const std::shared_ptr& completed_metrics) + : arrow_pool_(GetArrowPool(pool)), + readers_(std::move(readers)), + completed_metrics_(completed_metrics), + current_(0) {} Result ConcatBatchReader::NextBatch() { PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, @@ -46,7 +50,9 @@ void ConcatBatchReader::Close() { } std::shared_ptr ConcatBatchReader::GetReaderMetrics() const { - return MetricsImpl::CollectReadMetrics(readers_); + std::shared_ptr metrics = MetricsImpl::CollectReadMetrics(readers_); + metrics->Merge(completed_metrics_); + return metrics; } Result ConcatBatchReader::NextBatchWithBitmap() { diff --git a/src/paimon/common/reader/concat_batch_reader.h b/src/paimon/common/reader/concat_batch_reader.h index 7a1b6a19..d149d8e1 100644 --- a/src/paimon/common/reader/concat_batch_reader.h +++ b/src/paimon/common/reader/concat_batch_reader.h @@ -36,7 +36,8 @@ class MemoryPool; class ConcatBatchReader : public BatchReader { public: ConcatBatchReader(std::vector>&& readers, - const std::shared_ptr& pool); + const std::shared_ptr& pool, + const std::shared_ptr& completed_metrics = nullptr); Result NextBatch() override; Result NextBatchWithBitmap() override; @@ -46,6 +47,7 @@ class ConcatBatchReader : public BatchReader { private: std::unique_ptr arrow_pool_; std::vector> readers_; + std::shared_ptr completed_metrics_; size_t current_; }; } // namespace paimon diff --git a/src/paimon/common/reader/concat_batch_reader_test.cpp b/src/paimon/common/reader/concat_batch_reader_test.cpp index 37a415de..18a1d268 100644 --- a/src/paimon/common/reader/concat_batch_reader_test.cpp +++ b/src/paimon/common/reader/concat_batch_reader_test.cpp @@ -29,6 +29,7 @@ #include "arrow/array/array_nested.h" #include "arrow/ipc/json_simple.h" #include "gtest/gtest.h" +#include "paimon/common/metrics/metrics_impl.h" #include "paimon/memory/memory_pool.h" #include "paimon/status.h" #include "paimon/testing/mock/mock_file_batch_reader.h" @@ -169,4 +170,24 @@ TEST_F(ConcatBatchReaderTest, TestSimpleWithBitmap) { } } +TEST_F(ConcatBatchReaderTest, TestMergeCompletedReaderMetrics) { + std::shared_ptr data = + arrow::StructArray::Make( + {arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[1, 2, 3]").ValueOrDie()}, + {arrow::field("f1", arrow::int32())}) + .ValueOrDie(); + std::vector> readers; + readers.push_back( + std::make_unique(data, data->type(), /*read_batch_size=*/2)); + + std::shared_ptr completed_metrics = std::make_shared(); + completed_metrics->SetCounter("mock.number.of.rows", 4); + std::unique_ptr concat_reader = std::make_unique( + std::move(readers), GetDefaultPool(), completed_metrics); + + ASSERT_OK_AND_ASSIGN(uint64_t row_count, + concat_reader->GetReaderMetrics()->GetCounter("mock.number.of.rows")); + ASSERT_EQ(7, row_count); +} + } // namespace paimon::test diff --git a/src/paimon/common/reader/late_materialization_batch_reader.cpp b/src/paimon/common/reader/late_materialization_batch_reader.cpp new file mode 100644 index 00000000..169226d3 --- /dev/null +++ b/src/paimon/common/reader/late_materialization_batch_reader.cpp @@ -0,0 +1,198 @@ +/* + * 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/reader/late_materialization_batch_reader.h" + +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "arrow/array/array_base.h" +#include "arrow/array/array_nested.h" +#include "arrow/c/bridge.h" +#include "arrow/util/checked_cast.h" +#include "fmt/format.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/reader/reader_utils.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/status.h" + +namespace paimon { + +Result> LateMaterializationBatchReader::Create( + const std::shared_ptr& read_schema, + const std::shared_ptr& probe_schema, + std::shared_ptr probe_data, + const std::shared_ptr& payload_schema, + std::unique_ptr&& payload_reader, int32_t read_batch_size, + const std::shared_ptr& pool, std::unique_ptr arrow_pool) { + if (!read_schema || !probe_schema || !payload_schema || !probe_data) { + return Status::Invalid("late materialization reader requires non-null schemas and data"); + } + if (read_batch_size <= 0) { + return Status::Invalid("late materialization read batch size should be positive"); + } + + std::vector field_sources; + field_sources.reserve(read_schema->num_fields()); + bool needs_payload = false; + for (const std::shared_ptr& read_field : read_schema->fields()) { + int32_t probe_idx = probe_schema->GetFieldIndex(read_field->name()); + if (probe_idx >= 0) { + field_sources.push_back({Source::PROBE, probe_idx}); + continue; + } + int32_t payload_idx = payload_schema->GetFieldIndex(read_field->name()); + if (payload_idx >= 0) { + field_sources.push_back({Source::PAYLOAD, payload_idx}); + needs_payload = true; + continue; + } + return Status::Invalid(fmt::format( + "field {} is missing from both probe and payload schemas", read_field->name())); + } + if (needs_payload && !payload_reader) { + return Status::Invalid( + "late materialization reader requires a payload reader for payload fields"); + } + + return std::unique_ptr(new LateMaterializationBatchReader( + read_schema, std::move(probe_data), std::move(payload_reader), std::move(field_sources), + read_batch_size, pool, std::move(arrow_pool))); +} + +LateMaterializationBatchReader::LateMaterializationBatchReader( + const std::shared_ptr& read_schema, + std::shared_ptr probe_data, std::unique_ptr&& payload_reader, + std::vector&& field_sources, int32_t read_batch_size, + const std::shared_ptr& pool, std::unique_ptr arrow_pool) + : arrow_pool_(arrow_pool ? std::move(arrow_pool) : GetArrowPool(pool)), + read_schema_(read_schema), + probe_data_(std::move(probe_data)), + payload_reader_(std::move(payload_reader)), + field_sources_(std::move(field_sources)), + read_batch_size_(read_batch_size) {} + +Result LateMaterializationBatchReader::NextBatch() { + PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap batch_with_bitmap, NextBatchWithBitmap()); + return ReaderUtils::ApplyBitmapToReadBatch(std::move(batch_with_bitmap), arrow_pool_.get()); +} + +Result LateMaterializationBatchReader::NextBatchWithBitmap() { + if (payload_reader_) { + PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap payload_batch_with_bitmap, + payload_reader_->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(payload_batch_with_bitmap)) { + if (probe_offset_ != probe_data_->length()) { + return Status::Invalid(fmt::format( + "late materialization payload ended at {}, but probe row count is {}", + probe_offset_, probe_data_->length())); + } + return BatchReader::MakeEofBatchWithBitmap(); + } + + ReadBatchWithBitmap moved_payload_batch = std::move(payload_batch_with_bitmap); + ReadBatch& payload_batch = moved_payload_batch.first; + RoaringBitmap32& payload_bitmap = moved_payload_batch.second; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr payload_array, + arrow::ImportArray(payload_batch.first.get(), payload_batch.second.get())); + if (payload_bitmap.Cardinality() != payload_array->length()) { + PAIMON_ASSIGN_OR_RAISE( + arrow::ArrayVector filtered_payload, + ReaderUtils::GenerateFilteredArrayVector(payload_array, payload_bitmap)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + payload_array, arrow::Concatenate(filtered_payload, arrow_pool_.get())); + } + std::shared_ptr payload_struct = + arrow::internal::checked_pointer_cast(payload_array); + return MakeBatch(payload_struct); + } + + if (probe_offset_ >= probe_data_->length()) { + return BatchReader::MakeEofBatchWithBitmap(); + } + return MakeBatch(/*payload_data=*/nullptr); +} + +Result LateMaterializationBatchReader::MakeBatch( + const std::shared_ptr& payload_data) { + int64_t length = + payload_data ? payload_data->length() + : std::min(read_batch_size_, probe_data_->length() - probe_offset_); + if (probe_offset_ + length > probe_data_->length()) { + return Status::Invalid(fmt::format( + "late materialization payload row count exceeds probe row count: offset {}, length {}, " + "probe {}", + probe_offset_, length, probe_data_->length())); + } + + std::shared_ptr probe_slice = + arrow::internal::checked_pointer_cast( + probe_data_->Slice(probe_offset_, length)); + arrow::ArrayVector arrays; + arrays.reserve(field_sources_.size()); + for (const FieldSource& field_source : field_sources_) { + if (field_source.source == Source::PROBE) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr normalized_probe_field, + arrow::Concatenate({probe_slice->field(field_source.index)}, arrow_pool_.get())); + arrays.push_back(std::move(normalized_probe_field)); + } else { + if (!payload_data) { + return Status::Invalid( + "late materialization payload field requested without " + "payload reader"); + } + arrays.push_back(payload_data->field(field_source.index)); + } + } + + std::shared_ptr struct_array = std::make_shared( + arrow::struct_(read_schema_->fields()), length, arrays); + std::unique_ptr c_array = std::make_unique(); + std::unique_ptr c_schema = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportArray(*struct_array, c_array.get(), c_schema.get())); + + RoaringBitmap32 bitmap; + bitmap.AddRange(0, static_cast(length)); + probe_offset_ += length; + return std::make_pair(std::make_pair(std::move(c_array), std::move(c_schema)), + std::move(bitmap)); +} + +std::shared_ptr LateMaterializationBatchReader::GetReaderMetrics() const { + if (payload_reader_) { + return payload_reader_->GetReaderMetrics(); + } + return std::make_shared(); +} + +void LateMaterializationBatchReader::Close() { + if (payload_reader_) { + payload_reader_->Close(); + } +} + +} // namespace paimon diff --git a/src/paimon/common/reader/late_materialization_batch_reader.h b/src/paimon/common/reader/late_materialization_batch_reader.h new file mode 100644 index 00000000..8d7260d1 --- /dev/null +++ b/src/paimon/common/reader/late_materialization_batch_reader.h @@ -0,0 +1,80 @@ +/* + * 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 + +#include "arrow/memory_pool.h" +#include "arrow/type_fwd.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/result.h" + +namespace paimon { +class MemoryPool; + +/// Combines predicate (probe) columns with selectively read payload columns. +/// +/// Probe rows and payload rows must have a one-to-one positional correspondence. The reader +/// validates this invariant while consuming the payload reader. +class LateMaterializationBatchReader : public BatchReader { + public: + static Result> Create( + const std::shared_ptr& read_schema, + const std::shared_ptr& probe_schema, + std::shared_ptr probe_data, + const std::shared_ptr& payload_schema, + std::unique_ptr&& payload_reader, int32_t read_batch_size, + const std::shared_ptr& pool, + std::unique_ptr arrow_pool = nullptr); + + Result NextBatch() override; + Result NextBatchWithBitmap() override; + std::shared_ptr GetReaderMetrics() const override; + void Close() override; + + private: + enum class Source { PROBE, PAYLOAD }; + + struct FieldSource { + Source source; + int32_t index; + }; + + LateMaterializationBatchReader(const std::shared_ptr& read_schema, + std::shared_ptr probe_data, + std::unique_ptr&& payload_reader, + std::vector&& field_sources, + int32_t read_batch_size, const std::shared_ptr& pool, + std::unique_ptr arrow_pool); + + Result MakeBatch(const std::shared_ptr& payload_data); + + std::unique_ptr arrow_pool_; + std::shared_ptr read_schema_; + std::shared_ptr probe_data_; + std::unique_ptr payload_reader_; + std::vector field_sources_; + int32_t read_batch_size_; + int64_t probe_offset_ = 0; +}; + +} // namespace paimon diff --git a/src/paimon/common/reader/late_materialization_batch_reader_test.cpp b/src/paimon/common/reader/late_materialization_batch_reader_test.cpp new file mode 100644 index 00000000..530b27f4 --- /dev/null +++ b/src/paimon/common/reader/late_materialization_batch_reader_test.cpp @@ -0,0 +1,249 @@ +/* + * 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/reader/late_materialization_batch_reader.h" + +#include +#include +#include +#include + +#include "arrow/api.h" +#include "arrow/array/array_nested.h" +#include "arrow/c/bridge.h" +#include "arrow/ipc/json_simple.h" +#include "gtest/gtest.h" +#include "paimon/common/reader/reader_utils.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/testing/mock/mock_file_batch_reader.h" +#include "paimon/testing/utils/read_result_collector.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { + +std::shared_ptr StructFromJson(const std::shared_ptr& schema, + const std::string& json) { + return std::static_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()), json) + .ValueOrDie()); +} + +Result> Collect(BatchReader* reader) { + return ReadResultCollector::CollectResult(reader); +} + +void AssertZeroOffsets(const ArrowArray* array) { + ASSERT_NE(nullptr, array); + ASSERT_EQ(0, array->offset); + for (int64_t i = 0; i < array->n_children; ++i) { + AssertZeroOffsets(array->children[i]); + } +} + +} // namespace + +class LateMaterializationBatchReaderTest : public ::testing::Test { + public: + void SetUp() override { + read_schema_ = + arrow::schema({arrow::field("k", arrow::int64()), arrow::field("v1", arrow::utf8()), + arrow::field("v2", arrow::boolean())}); + probe_schema_ = arrow::schema({arrow::field("k", arrow::int64())}); + payload_schema_ = arrow::schema( + {arrow::field("v1", arrow::utf8()), arrow::field("v2", arrow::boolean())}); + } + + protected: + std::shared_ptr read_schema_; + std::shared_ptr probe_schema_; + std::shared_ptr payload_schema_; +}; + +TEST_F(LateMaterializationBatchReaderTest, TestMergeProbeAndPayloadColumns) { + std::shared_ptr probe_data = + StructFromJson(probe_schema_, R"([[1], [3], [5]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, R"([["v1", true], ["v3", false], ["v5", true]])"); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), /*read_batch_size=*/2); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/2, GetDefaultPool())); + + std::shared_ptr expected = std::make_shared( + StructFromJson(read_schema_, R"([[1, "v1", true], [3, "v3", false], [5, "v5", true]])")); + ASSERT_OK_AND_ASSIGN(std::shared_ptr actual, Collect(reader.get())); + ASSERT_TRUE(actual->Equals(expected)); +} + +TEST_F(LateMaterializationBatchReaderTest, TestProbeOnlyRead) { + std::shared_ptr probe_data = + StructFromJson(probe_schema_, R"([[1], [3], [5]])"); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + probe_schema_, probe_schema_, probe_data, arrow::schema({}), + /*payload_reader=*/nullptr, /*read_batch_size=*/2, GetDefaultPool())); + + std::shared_ptr expected = + std::make_shared(probe_data); + ASSERT_OK_AND_ASSIGN(std::shared_ptr actual, Collect(reader.get())); + ASSERT_TRUE(actual->Equals(expected)); +} + +TEST_F(LateMaterializationBatchReaderTest, TestPayloadBitmapIsApplied) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[1], [5]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, + R"([["skip", false], ["v1", true], ["skip", false], + ["skip", false], ["v5", true]])"); + RoaringBitmap32 bitmap; + bitmap.Add(1); + bitmap.Add(4); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), bitmap, /*read_batch_size=*/5); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/2, GetDefaultPool())); + + std::shared_ptr expected = std::make_shared( + StructFromJson(read_schema_, R"([[1, "v1", true], [5, "v5", true]])")); + ASSERT_OK_AND_ASSIGN(std::shared_ptr actual, Collect(reader.get())); + ASSERT_TRUE(actual->Equals(expected)); +} + +TEST_F(LateMaterializationBatchReaderTest, TestImportedPayloadBatchRemainsValid) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[7]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, R"([["v7", true]])"); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), /*read_batch_size=*/1); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/1, GetDefaultPool())); + + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + reader->NextBatchWithBitmap()); + ASSERT_EQ(1, batch_with_bitmap.second.Cardinality()); + arrow::Result> import_result = arrow::ImportArray( + batch_with_bitmap.first.first.get(), batch_with_bitmap.first.second.get()); + ASSERT_TRUE(import_result.ok()) << import_result.status().ToString(); + std::shared_ptr array = import_result.ValueOrDie(); + std::shared_ptr struct_array = + std::static_pointer_cast(array); + ASSERT_EQ(1, struct_array->length()); + ASSERT_EQ(3, struct_array->num_fields()); + ASSERT_EQ(7, std::static_pointer_cast(struct_array->field(0))->Value(0)); + ASSERT_EQ("v7", + std::static_pointer_cast(struct_array->field(1))->GetString(0)); + ASSERT_TRUE(std::static_pointer_cast(struct_array->field(2))->Value(0)); +} + +TEST_F(LateMaterializationBatchReaderTest, TestSecondBatchOffsetsAreZero) { + std::shared_ptr probe_data = + StructFromJson(probe_schema_, R"([[1], [3], [5]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, R"([["v1", true], ["v3", false], ["v5", true]])"); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), /*read_batch_size=*/2); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/2, GetDefaultPool())); + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap first_batch, + reader->NextBatchWithBitmap()); + ReaderUtils::ReleaseReadBatch(std::move(first_batch.first)); + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap second_batch, + reader->NextBatchWithBitmap()); + AssertZeroOffsets(second_batch.first.first.get()); + ReaderUtils::ReleaseReadBatch(std::move(second_batch.first)); +} + +TEST_F(LateMaterializationBatchReaderTest, TestPayloadLongerThanProbeFails) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[1]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, R"([["v1", true], ["v2", false]])"); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), /*read_batch_size=*/2); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/2, GetDefaultPool())); + ASSERT_NOK_WITH_MSG(reader->NextBatch(), "payload row count exceeds probe row count"); +} + +TEST_F(LateMaterializationBatchReaderTest, TestPayloadShorterThanProbeFails) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[1], [2]])"); + std::shared_ptr payload_data = + StructFromJson(payload_schema_, R"([["v1", true]])"); + std::unique_ptr payload_reader = std::make_unique( + payload_data, arrow::struct_(payload_schema_->fields()), /*read_batch_size=*/1); + payload_reader->EnableRandomizeBatchSize(false); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + std::move(payload_reader), /*read_batch_size=*/2, GetDefaultPool())); + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch()); + ReaderUtils::ReleaseReadBatch(std::move(batch)); + ASSERT_NOK_WITH_MSG(reader->NextBatch(), "payload ended at 1, but probe row count is 2"); +} + +TEST_F(LateMaterializationBatchReaderTest, TestMissingPayloadReaderFailsAtCreation) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[1]])"); + + ASSERT_NOK_WITH_MSG(LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, payload_schema_, + /*payload_reader=*/nullptr, /*read_batch_size=*/1, GetDefaultPool()), + "requires a payload reader for payload fields"); +} + +TEST_F(LateMaterializationBatchReaderTest, TestInvalidArgumentsFailAtCreation) { + std::shared_ptr probe_data = StructFromJson(probe_schema_, R"([[1]])"); + + ASSERT_NOK_WITH_MSG(LateMaterializationBatchReader::Create( + /*read_schema=*/nullptr, probe_schema_, probe_data, arrow::schema({}), + /*payload_reader=*/nullptr, /*read_batch_size=*/1, GetDefaultPool()), + "requires non-null schemas and data"); + ASSERT_NOK_WITH_MSG(LateMaterializationBatchReader::Create( + probe_schema_, probe_schema_, probe_data, arrow::schema({}), + /*payload_reader=*/nullptr, /*read_batch_size=*/0, GetDefaultPool()), + "read batch size should be positive"); + ASSERT_NOK_WITH_MSG(LateMaterializationBatchReader::Create( + read_schema_, probe_schema_, probe_data, arrow::schema({}), + /*payload_reader=*/nullptr, /*read_batch_size=*/1, GetDefaultPool()), + "field v1 is missing from both probe and payload schemas"); +} + +} // namespace paimon::test diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index e3366cd0..4f7fa257 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -425,6 +425,7 @@ struct CoreOptions::Impl { int32_t manifest_merge_min_count = 30; int32_t scan_manifest_entry_cache_max_snapshots = 0; int32_t read_batch_size = 1024; + int32_t read_late_materialization_max_match_rows = 1024; int32_t write_batch_size = 1024; int32_t local_sort_max_num_file_handles = 128; int32_t commit_max_retries = 10; @@ -463,6 +464,7 @@ struct CoreOptions::Impl { bool table_read_sequence_number_enabled = false; bool key_value_sequence_number_enabled = false; bool file_index_read_enabled = true; + bool read_late_materialization_enabled = false; bool enable_adaptive_prefetch_strategy = true; bool index_file_in_data_file_dir = false; bool row_tracking_enabled = false; @@ -543,6 +545,16 @@ struct CoreOptions::Impl { &source_split_open_file_cost)); // Parse read.batch-size - read batch size for file formats PAIMON_RETURN_NOT_OK(parser.Parse(Options::READ_BATCH_SIZE, &read_batch_size)); + // Parse read.late-materialization.enabled - enable selective lookup read path + PAIMON_RETURN_NOT_OK(parser.Parse(Options::READ_LATE_MATERIALIZATION_ENABLED, + &read_late_materialization_enabled)); + PAIMON_RETURN_NOT_OK( + parser.Parse(Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS, + &read_late_materialization_max_match_rows)); + if (read_late_materialization_max_match_rows <= 0) { + return Status::Invalid(fmt::format("{} should be at least 1", + Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS)); + } // Parse write.batch-size - write batch size for file formats PAIMON_RETURN_NOT_OK(parser.Parse(Options::WRITE_BATCH_SIZE, &write_batch_size)); // Parse write-buffer-size - data to build up in memory before flushing to disk @@ -1201,6 +1213,14 @@ int32_t CoreOptions::GetReadBatchSize() const { return impl_->read_batch_size; } +bool CoreOptions::ReadLateMaterializationEnabled() const { + return impl_->read_late_materialization_enabled; +} + +int32_t CoreOptions::GetReadLateMaterializationMaxMatchRows() const { + return impl_->read_late_materialization_max_match_rows; +} + int32_t CoreOptions::GetWriteBatchSize() const { return impl_->write_batch_size; } diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index ebf4eceb..2c422f78 100644 --- a/src/paimon/core/core_options.h +++ b/src/paimon/core/core_options.h @@ -114,6 +114,8 @@ class PAIMON_EXPORT CoreOptions { StartupMode GetStartupMode() const; int32_t GetReadBatchSize() const; + bool ReadLateMaterializationEnabled() const; + int32_t GetReadLateMaterializationMaxMatchRows() const; int32_t GetWriteBatchSize() const; int64_t GetWriteBufferSize() const; bool GetWriteBufferSpillable() const; diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp index ab6fc690..b0ee1875 100644 --- a/src/paimon/core/core_options_test.cpp +++ b/src/paimon/core/core_options_test.cpp @@ -68,6 +68,8 @@ TEST(CoreOptionsTest, TestDefaultValue) { ASSERT_EQ(128 * 1024 * 1024L, core_options.GetSourceSplitTargetSize()); ASSERT_EQ(4 * 1024 * 1024L, core_options.GetSourceSplitOpenFileCost()); ASSERT_EQ(1024, core_options.GetReadBatchSize()); + ASSERT_FALSE(core_options.ReadLateMaterializationEnabled()); + ASSERT_EQ(1024, core_options.GetReadLateMaterializationMaxMatchRows()); ASSERT_EQ(1024, core_options.GetWriteBatchSize()); ASSERT_EQ(256 * 1024 * 1024, core_options.GetWriteBufferSize()); ASSERT_TRUE(core_options.GetWriteBufferSpillable()); @@ -304,6 +306,8 @@ TEST(CoreOptionsTest, TestFromMap) { {Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED, "true"}, {Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED, "true"}, {Options::BUCKET_FUNCTION_TYPE, "mod"}, + {Options::READ_LATE_MATERIALIZATION_ENABLED, "true"}, + {Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS, "8"}, {"fields.metrics.map.storage-layout", "shared-shredding"}, {"fields.metrics.map.shared-shredding.max-columns", "128"}, {"fields.metrics.map.shared-shredding.column-placement-policy", "lru"}}; @@ -465,6 +469,8 @@ TEST(CoreOptionsTest, TestFromMap) { ASSERT_TRUE(core_options.LookupRemoteFileEnabled()); ASSERT_EQ(core_options.GetLookupRemoteLevelThreshold(), 2); ASSERT_EQ(BucketFunctionType::MOD, core_options.GetBucketFunctionType()); + ASSERT_TRUE(core_options.ReadLateMaterializationEnabled()); + ASSERT_EQ(8, core_options.GetReadLateMaterializationMaxMatchRows()); ASSERT_EQ(MapStorageLayout::SHARED_SHREDDING, core_options.GetMapStorageLayout("metrics").value()); ASSERT_EQ(128, core_options.GetMapSharedShreddingMaxColumns("metrics").value()); @@ -502,7 +508,6 @@ TEST(CoreOptionsTest, TestInvalidCase) { ASSERT_NOK_WITH_MSG( CoreOptions::FromMap({{Options::WRITE_SEQUENCE_NUMBER_INIT_MODE, "invalid"}}), "invalid write sequence number init mode: invalid"); - ASSERT_OK_AND_ASSIGN(CoreOptions invalid_strategy, CoreOptions::FromMap({{"fields.f0.nested-key-null-strategy", "invalid"}})); ASSERT_NOK_WITH_MSG(invalid_strategy.FieldNestedUpdateAggNestedKeyNullStrategy("f0"), @@ -511,6 +516,9 @@ TEST(CoreOptionsTest, TestInvalidCase) { CoreOptions::FromMap({{"fields.f0.count-limit", "-1"}})); ASSERT_NOK_WITH_MSG(negative_limit.FieldNestedUpdateAggCountLimit("f0"), "must not be negative"); + ASSERT_NOK_WITH_MSG( + CoreOptions::FromMap({{Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS, "0"}}), + "read.late-materialization.max-match-rows should be at least 1"); } TEST(CoreOptionsTest, TestNestedKeyNullStrategyIsCaseInsensitive) { diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index 0850d943..542d266b 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -79,7 +79,8 @@ Result>> AbstractSplitRead::CreateR const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory, - const std::map& extra_format_options) const { + const std::map& extra_format_options, + const std::optional& file_selection) const { if (data_files.empty()) { return std::vector>(); } @@ -98,7 +99,7 @@ Result>> AbstractSplitRead::CreateR std::unique_ptr file_reader, CreateFieldMappingReader(data_file_path, file, partition, reader_builder.get(), field_mapping_builder.get(), dv_factory, row_ranges, - data_file_path_factory)); + data_file_path_factory, file_selection)); if (file_reader) { raw_file_readers.push_back(std::move(file_reader)); } @@ -169,7 +170,8 @@ Result> AbstractSplitRead::CreateFieldMappingRe const BinaryRow& partition, const ReaderBuilder* reader_builder, const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, - const std::shared_ptr& data_file_path_factory) const { + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const { std::shared_ptr data_schema; if (file_meta->schema_id == context_->GetTableSchema()->Id()) { data_schema = context_->GetTableSchema(); @@ -230,10 +232,11 @@ Result> AbstractSplitRead::CreateFieldMappingRe } const auto& predicate = field_mapping->non_partition_info.non_partition_filter; auto all_data_schema = DataField::ConvertDataFieldsToArrowSchema(data_schema->Fields()); - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr final_reader, - ApplyIndexAndDvReaderIfNeeded( - std::move(file_reader), file_meta, all_data_schema, read_schema, - predicate, dv_factory, row_ranges, data_file_path_factory)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr final_reader, + ApplyIndexAndDvReaderIfNeeded(std::move(file_reader), file_meta, all_data_schema, + read_schema, predicate, dv_factory, row_ranges, + data_file_path_factory, file_selection)); if (!final_reader) { // file is skipped by index or dv return std::unique_ptr(); diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h index ea3f9070..289ff838 100644 --- a/src/paimon/core/operation/abstract_split_read.h +++ b/src/paimon/core/operation/abstract_split_read.h @@ -66,13 +66,16 @@ class AbstractSplitRead : public SplitRead { /// `extra_format_options` are merged over the table options when building the format /// reader, e.g. to switch the blob format reader into placeholder-aware mode for the /// data-evolution blob fallback read path. + /// `file_selection`, when present, contains file-local row IDs and is applied independently + /// to every file in `data_files`. Result>> CreateRawFileReaders( const BinaryRow& partition, const std::vector>& data_files, const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory, - const std::map& extra_format_options) const; + const std::map& extra_format_options, + const std::optional& file_selection = std::nullopt) const; protected: AbstractSplitRead(const std::shared_ptr& path_factory, @@ -91,7 +94,8 @@ class AbstractSplitRead : public SplitRead { const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, - const std::shared_ptr& data_file_path_factory) const = 0; + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const = 0; // 1. project write cols to data schema // 2. add partition fields (if write cols not contain) @@ -115,7 +119,8 @@ class AbstractSplitRead : public SplitRead { const BinaryRow& partition, const ReaderBuilder* reader_builder, const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, - const std::shared_ptr& data_file_path_factory) const; + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const; Result, std::set>> ApplySharedShreddingReaderIfNeeded(std::unique_ptr&& file_reader, diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index e6658f7b..06c572de 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -428,7 +428,8 @@ Result> DataEvolutionSplitRead::ApplyIndexAndDv const std::shared_ptr& data_schema, const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, - const std::shared_ptr& data_file_path_factory) const { + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const { if (predicate) { assert(false); // as DataEvolutionSplitRead will skip predicate @@ -443,6 +444,11 @@ Result> DataEvolutionSplitRead::ApplyIndexAndDv } PAIMON_ASSIGN_OR_RAISE(std::optional selection_row_ids, file->ToFileSelection(row_ranges)); + if (file_selection) { + selection_row_ids = selection_row_ids ? RoaringBitmap32::And(selection_row_ids.value(), + file_selection.value()) + : file_selection; + } ::ArrowSchema c_read_schema; PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, &c_read_schema)); PAIMON_RETURN_NOT_OK( diff --git a/src/paimon/core/operation/data_evolution_split_read.h b/src/paimon/core/operation/data_evolution_split_read.h index 983ca29d..36dc1bc7 100644 --- a/src/paimon/core/operation/data_evolution_split_read.h +++ b/src/paimon/core/operation/data_evolution_split_read.h @@ -94,7 +94,8 @@ class DataEvolutionSplitRead : public AbstractSplitRead { const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, - const std::shared_ptr& data_file_path_factory) const override; + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const override; private: /// Files for partial field. diff --git a/src/paimon/core/operation/merge_file_split_read.cpp b/src/paimon/core/operation/merge_file_split_read.cpp index b753ea43..bbcfda53 100644 --- a/src/paimon/core/operation/merge_file_split_read.cpp +++ b/src/paimon/core/operation/merge_file_split_read.cpp @@ -197,7 +197,8 @@ Result> MergeFileSplitRead::ApplyIndexAndDvRead const std::shared_ptr& data_schema, const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& ranges, - const std::shared_ptr& data_file_path_factory) const { + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const { // merge read does not use index std::shared_ptr deletion_vector; if (dv_factory) { @@ -215,6 +216,11 @@ Result> MergeFileSplitRead::ApplyIndexAndDvRead PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows, file_reader->GetNumberOfRows()); actual_selection.value().Flip(0, num_rows); } + if (file_selection) { + actual_selection = actual_selection ? RoaringBitmap32::And(actual_selection.value(), + file_selection.value()) + : file_selection; + } ::ArrowSchema c_read_schema; PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, &c_read_schema)); diff --git a/src/paimon/core/operation/merge_file_split_read.h b/src/paimon/core/operation/merge_file_split_read.h index d4bfa727..a6ba82cf 100644 --- a/src/paimon/core/operation/merge_file_split_read.h +++ b/src/paimon/core/operation/merge_file_split_read.h @@ -97,7 +97,8 @@ class MergeFileSplitRead : public AbstractSplitRead { const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& ranges, - const std::shared_ptr& data_file_path_factory) const override; + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const override; Result> CreateSortMergeReaderForSection( const std::vector& section, const BinaryRow& partition, diff --git a/src/paimon/core/operation/raw_file_split_read.cpp b/src/paimon/core/operation/raw_file_split_read.cpp index eabe8426..481410fc 100644 --- a/src/paimon/core/operation/raw_file_split_read.cpp +++ b/src/paimon/core/operation/raw_file_split_read.cpp @@ -18,15 +18,28 @@ #include "paimon/core/operation/raw_file_split_read.h" +#include +#include +#include #include +#include #include #include +#include "arrow/api.h" #include "arrow/c/abi.h" #include "arrow/c/bridge.h" +#include "arrow/util/checked_cast.h" +#include "fmt/format.h" #include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/predicate/predicate_filter.h" #include "paimon/common/reader/complete_row_kind_batch_reader.h" #include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/reader/late_materialization_batch_reader.h" +#include "paimon/common/reader/reader_utils.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/object_utils.h" #include "paimon/core/core_options.h" @@ -42,6 +55,7 @@ #include "paimon/file_index/bitmap_index_result.h" #include "paimon/file_index/file_index_result.h" #include "paimon/memory/memory_pool.h" +#include "paimon/predicate/predicate_utils.h" #include "paimon/reader/file_batch_reader.h" #include "paimon/status.h" #include "paimon/table/source/data_split.h" @@ -52,6 +66,17 @@ class DataFilePathFactory; class Executor; class Predicate; +struct RawFileSplitRead::LateMaterializationPlan { + std::shared_ptr probe_schema; + std::shared_ptr payload_schema; +}; + +struct RawFileSplitRead::LateMaterializationReadResult { + bool applied = false; + std::vector> readers; + std::shared_ptr completed_metrics; +}; + RawFileSplitRead::RawFileSplitRead(const std::shared_ptr& path_factory, const std::shared_ptr& context, const std::shared_ptr& memory_pool, @@ -80,6 +105,19 @@ Result> RawFileSplitRead::CreateReader( PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, path_factory_->CreateDataFilePathFactory(partition, bucket)); + std::shared_ptr completed_late_materialization_metrics; + if (options_.ReadLateMaterializationEnabled()) { + PAIMON_ASSIGN_OR_RAISE(LateMaterializationReadResult late_result, + TryCreateLateMaterializedReader(partition, data_files, predicate, + dv_factory, data_file_path_factory)); + completed_late_materialization_metrics = std::move(late_result.completed_metrics); + if (late_result.applied) { + std::unique_ptr late_reader = std::make_unique( + std::move(late_result.readers), pool_, completed_late_materialization_metrics); + return std::make_unique(std::move(late_reader), pool_); + } + } + PAIMON_ASSIGN_OR_RAISE( std::vector> raw_file_readers, CreateRawFileReaders(partition, data_files, raw_read_schema_, predicate, dv_factory, @@ -88,7 +126,8 @@ Result> RawFileSplitRead::CreateReader( auto raw_readers = ObjectUtils::MoveVector>(std::move(raw_file_readers)); - auto concat_batch_reader = std::make_unique(std::move(raw_readers), pool_); + auto concat_batch_reader = std::make_unique( + std::move(raw_readers), pool_, completed_late_materialization_metrics); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr batch_reader, ApplyPredicateFilterIfNeeded(std::move(concat_batch_reader), predicate)); return std::make_unique(std::move(batch_reader), pool_); @@ -129,12 +168,205 @@ Result RawFileSplitRead::Match(const std::shared_ptr& split, return matched; } +Result> +RawFileSplitRead::BuildLateMaterializationPlan(const std::shared_ptr& predicate) const { + if (!context_->EnablePredicateFilter() || predicate == nullptr) { + return std::optional(); + } + + std::set predicate_field_names; + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &predicate_field_names)); + if (predicate_field_names.empty()) { + return std::optional(); + } + + std::vector> probe_fields; + std::set probe_names; + for (const auto& field : raw_read_schema_->fields()) { + if (predicate_field_names.count(field->name()) > 0) { + probe_fields.push_back(field); + probe_names.insert(field->name()); + } + } + for (const auto& field_name : predicate_field_names) { + if (probe_names.count(field_name) > 0) { + continue; + } + PAIMON_ASSIGN_OR_RAISE(DataField data_field, + context_->GetTableSchema()->GetField(field_name)); + probe_fields.push_back(data_field.ArrowField()); + probe_names.insert(field_name); + } + + if (probe_fields.empty()) { + return std::optional(); + } + + std::vector> payload_fields; + for (const auto& field : raw_read_schema_->fields()) { + if (probe_names.count(field->name()) == 0) { + payload_fields.push_back(field); + } + } + if (payload_fields.empty()) { + return std::optional(); + } + + return std::optional(LateMaterializationPlan{ + arrow::schema(std::move(probe_fields)), arrow::schema(std::move(payload_fields))}); +} + +auto RawFileSplitRead::TryCreateLateMaterializedReader( + const BinaryRow& partition, const std::vector>& data_files, + const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, + const std::shared_ptr& data_file_path_factory) const + -> Result { + PAIMON_ASSIGN_OR_RAISE(std::optional plan, + BuildLateMaterializationPlan(predicate)); + if (!plan) { + return LateMaterializationReadResult(); + } + + std::map probe_field_name_to_idx; + for (int32_t i = 0; i < plan->probe_schema->num_fields(); ++i) { + probe_field_name_to_idx[plan->probe_schema->field(i)->name()] = i; + } + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr probe_predicate, + PredicateUtils::CreatePickedFieldFilter(predicate, probe_field_name_to_idx)); + std::shared_ptr predicate_filter = + std::dynamic_pointer_cast(probe_predicate); + if (!predicate_filter) { + return LateMaterializationReadResult(); + } + + int64_t total_match_rows = 0; + std::vector> readers; + std::shared_ptr completed_metrics = std::make_shared(); + for (const auto& file : data_files) { + PAIMON_ASSIGN_OR_RAISE( + std::vector> probe_readers, + CreateRawFileReaders(partition, {file}, plan->probe_schema, predicate, dv_factory, + /*row_ranges=*/{}, data_file_path_factory, + /*extra_format_options=*/{})); + if (probe_readers.empty()) { + continue; + } + if (probe_readers.size() != 1) { + return Status::Invalid("late materialization expects one probe reader per data file"); + } + + std::vector> probe_chunks; + RoaringBitmap32 selected_file_rows; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + probe_readers[0]->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(batch_with_bitmap)) { + break; + } + BatchReader::ReadBatchWithBitmap moved_batch = std::move(batch_with_bitmap); + BatchReader::ReadBatch& batch = moved_batch.first; + RoaringBitmap32& valid_bitmap = moved_batch.second; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr probe_array, + arrow::ImportArray(batch.first.get(), batch.second.get())); + std::shared_ptr probe_struct = + arrow::internal::checked_pointer_cast(probe_array); + PAIMON_ASSIGN_OR_RAISE(std::vector predicate_result, + predicate_filter->Test(*probe_struct)); + if (static_cast(predicate_result.size()) != probe_struct->length()) { + return Status::Invalid(fmt::format( + "late materialization predicate returned {} results for {} probe rows", + predicate_result.size(), probe_struct->length())); + } + + RoaringBitmap32 selected_bitmap; + for (auto iter = valid_bitmap.Begin(); iter != valid_bitmap.End(); ++iter) { + uint32_t batch_row_id = *iter; + if (batch_row_id >= predicate_result.size()) { + return Status::Invalid(fmt::format( + "late materialization bitmap row {} exceeds probe batch length {}", + batch_row_id, predicate_result.size())); + } + if (!predicate_result[batch_row_id]) { + continue; + } + PAIMON_ASSIGN_OR_RAISE(uint64_t file_row_id, + probe_readers[0]->GetPreviousBatchFileRowId(batch_row_id)); + if (file_row_id > std::numeric_limits::max()) { + return Status::Invalid( + fmt::format("late materialization file row id {} exceeds bitmap capacity", + file_row_id)); + } + selected_file_rows.Add(static_cast(file_row_id)); + selected_bitmap.Add(batch_row_id); + } + + if (!selected_bitmap.IsEmpty()) { + total_match_rows += static_cast(selected_bitmap.Cardinality()); + if (total_match_rows > + static_cast(options_.GetReadLateMaterializationMaxMatchRows())) { + completed_metrics->Merge(probe_readers[0]->GetReaderMetrics()); + probe_readers[0]->Close(); + for (const auto& reader : readers) { + completed_metrics->Merge(reader->GetReaderMetrics()); + reader->Close(); + } + return LateMaterializationReadResult{/*applied=*/false, /*readers=*/{}, + std::move(completed_metrics)}; + } + PAIMON_ASSIGN_OR_RAISE( + arrow::ArrayVector selected_arrays, + ReaderUtils::GenerateFilteredArrayVector(probe_struct, selected_bitmap)); + probe_chunks.insert(probe_chunks.end(), selected_arrays.begin(), + selected_arrays.end()); + } + } + completed_metrics->Merge(probe_readers[0]->GetReaderMetrics()); + probe_readers[0]->Close(); + + if (selected_file_rows.IsEmpty()) { + continue; + } + + std::unique_ptr probe_arrow_pool = GetArrowPool(pool_); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr probe_data, + arrow::Concatenate(probe_chunks, probe_arrow_pool.get())); + std::shared_ptr probe_struct = + arrow::internal::checked_pointer_cast(probe_data); + PAIMON_ASSIGN_OR_RAISE( + std::vector> payload_readers, + CreateRawFileReaders(partition, {file}, plan->payload_schema, + /*predicate=*/nullptr, dv_factory, + /*row_ranges=*/std::nullopt, data_file_path_factory, + /*extra_format_options=*/{}, selected_file_rows)); + if (payload_readers.empty()) { + return Status::Invalid("late materialization payload reader was filtered out"); + } + if (payload_readers.size() != 1) { + return Status::Invalid("late materialization expects one payload reader per data file"); + } + std::unique_ptr payload_reader = std::move(payload_readers[0]); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr late_reader, + LateMaterializationBatchReader::Create( + raw_read_schema_, plan->probe_schema, std::move(probe_struct), plan->payload_schema, + std::move(payload_reader), options_.GetReadBatchSize(), pool_, + std::move(probe_arrow_pool))); + readers.push_back(std::move(late_reader)); + } + + return LateMaterializationReadResult{/*applied=*/true, std::move(readers), + std::move(completed_metrics)}; +} + Result> RawFileSplitRead::ApplyIndexAndDvReaderIfNeeded( std::unique_ptr&& file_reader, const std::shared_ptr& file, const std::shared_ptr& data_schema, const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& ranges, - const std::shared_ptr& data_file_path_factory) const { + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const { std::shared_ptr file_index_result; if (options_.FileIndexReadEnabled()) { PAIMON_ASSIGN_OR_RAISE( @@ -174,6 +406,27 @@ Result> RawFileSplitRead::ApplyIndexAndDvReader actual_selection.value().Flip(0, num_rows); } + // `ranges` uses global row ids. ToFileSelection intersects it with this file's global row id + // span and returns a file-local bitmap, which can be merged with index and deletion bitmaps. + PAIMON_ASSIGN_OR_RAISE(std::optional range_selection, + file->ToFileSelection(ranges)); + if (range_selection) { + if (actual_selection) { + actual_selection = + RoaringBitmap32::And(actual_selection.value(), range_selection.value()); + } else { + actual_selection = std::move(range_selection); + } + } + + // Late materialization already operates on one FileBatchReader at a time, so its selection + // can stay file-local instead of making a round trip through global row IDs. + if (file_selection) { + actual_selection = actual_selection ? RoaringBitmap32::And(actual_selection.value(), + file_selection.value()) + : file_selection; + } + if (actual_selection && actual_selection.value().IsEmpty()) { return std::unique_ptr(); } diff --git a/src/paimon/core/operation/raw_file_split_read.h b/src/paimon/core/operation/raw_file_split_read.h index 1580cd84..f9d69ef1 100644 --- a/src/paimon/core/operation/raw_file_split_read.h +++ b/src/paimon/core/operation/raw_file_split_read.h @@ -19,6 +19,7 @@ #pragma once #include +#include #include #include #include @@ -84,7 +85,20 @@ class RawFileSplitRead : public AbstractSplitRead { const std::shared_ptr& read_schema, const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& ranges, - const std::shared_ptr& data_file_path_factory) const override; + const std::shared_ptr& data_file_path_factory, + const std::optional& file_selection) const override; + + private: + struct LateMaterializationPlan; + struct LateMaterializationReadResult; + + Result TryCreateLateMaterializedReader( + const BinaryRow& partition, const std::vector>& data_files, + const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, + const std::shared_ptr& data_file_path_factory) const; + + Result> BuildLateMaterializationPlan( + const std::shared_ptr& predicate) const; }; } // namespace paimon diff --git a/src/paimon/core/operation/raw_file_split_read_test.cpp b/src/paimon/core/operation/raw_file_split_read_test.cpp index 569f8dfd..8daaffad 100644 --- a/src/paimon/core/operation/raw_file_split_read_test.cpp +++ b/src/paimon/core/operation/raw_file_split_read_test.cpp @@ -18,7 +18,9 @@ #include "paimon/core/operation/raw_file_split_read.h" +#include #include +#include #include #include #include @@ -43,6 +45,8 @@ #include "paimon/format/file_format.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" +#include "paimon/predicate/literal.h" +#include "paimon/predicate/predicate_builder.h" #include "paimon/read_context.h" #include "paimon/status.h" #include "paimon/table/source/data_split.h" @@ -130,12 +134,20 @@ class RawFileSplitReadTest : public ::testing::Test { } void CheckReadResult(const std::shared_ptr& read_schema, - const std::shared_ptr& expected_array) const { + const std::shared_ptr& expected_array, + const std::shared_ptr& predicate = nullptr, + bool enable_predicate_filter = false, + const std::map& options = {}) const { std::string path = paimon::test::GetDataDir() + "/orc/multi_partition_append_table.db/" "multi_partition_append_table"; ReadContextBuilder context_builder(path); context_builder.SetReadFieldNames(read_schema->field_names()); + context_builder.SetOptions(options); + if (predicate) { + context_builder.SetPredicate(predicate); + } + context_builder.EnablePredicateFilter(enable_predicate_filter); ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, context_builder.Finish()); SchemaManager schema_manager(std::make_shared(), read_context->GetPath()); ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0)); @@ -377,6 +389,106 @@ TEST_F(RawFileSplitReadTest, TestCreateReaderWithNonPartitionWithReserveSequence CheckReadResult(read_schema, expected_array); } +TEST_F(RawFileSplitReadTest, TestLateMaterializationWithoutFirstRowId) { + std::vector read_fields = {DataField(0, arrow::field("f0", arrow::utf8())), + DataField(3, arrow::field("f3", arrow::float64()))}; + std::shared_ptr read_schema = + DataField::ConvertDataFieldsToArrowSchema(read_fields); + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/1, /*field_name=*/"f3", FieldType::DOUBLE, Literal(12.5)); + + std::vector> expected_fields = read_schema->fields(); + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON( + arrow::struct_(expected_fields), + {R"([[0, "Emily", 13.1], [0, "Tony", 14.1], [0, "Lucy", 14.1]])"}, &expected_array); + ASSERT_TRUE(array_status.ok()) << array_status.ToString(); + + CheckReadResult(read_schema, expected_array, predicate, /*enable_predicate_filter=*/true, + {{Options::READ_LATE_MATERIALIZATION_ENABLED, "true"}}); +} + +TEST_F(RawFileSplitReadTest, TestLateMaterializationDisabledUsesNormalPath) { + std::vector read_fields = {DataField(0, arrow::field("f0", arrow::utf8())), + DataField(3, arrow::field("f3", arrow::float64()))}; + std::shared_ptr read_schema = + DataField::ConvertDataFieldsToArrowSchema(read_fields); + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/1, /*field_name=*/"f3", FieldType::DOUBLE, Literal(12.5)); + + std::vector> expected_fields = read_schema->fields(); + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON( + arrow::struct_(expected_fields), + {R"([[0, "Emily", 13.1], [0, "Tony", 14.1], [0, "Lucy", 14.1]])"}, &expected_array); + ASSERT_TRUE(array_status.ok()) << array_status.ToString(); + + CheckReadResult(read_schema, expected_array, predicate, /*enable_predicate_filter=*/true); +} + +TEST_F(RawFileSplitReadTest, TestLateMaterializationFallsBackWithoutPredicateFilter) { + std::vector read_fields = {DataField(0, arrow::field("f0", arrow::utf8())), + DataField(3, arrow::field("f3", arrow::float64()))}; + std::shared_ptr read_schema = + DataField::ConvertDataFieldsToArrowSchema(read_fields); + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/1, /*field_name=*/"f3", FieldType::DOUBLE, Literal(0.0)); + + std::vector> expected_fields = read_schema->fields(); + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON( + arrow::struct_(expected_fields), + {R"([[0, "Bob", 12.1], [0, "Emily", 13.1], [0, "Tony", 14.1], + [0, "Lucy", 14.1], [0, "Alice", 11.1]])"}, + &expected_array); + ASSERT_TRUE(array_status.ok()) << array_status.ToString(); + + CheckReadResult(read_schema, expected_array, predicate, /*enable_predicate_filter=*/false, + {{Options::READ_LATE_MATERIALIZATION_ENABLED, "true"}}); +} + +TEST_F(RawFileSplitReadTest, TestLateMaterializationFallsBackWithoutPayloadColumns) { + std::vector read_fields = {DataField(3, arrow::field("f3", arrow::float64()))}; + std::shared_ptr read_schema = + DataField::ConvertDataFieldsToArrowSchema(read_fields); + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/0, /*field_name=*/"f3", FieldType::DOUBLE, Literal(12.5)); + + std::vector> expected_fields = read_schema->fields(); + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON( + arrow::struct_(expected_fields), {R"([[0, 13.1], [0, 14.1], [0, 14.1]])"}, &expected_array); + ASSERT_TRUE(array_status.ok()) << array_status.ToString(); + + CheckReadResult(read_schema, expected_array, predicate, /*enable_predicate_filter=*/true, + {{Options::READ_LATE_MATERIALIZATION_ENABLED, "true"}}); +} + +TEST_F(RawFileSplitReadTest, TestLateMaterializationFallsBackWhenMatchLimitExceeded) { + std::vector read_fields = {DataField(0, arrow::field("f0", arrow::utf8())), + DataField(3, arrow::field("f3", arrow::float64()))}; + std::shared_ptr read_schema = + DataField::ConvertDataFieldsToArrowSchema(read_fields); + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/1, /*field_name=*/"f3", FieldType::DOUBLE, Literal(12.5)); + + std::vector> expected_fields = read_schema->fields(); + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + std::shared_ptr expected_array; + auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON( + arrow::struct_(expected_fields), + {R"([[0, "Emily", 13.1], [0, "Tony", 14.1], [0, "Lucy", 14.1]])"}, &expected_array); + ASSERT_TRUE(array_status.ok()) << array_status.ToString(); + + CheckReadResult(read_schema, expected_array, predicate, /*enable_predicate_filter=*/true, + {{Options::READ_LATE_MATERIALIZATION_ENABLED, "true"}, + {Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS, "1"}}); +} + TEST_F(RawFileSplitReadTest, TestEmptyPlan) { std::string path = paimon::test::GetDataDir() + "/orc/multi_partition_append_table.db/" @@ -505,9 +617,7 @@ TEST_F(RawFileSplitReadTest, TestMatch) { split_read->Match(data_split, /*force_keep_delete=*/false)); ASSERT_FALSE(match_result); } - { - ASSERT_NOK(split_read->Match(nullptr, /*force_keep_delete=*/false)); - } + { ASSERT_NOK(split_read->Match(nullptr, /*force_keep_delete=*/false)); } } } // namespace paimon::test diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index 9b8f9444..ba619743 100644 --- a/test/inte/write_and_read_inte_test.cpp +++ b/test/inte/write_and_read_inte_test.cpp @@ -290,6 +290,82 @@ TEST_P(WriteAndReadInteTest, TestAppendSimple) { ASSERT_TRUE(success); } +TEST_P(WriteAndReadInteTest, TestAppendLateMaterializationWithoutFirstRowIdAndFallback) { + auto [file_format, file_system] = GetParam(); + if (file_format != "parquet" || file_system != "local") { + return; + } + + arrow::FieldVector fields = {arrow::field("id", arrow::int32()), + arrow::field("payload", arrow::utf8())}; + std::shared_ptr schema = arrow::schema(fields); + std::map options = { + {Options::MANIFEST_FORMAT, "avro"}, + {Options::FILE_FORMAT, file_format}, + {Options::TARGET_FILE_SIZE, "1048576"}, + {Options::BUCKET, "-1"}, + {Options::FILE_SYSTEM, file_system}, + }; + ASSERT_OK_AND_ASSIGN( + auto helper, TestHelper::Create(test_dir_, schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, options, + /*is_streaming_mode=*/false)); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), + R"([[0, "v0"], [1, "v1"], [2, "v2"], + [3, "v3"], [4, "v4"], [5, "v5"]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + + ASSERT_OK_AND_ASSIGN(auto files, CurrentDataFiles(options)); + ASSERT_FALSE(files.empty()); + for (const auto& [bucket_path, file] : files) { + ASSERT_FALSE(file->first_row_id.has_value()) << bucket_path << "/" << file->file_name; + } + + std::shared_ptr predicate = PredicateBuilder::GreaterThan( + /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(2)); + ASSERT_TRUE(predicate); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, InnerScan(options)); + + arrow::FieldVector expected_fields = fields; + expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8())); + arrow::Result> expected_array_result = + arrow::ipc::internal::json::ArrayFromJSON( + arrow::struct_(expected_fields), R"([[0, 3, "v3"], [0, 4, "v4"], [0, 5, "v5"]])"); + ASSERT_TRUE(expected_array_result.ok()) << expected_array_result.status().ToString(); + std::shared_ptr expected_array = std::move(expected_array_result).ValueOrDie(); + std::shared_ptr expected = + std::make_shared(expected_array); + + auto read_and_check = [&](int32_t max_match_rows) { + std::map read_options = options; + read_options[Options::READ_LATE_MATERIALIZATION_ENABLED] = "true"; + read_options[Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS] = + std::to_string(max_match_rows); + ReadContextBuilder read_context_builder(PathUtil::JoinPath(test_dir_, "foo.db/bar")); + read_context_builder.SetOptions(read_options) + .SetPredicate(predicate) + .EnablePredicateFilter(true); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, + read_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_read, + TableRead::Create(std::move(read_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_reader, + table_read->CreateReader(plan->Splits())); + ASSERT_OK_AND_ASSIGN(std::shared_ptr actual, + ReadResultCollector::CollectResult(batch_reader.get())); + ASSERT_TRUE(expected->Equals(actual)) << actual->ToString(); + }; + + // Three rows match. A larger bound exercises late materialization; a bound of one forces the + // probe to abort and the split to be read again through the normal predicate-filtered path. + read_and_check(/*max_match_rows=*/10); + read_and_check(/*max_match_rows=*/1); +} + TEST_P(WriteAndReadInteTest, TestPKSimple) { arrow::FieldVector fields = { arrow::field("pk", arrow::utf8()),