diff --git a/src/paimon/common/reader/blob_fallback_batch_reader.cpp b/src/paimon/common/reader/blob_fallback_batch_reader.cpp index a5e238557..24d902169 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader.cpp +++ b/src/paimon/common/reader/blob_fallback_batch_reader.cpp @@ -35,6 +35,7 @@ #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/scope_guard.h" namespace paimon { @@ -74,6 +75,9 @@ Result> BlobFallbackBatchReader::Create return Status::Invalid("Blob fallback sequence group should not be empty."); } for (const auto& segment : segments) { + if (segment.reader != nullptr && segment.blob_context == nullptr) { + return Status::Invalid("Blob fallback file segment needs a blob read context."); + } if (segment.reader == nullptr && segment.gap_selected_ranges.empty()) { return Status::Invalid( "Blob fallback gap segment should cover at least one selected row id."); @@ -101,24 +105,35 @@ BlobFallbackBatchReader::BlobFallbackBatchReader( arrow_pool_(arrow_pool), finished_reader_metrics_(std::make_shared()) {} +Result BlobFallbackBatchReader::ReadSegmentBatch( + const Segment& segment, int32_t max_rows, bool read_payload) { + const std::optional previous_request = segment.blob_context->request; + segment.blob_context->request = BlobBatchRequest{max_rows, read_payload}; + ScopeGuard reset_request([context = segment.blob_context, previous_request]() { + context->request = previous_request; + }); + return segment.reader->NextBatchWithBitmap(); +} + Result BlobFallbackBatchReader::FillWindow(size_t group_idx, int64_t want, + const std::vector* group_choice, std::vector* chunks) { GroupCursor& cursor = groups_[group_idx]; int64_t collected = 0; + int64_t run_end = 0; + bool materialize = true; while (collected < want) { - if (!cursor.pending.empty()) { - const std::shared_ptr& front = cursor.pending.front(); - int64_t available = front->length() - cursor.pending_pos; - int64_t take = std::min(available, want - collected); - chunks->push_back( - Chunk{front, cursor.pending_pos, take, {}}); // NOLINT(modernize-use-emplace) - cursor.pending_pos += take; - collected += take; - if (cursor.pending_pos == front->length()) { - cursor.pending.pop_front(); - cursor.pending_pos = 0; + // Split the window into contiguous read/skip runs, grouping rows with the same operation. + if (collected == run_end) { + materialize = !group_choice || (*group_choice)[collected] == -1; + run_end = collected + 1; + if (!group_choice) { + run_end = want; + } else { + while (run_end < want && ((*group_choice)[run_end] == -1) == materialize) { + run_end++; + } } - continue; } if (cursor.segment_idx >= cursor.segments.size()) { // group exhausted; only the first group may define a shorter window @@ -136,9 +151,9 @@ Result BlobFallbackBatchReader::FillWindow(size_t group_idx, int64_t wa } const Range& range = segment.gap_selected_ranges[cursor.gap_range_idx]; int64_t remaining = range.Count() - cursor.gap_range_pos; - int64_t take = std::min(remaining, want - collected); - Chunk chunk{nullptr, 0, take, {}}; - if (row_id_field_idx_ >= 0) { + int64_t take = std::min(remaining, run_end - collected); + Chunk chunk{nullptr, take, {}}; + if (group_idx == 0 && row_id_field_idx_ >= 0) { chunk.gap_row_ids.reserve(take); for (int64_t k = 0; k < take; k++) { chunk.gap_row_ids.push_back(range.from + cursor.gap_range_pos + k); @@ -153,8 +168,12 @@ Result BlobFallbackBatchReader::FillWindow(size_t group_idx, int64_t wa } continue; } - PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap batch_with_bitmap, - segment.reader->NextBatchWithBitmap()); + const int64_t remaining = run_end - collected; + // Bound selected input rows by this run. Outer filtering can return fewer valid rows, + // but must not read ahead into a run requiring a different payload request. + PAIMON_ASSIGN_OR_RAISE( + ReadBatchWithBitmap batch_with_bitmap, + ReadSegmentBatch(segment, static_cast(remaining), materialize)); if (BatchReader::IsEofBatch(batch_with_bitmap)) { segment.reader->Close(); finished_reader_metrics_->Merge(segment.reader->GetReaderMetrics()); @@ -166,17 +185,24 @@ Result BlobFallbackBatchReader::FillWindow(size_t group_idx, int64_t wa auto& [c_array, c_schema] = read_batch; PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr src_array, arrow::ImportArray(c_array.get(), c_schema.get())); + if (src_array->length() > remaining) { + return Status::Invalid( + fmt::format("Blob fallback reader returned {} rows exceeding requested limit {}", + src_array->length(), remaining)); + } PAIMON_ASSIGN_OR_RAISE(arrow::ArrayVector selected_array_vec, ReaderUtils::GenerateFilteredArrayVector(src_array, bitmap)); for (const auto& selected_array : selected_array_vec) { - if (selected_array->length() == 0) { - continue; - } if (!selected_array || selected_array->type_id() != arrow::Type::STRUCT) { return Status::Invalid("Blob fallback expects file readers to emit struct arrays."); } + if (selected_array->length() == 0) { + continue; + } + const int64_t count = selected_array->length(); auto struct_array = checked_pointer_cast(selected_array); - cursor.pending.push_back(std::move(struct_array)); + chunks->push_back(Chunk{materialize ? struct_array : nullptr, count, {}}); + collected += count; } } return collected; @@ -198,7 +224,7 @@ Result> BlobFallbackBatchReader::ComputePlaceholderFlags( if (blob_col->type_id() == arrow::Type::MAP) { auto map_col = checked_pointer_cast(blob_col); for (int64_t k = 0; k < chunk.length; k++) { - flags[pos + k] = BlobUtils::IsMapBlobPlaceholder(*map_col, chunk.offset + k); + flags[pos + k] = BlobUtils::IsMapBlobPlaceholder(*map_col, k); } pos += chunk.length; continue; @@ -206,7 +232,7 @@ Result> BlobFallbackBatchReader::ComputePlaceholderFlags( if (blob_col->type_id() == arrow::Type::LIST) { auto list_col = checked_pointer_cast(blob_col); for (int64_t k = 0; k < chunk.length; k++) { - flags[pos + k] = BlobUtils::IsArrayBlobPlaceholder(*list_col, chunk.offset + k); + flags[pos + k] = BlobUtils::IsArrayBlobPlaceholder(*list_col, k); } pos += chunk.length; continue; @@ -219,11 +245,10 @@ Result> BlobFallbackBatchReader::ComputePlaceholderFlags( } auto binary_col = checked_pointer_cast(blob_col); for (int64_t k = 0; k < chunk.length; k++) { - int64_t idx = chunk.offset + k; - if (binary_col->IsNull(idx)) { + if (binary_col->IsNull(k)) { continue; } - std::string_view value = binary_col->GetView(idx); + std::string_view value = binary_col->GetView(k); if (BlobDefs::IsPlaceholderSentinel(value.data(), value.size())) { flags[pos + k] = true; } @@ -235,17 +260,18 @@ Result> BlobFallbackBatchReader::ComputePlaceholderFlags( } Result> BlobFallbackBatchReader::AssembleRowIdRun( - const std::vector& chunks, int64_t run_start, int64_t run_end) const { + const std::vector& chunks, int64_t run_start, int64_t run_end, + ChunkCursor* cursor) const { arrow::ArrayVector pieces; - int64_t pos = 0; - for (const auto& chunk : chunks) { + while (cursor->chunk_idx < chunks.size() && cursor->chunk_start < run_end) { + const auto& chunk = chunks[cursor->chunk_idx]; + const int64_t pos = cursor->chunk_start; int64_t overlap_start = std::max(run_start, pos); int64_t overlap_end = std::min(run_end, pos + chunk.length); if (overlap_start < overlap_end) { if (chunk.array != nullptr) { std::shared_ptr column = chunk.array->field(row_id_field_idx_); - pieces.push_back(column->Slice(chunk.offset + (overlap_start - pos), - overlap_end - overlap_start)); + pieces.push_back(column->Slice(overlap_start - pos, overlap_end - overlap_start)); } else { arrow::Int64Builder builder(arrow_pool_.get()); for (int64_t r = overlap_start; r < overlap_end; r++) { @@ -256,10 +282,11 @@ Result> BlobFallbackBatchReader::AssembleRowIdRun( pieces.push_back(std::move(piece)); } } - pos += chunk.length; - if (pos >= run_end) { + if (pos + chunk.length > run_end) { break; } + cursor->chunk_start += chunk.length; + cursor->chunk_idx++; } if (pieces.size() == 1) { return pieces[0]; @@ -274,6 +301,7 @@ Result> BlobFallbackBatchReader::AssembleColumn( const std::vector>& group_chunks) const { const auto row_count = static_cast(group_choice.size()); arrow::ArrayVector pieces; + std::vector cursors(group_chunks.size()); int64_t run_start = 0; while (run_start < row_count) { const int32_t group = group_choice[run_start]; @@ -286,8 +314,9 @@ Result> BlobFallbackBatchReader::AssembleColumn( // (taken from the newest group, which steps in lockstep), reports -1 as its // sequence number, and returns null for every other field if (field_idx == row_id_field_idx_) { - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr row_id_piece, - AssembleRowIdRun(group_chunks[0], run_start, run_end)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr row_id_piece, + AssembleRowIdRun(group_chunks[0], run_start, run_end, &cursors[0])); pieces.push_back(std::move(row_id_piece)); } else if (field_idx == seq_num_field_idx_) { arrow::Int64Scalar seq_scalar(-1); @@ -303,23 +332,29 @@ Result> BlobFallbackBatchReader::AssembleColumn( pieces.push_back(std::move(null_piece)); } } else { - int64_t pos = 0; - for (const auto& chunk : group_chunks[group]) { + ChunkCursor& cursor = cursors[group]; + const auto& chunks = group_chunks[group]; + while (cursor.chunk_idx < chunks.size() && cursor.chunk_start < run_end) { + const auto& chunk = chunks[cursor.chunk_idx]; + const int64_t pos = cursor.chunk_start; int64_t overlap_start = std::max(run_start, pos); int64_t overlap_end = std::min(run_end, pos + chunk.length); if (overlap_start < overlap_end) { if (chunk.array == nullptr) { return Status::Invalid( - "Unexpected: a gap row was chosen as a blob fallback result."); + "Unexpected: a gap or skipped row was chosen as a blob fallback " + "result."); } std::shared_ptr column = chunk.array->field(field_idx); - pieces.push_back(column->Slice(chunk.offset + (overlap_start - pos), - overlap_end - overlap_start)); + pieces.push_back( + column->Slice(overlap_start - pos, overlap_end - overlap_start)); } - pos += chunk.length; - if (pos >= run_end) { + // Keep a partially consumed chunk for a later run choosing the same group. + if (pos + chunk.length > run_end) { break; } + cursor.chunk_start += chunk.length; + cursor.chunk_idx++; } } run_start = run_end; @@ -337,43 +372,38 @@ Result BlobFallbackBatchReader::NextBatch() { return Status::Invalid("blob fallback batch reader is closed"); } std::vector> group_chunks(groups_.size()); - // TODO(xinyu.lxy): Older layers are materialized even when newer values cover them, adding - // unnecessary reads and potentially exposing stale payload errors. Add Java-like read/skip - // support to materialize older values only when fallback is needed. - // the first (newest) group defines the window; the others must step in lockstep - PAIMON_ASSIGN_OR_RAISE(int64_t row_count, FillWindow(0, read_batch_size_, &group_chunks[0])); - for (size_t g = 1; g < groups_.size(); g++) { - // ask for one row even when the first group is exhausted, so that a longer group is - // reported as a misalignment instead of silently truncating the read - const int64_t want = std::max(row_count, 1); - PAIMON_ASSIGN_OR_RAISE(int64_t got, FillWindow(g, want, &group_chunks[g])); - if (got != row_count) { - return Status::Invalid(fmt::format( - "All sequence groups of a blob fallback read should have the same number of " - "rows: group {} yielded {} rows in a window of {}", - g, got, row_count)); - } - } - if (row_count == 0) { - return BatchReader::MakeEofBatch(); - } - - std::vector> placeholder_flags(groups_.size()); + // The newest group defines the window. Resolve rows before opening each older group's + // payloads, while still advancing every group by the same number of selected rows. + PAIMON_ASSIGN_OR_RAISE(int64_t row_count, + FillWindow(0, read_batch_size_, nullptr, &group_chunks[0])); + std::vector group_choice(row_count, -1); + const std::vector eof_choice(1, 0); for (size_t g = 0; g < groups_.size(); g++) { - PAIMON_ASSIGN_OR_RAISE(placeholder_flags[g], + if (g > 0) { + // At EOF of the newest group, check that older groups are also exhausted without + // reading payloads. + const int64_t want = std::max(row_count, 1); + PAIMON_ASSIGN_OR_RAISE(int64_t got, + FillWindow(g, want, row_count == 0 ? &eof_choice : &group_choice, + &group_chunks[g])); + if (got != row_count) { + return Status::Invalid(fmt::format( + "All sequence groups of a blob fallback read should have the same number of " + "rows: group {} yielded {} rows in a window of {}", + g, got, row_count)); + } + } + PAIMON_ASSIGN_OR_RAISE(std::vector placeholder_flags, ComputePlaceholderFlags(group_chunks[g], row_count)); - } - // per row, the first group in max-sequence order with a real entry wins; -1 means the row is - // a placeholder in every group - std::vector group_choice(row_count, -1); - for (int64_t r = 0; r < row_count; r++) { - for (size_t g = 0; g < groups_.size(); g++) { - if (!placeholder_flags[g][r]) { + for (int64_t r = 0; r < row_count; r++) { + if (group_choice[r] == -1 && !placeholder_flags[r]) { group_choice[r] = static_cast(g); - break; } } } + if (row_count == 0) { + return BatchReader::MakeEofBatch(); + } arrow::ArrayVector columns; columns.reserve(read_schema_->num_fields()); @@ -404,7 +434,6 @@ Result BlobFallbackBatchReader::NextBatchWithB void BlobFallbackBatchReader::Close() { for (auto& group : groups_) { - group.pending.clear(); for (auto& segment : group.segments) { if (segment.reader) { segment.reader->Close(); diff --git a/src/paimon/common/reader/blob_fallback_batch_reader.h b/src/paimon/common/reader/blob_fallback_batch_reader.h index 36bceb990..0967bbe22 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader.h +++ b/src/paimon/common/reader/blob_fallback_batch_reader.h @@ -19,11 +19,11 @@ #pragma once -#include #include #include #include "arrow/type.h" +#include "paimon/common/reader/blob_read_context.h" #include "paimon/memory/memory_pool.h" #include "paimon/reader/batch_reader.h" #include "paimon/reader/file_batch_reader.h" @@ -57,6 +57,8 @@ namespace paimon { /// placeholder. The blob format reader emits placeholders as BlobDefs::kPlaceholderSentinel /// bytes for scalar BLOBs, as a one-element sentinel array for ARRAY, or as a two-entry /// map with duplicate keys for MAP<..., BLOB>. +/// Older groups use a shared BlobReadContext to advance resolved rows without reading their +/// payloads and to bound each batch by the current window and unresolved row runs. /// 4. A row that is a placeholder in every group degrades to a null blob: it keeps its /// _ROW_ID, reports -1 as its _SEQUENCE_NUMBER, and returns null for every other field. class BlobFallbackBatchReader : public BatchReader { @@ -70,6 +72,8 @@ class BlobFallbackBatchReader : public BatchReader { /// Gap segment only: the selected row ids the gap emits (all placeholders), as sorted /// disjoint ranges. Must not be empty for a gap segment. std::vector gap_selected_ranges; + /// File segment only: shared with the bottom blob reader before any per-file wrapping. + std::shared_ptr blob_context = nullptr; }; /// `sequence_groups` must be ordered by descending max sequence number and contain at least @@ -92,18 +96,22 @@ class BlobFallbackBatchReader : public BatchReader { std::shared_ptr GetReaderMetrics() const override; private: - /// A run of consecutive rows already fetched from one group: `array` rows - /// [offset, offset + length) for a file segment, or `length` placeholder rows for a gap - /// segment (array is null). + /// A run of consecutive rows already fetched from one group: the whole `array` for a file + /// segment, or `length` placeholder rows for a gap or an already resolved run (array is null). struct Chunk { std::shared_ptr array; - int64_t offset = 0; int64_t length = 0; - /// Gap chunk only: the row id of each of the `length` rows, filled when the read - /// schema contains _ROW_ID so all-placeholder rows can keep their row id. + /// Newest-group gap chunks only: one row id per row, filled when _ROW_ID is requested. + /// Empty for older groups and skipped runs. std::vector gap_row_ids; }; + /// Forward-only position in one group's chunks during assembly of a single output column. + struct ChunkCursor { + size_t chunk_idx = 0; + int64_t chunk_start = 0; + }; + /// Read progress of one sequence group. Move-only, matching Segment's unique_ptr member. struct GroupCursor { GroupCursor() = default; @@ -118,9 +126,6 @@ class BlobFallbackBatchReader : public BatchReader { /// number of rows already emitted from that range. size_t gap_range_idx = 0; int64_t gap_range_pos = 0; - /// Rows fetched from the current file segment but not yet consumed. - std::deque> pending; - int64_t pending_pos = 0; }; BlobFallbackBatchReader(std::vector&& groups, @@ -131,7 +136,16 @@ class BlobFallbackBatchReader : public BatchReader { /// Collects up to `want` rows from the group into chunks. Only the first group may come up /// short (which defines the window size); any later group ending early is a misalignment. - Result FillWindow(size_t group_idx, int64_t want, std::vector* chunks); + /// When `group_choice` is set, only rows still choosing -1 need to be materialized. Already + /// resolved runs advance without payload reads and are represented by unused placeholder + /// chunks. Every batch must fit in its requested run and is consumed completely. + Result FillWindow(size_t group_idx, int64_t want, + const std::vector* group_choice, + std::vector* chunks); + + /// Applies a blob-only request for the complete outer call. + static Result ReadSegmentBatch(const Segment& segment, int32_t max_rows, + bool read_payload); /// Flags each of the `row_count` window rows of the given chunks as placeholder or not. Result> ComputePlaceholderFlags(const std::vector& chunks, @@ -144,11 +158,12 @@ class BlobFallbackBatchReader : public BatchReader { int32_t field_idx, const std::vector& group_choice, const std::vector>& group_chunks) const; - /// Assembles the _ROW_ID values of rows [run_start, run_end) from the given group's chunks; - /// gap chunks contribute their synthesized row ids. + /// Assembles _ROW_ID values of rows [run_start, run_end) from the newest group's chunks. + /// Requires _ROW_ID in the read schema. This group has no skipped runs, so every null-array + /// chunk is a gap with complete gap_row_ids. Shares its cursor with runs choosing group 0. Result> AssembleRowIdRun(const std::vector& chunks, - int64_t run_start, - int64_t run_end) const; + int64_t run_start, int64_t run_end, + ChunkCursor* cursor) const; std::vector groups_; std::shared_ptr read_schema_; diff --git a/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp b/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp index 9f67737de..a27ccd5b0 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp +++ b/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp @@ -19,18 +19,24 @@ #include "paimon/common/reader/blob_fallback_batch_reader.h" +#include #include #include #include #include "arrow/api.h" +#include "arrow/c/bridge.h" #include "arrow/util/range.h" #include "gtest/gtest.h" #include "paimon/common/data/blob_defs.h" #include "paimon/common/data/blob_utils.h" +#include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h" #include "paimon/common/table/special_fields.h" #include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/checked_cast.h" +#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h" +#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h" #include "paimon/memory/memory_pool.h" #include "paimon/testing/mock/mock_file_batch_reader.h" #include "paimon/testing/utils/read_result_collector.h" @@ -42,6 +48,96 @@ namespace paimon::test { /// reader), std::nullopt for null. using BlobRows = std::vector>; +struct BlobReadCounts { + std::vector materialized; + std::vector skipped; +}; + +class ReadSkipBlobReader : public FileBatchReader { + public: + ReadSkipBlobReader(const std::shared_ptr& data, + const std::shared_ptr& schema, int32_t batch_size, + const std::shared_ptr& context, + const std::shared_ptr& counts = nullptr) + : data_(data), + schema_(schema), + context_(context), + counts_(counts), + batch_size_(batch_size) {} + + Result> GetFileSchema() const override { + auto c_schema = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportType(*schema_, c_schema.get())); + return c_schema; + } + + Status SetReadSchema(::ArrowSchema*, const std::shared_ptr&, + const std::optional&) override { + return Status::NotImplemented("Read/skip test reader does not support schema changes"); + } + + Result NextBatch() override { + if (current_pos_ == data_->length()) { + previous_batch_row_count_ = 0; + return BatchReader::MakeEofBatch(); + } + int32_t max_rows = context_->request ? context_->request->max_rows : batch_size_; + bool read_payload = !context_->request || context_->request->read_payload; + int32_t take = static_cast( + std::min({batch_size_, max_rows, data_->length() - current_pos_})); + std::shared_ptr array; + if (read_payload) { + array = data_->Slice(current_pos_, take); + } else { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(array, arrow::MakeArrayOfNull(schema_, take)); + } + if (counts_) { + std::vector& rows = read_payload ? counts_->materialized : counts_->skipped; + for (int32_t i = 0; i < take; i++) { + rows.push_back(current_pos_ + i); + } + } + previous_batch_first_row_num_ = current_pos_; + previous_batch_row_count_ = take; + current_pos_ += take; + auto c_array = std::make_unique(); + auto c_schema = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, c_array.get(), c_schema.get())); + return std::make_pair(std::move(c_array), std::move(c_schema)); + } + + Result GetPreviousBatchFileRowId(uint64_t batch_row_id) const override { + if (batch_row_id >= previous_batch_row_count_) { + return Status::Invalid("No matching row in the previous batch"); + } + return previous_batch_first_row_num_ + batch_row_id; + } + + Result GetNumberOfRows() const override { + return data_->length(); + } + + bool SupportPreciseBitmapSelection() const override { + return false; + } + + std::shared_ptr GetReaderMetrics() const override { + return nullptr; + } + + void Close() override {} + + private: + std::shared_ptr data_; + std::shared_ptr schema_; + std::shared_ptr context_; + std::shared_ptr counts_; + int32_t batch_size_; + int64_t current_pos_ = 0; + uint64_t previous_batch_first_row_num_ = 0; + uint64_t previous_batch_row_count_ = 0; +}; + class BlobFallbackBatchReaderTest : public ::testing::Test { public: void SetUp() override { @@ -92,14 +188,18 @@ class BlobFallbackBatchReaderTest : public ::testing::Test { } }; - std::vector MakeGroup(const std::vector& specs, - int32_t file_batch_size) const { + std::vector MakeGroup( + const std::vector& specs, int32_t file_batch_size, + const std::shared_ptr& counts = nullptr) const { std::vector segments; for (const auto& spec : specs) { if (spec.file_rows) { - auto reader = std::make_unique(MakeBlobStruct(*spec.file_rows), - struct_type_, file_batch_size); - segments.push_back(BlobFallbackBatchReader::Segment{std::move(reader), {}}); + auto context = std::make_shared(); + auto reader = std::make_unique(MakeBlobStruct(*spec.file_rows), + struct_type_, file_batch_size, + context, counts); + segments.push_back( + BlobFallbackBatchReader::Segment{std::move(reader), {}, std::move(context)}); } else { segments.push_back(BlobFallbackBatchReader::Segment{nullptr, spec.gap_ranges}); } @@ -145,6 +245,125 @@ TEST_F(BlobFallbackBatchReaderTest, TestBasicFallback) { {"b0", "u1", "b2"}); } +TEST_F(BlobFallbackBatchReaderTest, TestAlternatingUpdates) { + CheckFallback({{SegmentSpec::File({"u0", "PH", "u2", "PH", "u4", "PH", "u6", "PH"})}, + {SegmentSpec::File({"b0", "b1", "b2", "b3", "b4", "b5", "b6", "b7"})}}, + {"u0", "b1", "u2", "b3", "u4", "b5", "u6", "b7"}); +} + +TEST_F(BlobFallbackBatchReaderTest, TestSkipsFullyCoveredLayersIncludingNull) { + const BlobRows newest = {"u0", std::nullopt, "u2"}; + for (int32_t batch_size : {1, 2, 1024}) { + for (int32_t file_batch_size : {1, 3, 1024}) { + auto counts = std::make_shared(); + std::vector> groups; + groups.push_back(MakeGroup({SegmentSpec::File(newest)}, file_batch_size)); + groups.push_back( + MakeGroup({SegmentSpec::File({"b0", "b1", "b2"})}, file_batch_size, counts)); + ASSERT_OK_AND_ASSIGN(auto reader, + BlobFallbackBatchReader::Create(std::move(groups), read_schema_, + batch_size, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto result, + ReadResultCollector::CollectResult(std::move(reader))); + ASSERT_TRUE( + result->Equals(std::make_shared(MakeBlobStruct(newest)))); + ASSERT_TRUE(counts->materialized.empty()); + ASSERT_EQ(counts->skipped, (std::vector{0, 1, 2})); + } + } +} + +TEST_F(BlobFallbackBatchReaderTest, TestReadSkipRunsAcrossWindowsAndThreeLayers) { + for (int32_t batch_size : {1, 2, 4, 1024}) { + for (int32_t file_batch_size : {1, 3, 1024}) { + auto middle_counts = std::make_shared(); + auto oldest_counts = std::make_shared(); + std::vector> groups; + groups.push_back(MakeGroup({SegmentSpec::File({"u0", "PH", "PH", "u3", "PH", "PH"})}, + file_batch_size)); + groups.push_back( + MakeGroup({SegmentSpec::File({"stale", "m1", "PH", "stale", "PH", "m5"})}, + file_batch_size, middle_counts)); + groups.push_back( + MakeGroup({SegmentSpec::File({"stale", "stale", "o2", "stale", "o4", "stale"})}, + file_batch_size, oldest_counts)); + ASSERT_OK_AND_ASSIGN(auto reader, + BlobFallbackBatchReader::Create(std::move(groups), read_schema_, + batch_size, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto result, + ReadResultCollector::CollectResult(std::move(reader))); + ASSERT_TRUE(result->Equals(std::make_shared( + MakeBlobStruct({"u0", "m1", "o2", "u3", "o4", "m5"})))); + ASSERT_EQ(middle_counts->materialized, (std::vector{1, 2, 4, 5})); + ASSERT_EQ(middle_counts->skipped, (std::vector{0, 3})); + ASSERT_EQ(oldest_counts->materialized, (std::vector{2, 4})); + ASSERT_EQ(oldest_counts->skipped, (std::vector{0, 1, 3, 5})); + } + } +} + +TEST_F(BlobFallbackBatchReaderTest, TestSkippedGroupsStillDetectMisalignment) { + for (const BlobRows& older : {BlobRows{"b0"}, BlobRows{"b0", "b1", "b2"}}) { + auto counts = std::make_shared(); + std::vector> groups; + groups.push_back(MakeGroup({SegmentSpec::File({"u0", "u1"})}, 1024)); + groups.push_back(MakeGroup({SegmentSpec::File(older)}, 1024, counts)); + ASSERT_OK_AND_ASSIGN(auto reader, + BlobFallbackBatchReader::Create(std::move(groups), read_schema_, 2, + GetArrowPool(pool_))); + ASSERT_NOK_WITH_MSG(ReadResultCollector::CollectResult(std::move(reader)), + "same number of rows"); + ASSERT_TRUE(counts->materialized.empty()); + ASSERT_EQ(counts->skipped.size(), older.size()); + } +} + +TEST_F(BlobFallbackBatchReaderTest, TestRejectsBatchExceedingReadLimit) { + for (bool covered : {false, true}) { + auto context = std::make_shared(); + // This mock ignores the context and returns two rows for a one-row read/skip request. + auto oversized_reader = + std::make_unique(MakeBlobStruct({"b0", "b1"}), struct_type_, 1024); + std::vector> groups; + groups.push_back(MakeGroup( + {SegmentSpec::File(covered ? BlobRows{"u0", "u1"} : BlobRows{"PH", "PH"})}, 1024)); + std::vector older; + older.push_back({std::move(oversized_reader), {}, context}); + groups.push_back(std::move(older)); + ASSERT_OK_AND_ASSIGN(auto reader, + BlobFallbackBatchReader::Create(std::move(groups), read_schema_, 1, + GetArrowPool(pool_))); + ASSERT_NOK_WITH_MSG(reader->NextBatch(), "exceeding requested limit 1"); + ASSERT_FALSE(context->request.has_value()); + } +} + +TEST_F(BlobFallbackBatchReaderTest, TestSkipRequestsPreserveBitmapAndDvFiltering) { + auto counts = std::make_shared(); + auto context = std::make_shared(); + auto raw = std::make_unique(MakeBlobStruct({"b0", "b1", "b2", "b3", "b4"}), + struct_type_, 1024, context, counts); + auto bitmap_reader = std::make_unique( + std::move(raw), RoaringBitmap32::From({1, 2, 3, 4})); + auto dv = std::make_shared(); + ASSERT_OK(dv->Delete(2)); + auto filtered_reader = + std::make_unique(std::move(bitmap_reader), dv); + std::vector> groups; + groups.push_back(MakeGroup({SegmentSpec::File({"u1", "u3", "PH"})}, 1024)); + std::vector older; + older.push_back({std::move(filtered_reader), {}, context}); + groups.push_back(std::move(older)); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFallbackBatchReader::Create( + std::move(groups), read_schema_, 3, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto result, ReadResultCollector::CollectResult(std::move(reader))); + ASSERT_TRUE( + result->Equals(std::make_shared(MakeBlobStruct({"u1", "u3", "b4"})))); + ASSERT_EQ(counts->skipped, (std::vector{0, 1, 2, 3})); + ASSERT_EQ(counts->materialized, (std::vector{4})); + ASSERT_FALSE(context->request.has_value()); +} + TEST_F(BlobFallbackBatchReaderTest, TestSingleLayerFillsUnwrittenRows) { CheckFallback({{SegmentSpec::Gap(2), SegmentSpec::File({"b2", "PH"}), SegmentSpec::Gap(2)}}, {std::nullopt, std::nullopt, "b2", std::nullopt, std::nullopt, std::nullopt}); @@ -275,22 +494,38 @@ TEST_F(BlobFallbackBatchReaderTest, TestRowTrackingFieldsPreserved) { return array; }; - for (auto batch_size : arrow::internal::Iota(1, 5)) { - for (auto file_batch_size : {1, 1024}) { - // newest layer (seq 20) covers only row 2; rows 0-1 are a gap + for (auto batch_size : arrow::internal::Iota(1, 9)) { + for (auto file_batch_size : {1, 3, 1024}) { + auto new_context = std::make_shared(); + auto old_context = std::make_shared(); + // Rows 0-1 are a gap. Inside one file chunk, alternate newest-layer values + // with placeholders so normal and all-placeholder row-id runs share it. std::vector newest; newest.push_back(BlobFallbackBatchReader::Segment{nullptr, {Range(0, 1)}}); newest.push_back(BlobFallbackBatchReader::Segment{ - std::make_unique(make_rows({{"u2", 2, 20}}), - struct_type, file_batch_size), - {}}); - // oldest layer (seq 10) covers rows 0-2, row 1 is a placeholder there too + std::make_unique(make_rows({{"u2", 2, 20}, + {"PH", 3, 20}, + {"u4", 4, 20}, + {"PH", 5, 20}, + {"u6", 6, 20}}), + struct_type, file_batch_size, + new_context), + {}, + new_context}); + // Rows 1 and 3 are placeholders in both layers; row 5 falls back. std::vector oldest; oldest.push_back(BlobFallbackBatchReader::Segment{ - std::make_unique( - make_rows({{"b0", 0, 10}, {"PH", 1, 10}, {"PH", 2, 10}}), struct_type, - file_batch_size), - {}}); + std::make_unique(make_rows({{"b0", 0, 10}, + {"PH", 1, 10}, + {"PH", 2, 10}, + {"PH", 3, 10}, + {"PH", 4, 10}, + {"b5", 5, 10}, + {"PH", 6, 10}}), + struct_type, file_batch_size, + old_context), + {}, + old_context}); std::vector> groups; groups.push_back(std::move(newest)); groups.push_back(std::move(oldest)); @@ -302,10 +537,13 @@ TEST_F(BlobFallbackBatchReaderTest, TestRowTrackingFieldsPreserved) { auto result, paimon::test::ReadResultCollector::CollectResult(std::move(reader))); - // row 0 falls back to seq 10, row 1 is all-placeholder (null blob, row id - // kept, seq -1), row 2 takes seq 20 - auto expected_array = - make_rows({{"b0", 0, 10}, {std::nullopt, 1, -1}, {"u2", 2, 20}}); + auto expected_array = make_rows({{"b0", 0, 10}, + {std::nullopt, 1, -1}, + {"u2", 2, 20}, + {std::nullopt, 3, -1}, + {"u4", 4, 20}, + {"b5", 5, 10}, + {"u6", 6, 20}}); auto expected_chunk_array = std::make_shared(expected_array); ASSERT_TRUE(result->Equals(expected_chunk_array)) @@ -353,6 +591,14 @@ TEST_F(BlobFallbackBatchReaderTest, TestCreateValidation) { 1024, GetArrowPool(pool_)), "should not be empty"); + // File segments must share a context so their readers honor the read/skip row limits. + std::vector> without_context; + without_context.push_back(MakeGroup({SegmentSpec::File({"b0"})}, 1024)); + without_context[0][0].blob_context.reset(); + ASSERT_NOK_WITH_MSG(BlobFallbackBatchReader::Create(std::move(without_context), read_schema_, + 1024, GetArrowPool(pool_)), + "file segment needs a blob read context"); + // a gap segment must cover at least one selected row id std::vector> with_empty_gap; with_empty_gap.push_back(MakeGroup({SegmentSpec::File({"b0"})}, 1024)); diff --git a/src/paimon/common/reader/blob_read_context.h b/src/paimon/common/reader/blob_read_context.h new file mode 100644 index 000000000..57fc01141 --- /dev/null +++ b/src/paimon/common/reader/blob_read_context.h @@ -0,0 +1,56 @@ +/* + * 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 "paimon/visibility.h" + +namespace paimon { + +/// Bounds one blob batch and controls whether its values need to be materialized. +struct BlobBatchRequest { + /// Upper bound on selected rows before outer deletion-vector filtering. Must be positive + /// when the request is issued. + int32_t max_rows = 0; + /// False advances rows without reading their payloads. + bool read_payload = true; +}; + +/// Shared by one file's blob reader and its fallback segment. A request stays active for the +/// entire outer NextBatchWithBitmap() call, including retries for fully deleted batches. +/// Access is confined to the thread reading that segment. +struct BlobReadContext { + std::optional request; +}; + +/// Internal builder hook used before the format reader is wrapped. Only the blob builder +/// implements it; neither ReaderBuilder nor the wrapping readers need blob-specific methods. +class PAIMON_EXPORT BlobReadContextAwareBuilder { + public: + virtual ~BlobReadContextAwareBuilder() = default; + + /// Shares a file's request context with the blob reader created by this builder. + virtual void SetBlobReadContext(const std::shared_ptr& context) = 0; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index d28a4ed1c..c5298e2c6 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -40,6 +40,7 @@ #include "paimon/common/executor/future.h" #include "paimon/common/executor/reader_build_executor.h" #include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h" +#include "paimon/common/reader/blob_read_context.h" #include "paimon/common/reader/complete_index_score_file_batch_reader.h" #include "paimon/common/reader/data_file_reader_factory.h" #include "paimon/common/reader/delegating_prefetch_reader.h" @@ -104,7 +105,8 @@ Result>> AbstractSplitRead::CreateR PAIMON_ASSIGN_OR_RAISE( std::vector readers_with_meta, CreateRawFileReadersWithMeta(partition, data_files, read_schema, predicate, dv_factory, - row_ranges, data_file_path_factory, extra_format_options)); + row_ranges, data_file_path_factory, extra_format_options, + /*attach_blob_context=*/false)); std::vector> raw_file_readers; raw_file_readers.reserve(readers_with_meta.size()); for (auto& reader_with_meta : readers_with_meta) { @@ -119,7 +121,8 @@ AbstractSplitRead::CreateRawFileReadersWithMeta( 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, + bool attach_blob_context) const { if (data_files.empty()) { return std::vector(); } @@ -134,11 +137,12 @@ AbstractSplitRead::CreateRawFileReadersWithMeta( if (parallel_num <= 1) { for (const auto& file : data_files) { PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr file_reader, + RawFileReaderWithMeta 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({file, std::move(file_reader)}); + row_ranges, data_file_path_factory, extra_format_options, + attach_blob_context)); + if (file_reader.reader) { + raw_file_readers.push_back(std::move(file_reader)); } } return std::move(raw_file_readers); @@ -148,31 +152,30 @@ AbstractSplitRead::CreateRawFileReadersWithMeta( // I/O, so files are built concurrently. The tasks only reference locals of this frame, // which stay alive because CollectAll below drains every future before returning. std::shared_ptr build_executor = GetReaderBuildExecutor(); - std::vector>>> futures; + std::vector>> futures; futures.reserve(data_files.size()); for (const auto& file : data_files) { - 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); - })); + futures.push_back( + Via(build_executor.get(), + [this, file, &partition, &field_mapping_builder, &dv_factory, &row_ranges, + &data_file_path_factory, &extra_format_options, attach_blob_context]() { + return CreateRawFileReader(partition, file, field_mapping_builder.get(), + dv_factory, row_ranges, data_file_path_factory, + extra_format_options, attach_blob_context); + })); } // Readers keep the file order of the split, and CollectAll preserves the submit order. Status first_error; - size_t file_index = 0; for (auto& built : CollectAll(futures)) { - const std::shared_ptr& file = data_files[file_index++]; if (!built.ok()) { if (first_error.ok()) { first_error = built.status(); } continue; } - std::unique_ptr file_reader = std::move(built).value(); - if (file_reader) { - raw_file_readers.push_back({file, std::move(file_reader)}); + RawFileReaderWithMeta file_reader = std::move(built).value(); + if (file_reader.reader) { + raw_file_readers.push_back(std::move(file_reader)); } } PAIMON_RETURN_NOT_OK(first_error); @@ -277,19 +280,33 @@ Result> AbstractSplitRead::ApplyBitmapSelection return std::move(file_reader); } -Result> AbstractSplitRead::CreateRawFileReader( +Result AbstractSplitRead::CreateRawFileReader( const BinaryRow& partition, const std::shared_ptr& file, const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory, - const std::map& extra_format_options) const { + const std::map& extra_format_options, + bool attach_blob_context) const { auto data_file_path = data_file_path_factory->ToPath(file); PAIMON_ASSIGN_OR_RAISE(std::string data_file_identifier, file->FileFormat()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader_builder, PrepareReaderBuilder(data_file_identifier, extra_format_options)); - return CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder), - field_mapping_builder, dv_factory, row_ranges, - data_file_path_factory); + std::shared_ptr blob_context; + if (attach_blob_context) { + auto* blob_builder = dynamic_cast(reader_builder.get()); + if (blob_builder == nullptr) { + return Status::Invalid("Blob fallback requires a context-aware reader builder."); + } + // Attach before wrapping so the bottom blob reader receives the per-file context. + blob_context = std::make_shared(); + blob_builder->SetBlobReadContext(blob_context); + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr reader, + CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder), + field_mapping_builder, dv_factory, row_ranges, + data_file_path_factory)); + return RawFileReaderWithMeta{file, std::move(reader), std::move(blob_context)}; } bool AbstractSplitRead::NeedCompleteRowTrackingFields( diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h index 0fd840b50..a71a60594 100644 --- a/src/paimon/core/operation/abstract_split_read.h +++ b/src/paimon/core/operation/abstract_split_read.h @@ -59,6 +59,7 @@ class FileStorePathFactory; class InternalReadContext; class MemoryPool; class Predicate; +struct BlobReadContext; struct DataFileMeta; class TableSchema; @@ -81,6 +82,7 @@ class AbstractSplitRead : public SplitRead { struct RawFileReaderWithMeta { std::shared_ptr file; std::unique_ptr reader; + std::shared_ptr blob_context = nullptr; }; AbstractSplitRead(const std::shared_ptr& path_factory, @@ -90,14 +92,16 @@ class AbstractSplitRead : public SplitRead { const std::shared_ptr& executor); /// Creates raw readers while preserving the metadata associated with every reader which was - /// not eliminated by a file index or deletion vector. + /// not eliminated by a file index or deletion vector. Set `attach_blob_context` only for + /// data-evolution blob fallback readers. Result> CreateRawFileReadersWithMeta( 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, + bool attach_blob_context) const; Result> ApplyPredicateFilterIfNeeded( std::unique_ptr&& reader, const std::shared_ptr& predicate) const; @@ -131,13 +135,14 @@ class AbstractSplitRead : public SplitRead { /// that the reader builder and the file reader cannot be built from different answers. DataFileReadOptions BuildDataFileReadOptions() const; - // return nullptr if data file is skipped by index or dv - Result> CreateRawFileReader( + // Return metadata with a null reader if the data file is skipped by index or dv. + Result CreateRawFileReader( const BinaryRow& partition, const std::shared_ptr& file, const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory, - const std::map& extra_format_options) const; + const std::map& extra_format_options, + bool attach_blob_context) const; Result> PrepareReaderBuilder( const std::string& format_identifier, diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 87dc99ccd..772e63878 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -1279,16 +1279,19 @@ Result> DataEvolutionSplitRead::CreateBlobFallbackR if (!SelectedRangesInRange(first_row_id, first_row_id + file->row_count - 1, file_ranges) .empty()) { - PAIMON_ASSIGN_OR_RAISE( - std::vector> file_readers, - CreateRawFileReaders(partition, {file}, file_read_schema, - /*predicate=*/nullptr, group_dv_factory, file_ranges, - data_file_path_factory, blob_format_options)); + PAIMON_ASSIGN_OR_RAISE(std::vector file_readers, + CreateRawFileReadersWithMeta( + partition, {file}, file_read_schema, + /*predicate=*/nullptr, group_dv_factory, file_ranges, + data_file_path_factory, blob_format_options, + /*attach_blob_context=*/true)); if (file_readers.size() != 1) { return Status::Invalid("Unexpected: blob fallback file reader was skipped."); } segments.push_back( - BlobFallbackBatchReader::Segment{std::move(file_readers[0]), {}}); + BlobFallbackBatchReader::Segment{std::move(file_readers[0].reader), + {}, + std::move(file_readers[0].blob_context)}); } next_row_id = std::max(next_row_id, first_row_id + file->row_count); } diff --git a/src/paimon/core/operation/merge_file_split_read.cpp b/src/paimon/core/operation/merge_file_split_read.cpp index 2deb74ec0..8a256bc0b 100644 --- a/src/paimon/core/operation/merge_file_split_read.cpp +++ b/src/paimon/core/operation/merge_file_split_read.cpp @@ -828,7 +828,7 @@ Result> MergeFileSplitRead::CreateReaderFo std::vector raw_file_readers, CreateRawFileReadersWithMeta(partition, data_files, read_schema_, predicate, dv_factory, /*row_ranges=*/{}, data_file_path_factory, - /*extra_format_options=*/{})); + /*extra_format_options=*/{}, /*attach_blob_context=*/false)); // KeyValueDataFileRecordReader converts arrow array from format reader to KeyValue objects std::vector> file_record_readers; diff --git a/src/paimon/format/blob/blob_file_batch_reader.cpp b/src/paimon/format/blob/blob_file_batch_reader.cpp index 5856e88df..b7c1b961a 100644 --- a/src/paimon/format/blob/blob_file_batch_reader.cpp +++ b/src/paimon/format/blob/blob_file_batch_reader.cpp @@ -305,6 +305,7 @@ Status AppendMapBlobKey(const std::shared_ptr& key_type, const Result> BlobFileBatchReader::Create( const std::shared_ptr& input_stream, int32_t batch_size, bool blob_as_descriptor, bool emit_placeholder_sentinel, const std::shared_ptr& pool, + const std::shared_ptr& blob_context, const std::shared_ptr& arrow_pool) { if (input_stream == nullptr) { return Status::Invalid("blob file batch reader create failed: input stream is nullptr"); @@ -356,9 +357,9 @@ Result> BlobFileBatchReader::Create( } } PAIMON_ASSIGN_OR_RAISE(std::string file_path, input_stream->GetUri()); - auto reader = std::unique_ptr( - new BlobFileBatchReader(input_stream, file_path, blob_lengths, blob_offsets, batch_size, - blob_as_descriptor, emit_placeholder_sentinel, pool, arrow_pool)); + auto reader = std::unique_ptr(new BlobFileBatchReader( + input_stream, file_path, blob_lengths, blob_offsets, batch_size, blob_as_descriptor, + emit_placeholder_sentinel, pool, blob_context, arrow_pool)); return reader; } @@ -366,7 +367,8 @@ BlobFileBatchReader::BlobFileBatchReader( const std::shared_ptr& input_stream, const std::string& file_path, const std::vector& blob_lengths, const std::vector& blob_offsets, int32_t batch_size, bool blob_as_descriptor, bool emit_placeholder_sentinel, - const std::shared_ptr& pool, const std::shared_ptr& arrow_pool) + const std::shared_ptr& pool, const std::shared_ptr& blob_context, + const std::shared_ptr& arrow_pool) : input_stream_(input_stream), file_path_(file_path), all_blob_lengths_(blob_lengths), @@ -378,6 +380,7 @@ BlobFileBatchReader::BlobFileBatchReader( emit_placeholder_sentinel_(emit_placeholder_sentinel), pool_(pool), arrow_pool_(arrow_pool), + blob_context_(blob_context), metrics_(std::make_shared()) { target_blob_row_indexes_.resize(target_blob_lengths_.size()); std::iota(target_blob_row_indexes_.begin(), target_blob_row_indexes_.end(), 0); @@ -1193,6 +1196,12 @@ Result BlobFileBatchReader::NextBatch() { if (target_type_ == nullptr) { return Status::Invalid("target type is nullptr, call SetReadSchema first"); } + const std::optional request = + blob_context_ ? blob_context_->request : std::nullopt; + if (request && request->max_rows <= 0) { + return Status::Invalid("Blob read batch row limit must be positive"); + } + const bool read_payload = !request || request->read_payload; if (current_pos_ >= target_blob_lengths_.size()) { previous_batch_start_pos_ = target_blob_lengths_.size(); previous_batch_row_count_ = 0; @@ -1200,6 +1209,9 @@ Result BlobFileBatchReader::NextBatch() { } int32_t left_rows = target_blob_lengths_.size() - current_pos_; int32_t rows_to_read = std::min(left_rows, batch_size_); + if (request) { + rows_to_read = std::min(rows_to_read, request->max_rows); + } if (!emit_placeholder_sentinel_) { for (int32_t k = 0; k < rows_to_read; ++k) { if (IsTargetPlaceholder(current_pos_ + k)) { @@ -1211,8 +1223,15 @@ Result BlobFileBatchReader::NextBatch() { } } } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr blob_array, - BuildTargetArray(rows_to_read)); + std::shared_ptr blob_array; + if (read_payload) { + PAIMON_ASSIGN_OR_RAISE(blob_array, BuildTargetArray(rows_to_read)); + } else { + // Keep row positions available to the unchanged tracking and DV readers without + // reading scalar or collection payloads. Fallback never chooses these temporary values. + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + blob_array, arrow::MakeArrayOfNull(target_type_, rows_to_read, arrow_pool_.get())); + } std::unique_ptr c_array = std::make_unique(); std::unique_ptr c_schema = std::make_unique(); PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*blob_array, c_array.get(), c_schema.get())); diff --git a/src/paimon/format/blob/blob_file_batch_reader.h b/src/paimon/format/blob/blob_file_batch_reader.h index 372c742dd..3610d059f 100644 --- a/src/paimon/format/blob/blob_file_batch_reader.h +++ b/src/paimon/format/blob/blob_file_batch_reader.h @@ -28,6 +28,7 @@ #include "arrow/type.h" #include "fmt/format.h" #include "paimon/common/data/blob_defs.h" +#include "paimon/common/reader/blob_read_context.h" #include "paimon/fs/file_system.h" #include "paimon/memory/bytes.h" #include "paimon/predicate/predicate.h" @@ -99,6 +100,9 @@ namespace paimon::blob { /// /// Note: /// - Current version is 1. +/// +/// Batch prefetching is not supported. If added, it must honor BlobReadContext's row limit +/// and read/skip mode for each call. Skip results must not be reused for payload reads. class BlobFileBatchReader : public FileBatchReader { public: /// `emit_placeholder_sentinel` controls how placeholder entries (bin_length == @@ -111,6 +115,7 @@ class BlobFileBatchReader : public FileBatchReader { const std::shared_ptr& input_stream, int32_t batch_size, bool blob_as_descriptor, bool emit_placeholder_sentinel, const std::shared_ptr& pool, + const std::shared_ptr& blob_context, const std::shared_ptr& arrow_pool); Result> GetFileSchema() const override; @@ -176,6 +181,7 @@ class BlobFileBatchReader : public FileBatchReader { const std::vector& blob_offsets, int32_t batch_size, bool blob_as_descriptor, bool emit_placeholder_sentinel, const std::shared_ptr& pool, + const std::shared_ptr& blob_context, const std::shared_ptr& arrow_pool); Result GetOrCreateSerializedDescriptorSize() const; @@ -244,6 +250,7 @@ class BlobFileBatchReader : public FileBatchReader { const bool emit_placeholder_sentinel_; std::shared_ptr pool_; std::shared_ptr arrow_pool_; + std::shared_ptr blob_context_; mutable int64_t serialized_descriptor_size_ = -1; std::shared_ptr target_type_; diff --git a/src/paimon/format/blob/blob_file_batch_reader_test.cpp b/src/paimon/format/blob/blob_file_batch_reader_test.cpp index d51b44f25..014942b0d 100644 --- a/src/paimon/format/blob/blob_file_batch_reader_test.cpp +++ b/src/paimon/format/blob/blob_file_batch_reader_test.cpp @@ -30,6 +30,7 @@ #include "paimon/common/data/blob_utils.h" #include "paimon/common/reader/blob_fallback_batch_reader.h" #include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" #include "paimon/data/blob.h" #include "paimon/format/blob/blob_format_writer.h" #include "paimon/format/blob/blob_reader_builder.h" @@ -96,10 +97,10 @@ class BlobFileBatchReaderTest : public testing::Test, public ::testing::WithPara std::shared_ptr fs = std::make_shared(); ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/" + paimon_blob_file)); - ASSERT_OK_AND_ASSIGN(auto reader, - BlobFileBatchReader::Create( - input_stream, /*batch_size=*/1024, blob_as_descriptor, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + input_stream, /*batch_size=*/1024, blob_as_descriptor, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, selection_bitmap)); ASSERT_OK_AND_ASSIGN(auto chunked_array, paimon::test::ReadResultCollector::CollectResult(std::move(reader))); @@ -180,10 +181,11 @@ class BlobFileBatchReaderTest : public testing::Test, public ::testing::WithPara ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input, /*batch_size=*/1, /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/1, /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_NOK_WITH_MSG(reader->NextBatch(), expected_message); } @@ -201,10 +203,11 @@ class BlobFileBatchReaderTest : public testing::Test, public ::testing::WithPara ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input, /*batch_size=*/16, /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/true, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/16, /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/true, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_NOK_WITH_MSG(reader->NextBatch(), expected_message); } @@ -233,6 +236,70 @@ TEST_P(BlobFileBatchReaderTest, TestSimple) { {"blob_9_f54d253c.bin"}, blob_as_descriptor); } +TEST_P(BlobFileBatchReaderTest, ReadContextPreservesSelectionAndDefaultReads) { + auto schema = arrow::schema({BlobUtils::ToArrowField("blob_col", true)}); + const std::string path = paimon::test::GetDataDir() + + "/db_with_blob.db/table_with_blob/bucket-0/" + "data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob"; + auto file_system = std::make_shared(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(path)); + BlobReaderBuilder builder(1024, {{Options::BLOB_AS_DESCRIPTOR, GetParam() ? "true" : "false"}}); + builder.WithMemoryPool(pool_); + ReaderBuilder* generic_builder = &builder; + auto* context_builder = dynamic_cast(generic_builder); + ASSERT_NE(context_builder, nullptr); + auto context = std::make_shared(); + context_builder->SetBlobReadContext(context); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, builder.Build(input)); + auto set_selection = [&](const RoaringBitmap32& selection) { + ::ArrowSchema c_schema; + ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); + ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, selection)); + }; + std::shared_ptr blobs; + auto check_batch = [&](int64_t expected_rows, int64_t expected_nulls) { + ASSERT_OK_AND_ASSIGN(auto batch, reader->NextBatch()); + ASSERT_FALSE(BatchReader::IsEofBatch(batch)); + ASSERT_EQ(batch.first->length, expected_rows); + auto array = arrow::ImportArray(batch.first.get(), batch.second.get()); + ASSERT_TRUE(array.ok()); + auto struct_array = std::dynamic_pointer_cast(array.ValueOrDie()); + ASSERT_TRUE(struct_array); + blobs = std::dynamic_pointer_cast(struct_array->field(0)); + ASSERT_TRUE(blobs); + ASSERT_EQ(blobs->null_count(), expected_nulls); + }; + + ASSERT_NO_FATAL_FAILURE(set_selection(RoaringBitmap32::From({0, 2}))); + context->request = BlobBatchRequest{0, false}; + ASSERT_NOK_WITH_MSG(reader->NextBatch(), "positive"); + context->request = BlobBatchRequest{1, false}; + ASSERT_NO_FATAL_FAILURE(check_batch(1, /*expected_nulls=*/1)); + ASSERT_OK_AND_ASSIGN(uint64_t first_position, reader->GetPreviousBatchFileRowId(0)); + ASSERT_EQ(first_position, 0); + + context->request.reset(); + ASSERT_NO_FATAL_FAILURE(check_batch(1, /*expected_nulls=*/0)); + ASSERT_OK_AND_ASSIGN(uint64_t second_position, reader->GetPreviousBatchFileRowId(0)); + ASSERT_EQ(second_position, 2); + ASSERT_OK_AND_ASSIGN(std::string value, ReadBlobValue(blobs, 0, GetParam(), file_system)); + ASSERT_FALSE(value.empty()); + ASSERT_OK_AND_ASSIGN(auto eof, reader->NextBatch()); + ASSERT_TRUE(BatchReader::IsEofBatch(eof)); + ASSERT_NOK_WITH_MSG(reader->GetPreviousBatchFileRowId(0), "EOF"); + + ASSERT_NO_FATAL_FAILURE(set_selection(RoaringBitmap32::From({0, 2}))); + context->request = BlobBatchRequest{1, true}; + ASSERT_NO_FATAL_FAILURE(check_batch(1, /*expected_nulls=*/0)); + context->request = BlobBatchRequest{5, false}; + ASSERT_NO_FATAL_FAILURE(check_batch(1, /*expected_nulls=*/1)); + ASSERT_NO_FATAL_FAILURE(set_selection(RoaringBitmap32::From({0, 1, 2}))); + context->request = BlobBatchRequest{1, true}; + ASSERT_NO_FATAL_FAILURE(check_batch(1, /*expected_nulls=*/0)); + context->request.reset(); + ASSERT_NO_FATAL_FAILURE(check_batch(2, /*expected_nulls=*/0)); +} + TEST_P(BlobFileBatchReaderTest, TestMapBlob) { auto dir = paimon::test::UniqueTestDirectory::Create(); ASSERT_TRUE(dir); @@ -253,10 +320,11 @@ TEST_P(BlobFileBatchReaderTest, TestMapBlob) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system->Open(file_path)); const bool blob_as_descriptor = GetParam(); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input_stream, /*batch_size=*/2, blob_as_descriptor, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/2, blob_as_descriptor, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_OK_AND_ASSIGN(std::shared_ptr chunked_array, paimon::test::ReadResultCollector::CollectResult(std::move(reader))); @@ -310,10 +378,11 @@ TEST_P(BlobFileBatchReaderTest, TestArrayBlob) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); const bool blob_as_descriptor = GetParam(); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create(input, /*batch_size=*/2, blob_as_descriptor, - /*emit_placeholder_sentinel=*/true, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/2, blob_as_descriptor, + /*emit_placeholder_sentinel=*/true, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_OK_AND_ASSIGN(std::shared_ptr chunked_array, paimon::test::ReadResultCollector::CollectResult(std::move(reader))); @@ -407,27 +476,31 @@ TEST_P(BlobFileBatchReaderTest, MapBlobFallbackAcrossSequenceLayers) { auto map_schema = arrow::schema({map_field}); const bool blob_as_descriptor = GetParam(); + auto new_context = std::make_shared(); + auto old_context = std::make_shared(); ASSERT_OK_AND_ASSIGN(std::shared_ptr new_input, file_system->Open(new_file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr new_reader, - BlobFileBatchReader::Create( - new_input, /*batch_size=*/2, blob_as_descriptor, - /*emit_placeholder_sentinel=*/true, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr new_reader, + BlobFileBatchReader::Create(new_input, /*batch_size=*/2, blob_as_descriptor, + /*emit_placeholder_sentinel=*/true, pool_, new_context, + GetArrowPool(pool_))); ::ArrowSchema new_schema; ASSERT_TRUE(arrow::ExportSchema(*map_schema, &new_schema).ok()); ASSERT_OK(new_reader->SetReadSchema(&new_schema, nullptr, std::nullopt)); ASSERT_OK_AND_ASSIGN(std::shared_ptr old_input, file_system->Open(old_file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr old_reader, - BlobFileBatchReader::Create( - old_input, /*batch_size=*/2, blob_as_descriptor, - /*emit_placeholder_sentinel=*/true, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr old_reader, + BlobFileBatchReader::Create(old_input, /*batch_size=*/2, blob_as_descriptor, + /*emit_placeholder_sentinel=*/true, pool_, old_context, + GetArrowPool(pool_))); ::ArrowSchema old_schema; ASSERT_TRUE(arrow::ExportSchema(*map_schema, &old_schema).ok()); ASSERT_OK(old_reader->SetReadSchema(&old_schema, nullptr, std::nullopt)); std::vector> groups(2); - groups[0].push_back({std::move(new_reader), {}}); // NOLINT(modernize-use-emplace) - groups[1].push_back({std::move(old_reader), {}}); // NOLINT(modernize-use-emplace) + groups[0].push_back({std::move(new_reader), {}, new_context}); // NOLINT(modernize-use-emplace) + groups[1].push_back({std::move(old_reader), {}, old_context}); // NOLINT(modernize-use-emplace) ASSERT_OK_AND_ASSIGN( std::unique_ptr fallback, BlobFallbackBatchReader::Create(std::move(groups), map_schema, /*read_batch_size=*/2, @@ -456,6 +529,122 @@ TEST_P(BlobFileBatchReaderTest, MapBlobFallbackAcrossSequenceLayers) { ASSERT_EQ("world", last_value); } +TEST_P(BlobFileBatchReaderTest, CollectionBlobFallbackWithCoveredRows) { + auto dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto file_system = std::make_shared(); + const std::string new_path = dir->Str() + "/new.blob"; + const std::string old_path = dir->Str() + "/old.blob"; + auto scalar_field = BlobUtils::ToArrowField("blob_col", true); + auto scalar_type = arrow::struct_({scalar_field}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr output, + file_system->Create(new_path, /*overwrite=*/true)); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr writer, + BlobFormatWriter::Create(output, scalar_type, /*write_null_on_missing_file=*/false, + /*write_null_on_fetch_failure=*/false, file_system, pool_)); + auto rows = arrow::ipc::internal::json::ArrayFromJSON(scalar_type, R"([ + [null], + ["_PAIMON_BLOB_PLACEHOLDER"], + ["_PAIMON_BLOB_PLACEHOLDER"], + ["_PAIMON_BLOB_PLACEHOLDER"] + ])") + .ValueOrDie(); + for (int32_t i = 0; i < 4; i++) { + ::ArrowArray c_rows; + ASSERT_TRUE(arrow::ExportArray(*rows->Slice(i, 1), &c_rows).ok()); + ASSERT_OK(writer->AddBatch(&c_rows)); + } + ASSERT_OK(writer->Finish()); + ASSERT_OK(output->Close()); + + for (bool is_map : {false, true}) { + const std::string old_bytes = is_map ? MapBlobGoldenBytes() : ArrayBlobGoldenBytes(); + ASSERT_OK(file_system->WriteFile(old_path, old_bytes, /*overwrite=*/true)); + std::shared_ptr blob_type; + if (is_map) { + blob_type = arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", true)); + } else { + blob_type = arrow::list(BlobUtils::ToArrowField("item", true)); + } + auto field = arrow::field("blob_col", blob_type, true); + auto schema = arrow::schema({field}); + + for (int32_t batch_size : {1, 2, 1024}) { + for (bool with_selection : {false, true}) { + std::optional selection; + if (with_selection) { + selection = RoaringBitmap32::From({0, 1, 3}); + } + auto open = + [&](const std::string& path) -> Result { + auto context = std::make_shared(); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input, + file_system->Open(path)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr raw_reader, + BlobFileBatchReader::Create(input, /*batch_size=*/1024, GetParam(), + /*emit_placeholder_sentinel=*/true, pool_, + context, GetArrowPool(pool_))); + ::ArrowSchema c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); + PAIMON_RETURN_NOT_OK(raw_reader->SetReadSchema(&c_schema, nullptr, selection)); + return BlobFallbackBatchReader::Segment{ + std::move(raw_reader), {}, std::move(context)}; + }; + ASSERT_OK_AND_ASSIGN(auto new_segment, open(new_path)); + ASSERT_OK_AND_ASSIGN(auto old_segment, open(old_path)); + std::vector> groups(2); + groups[0].push_back(std::move(new_segment)); + groups[1].push_back(std::move(old_segment)); + ASSERT_OK_AND_ASSIGN( + auto fallback, BlobFallbackBatchReader::Create( + std::move(groups), schema, batch_size, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto result, paimon::test::ReadResultCollector::CollectResult( + std::move(fallback))); + auto combined = arrow::Concatenate(result->chunks()); + ASSERT_TRUE(combined.ok()); + auto actual = std::dynamic_pointer_cast(combined.ValueOrDie()); + ASSERT_TRUE(actual); + ASSERT_EQ(actual->length(), with_selection ? 3 : 4); + ASSERT_TRUE(actual->field(0)->IsNull(0)); + const int64_t last_row = actual->length() - 1; + if (is_map) { + auto maps = std::dynamic_pointer_cast(actual->field(0)); + ASSERT_TRUE(maps); + ASSERT_TRUE(maps->IsNull(1)); + if (!with_selection) { + ASSERT_FALSE(maps->IsNull(2)); + ASSERT_EQ(maps->value_length(2), 0); + } + ASSERT_EQ(maps->value_length(last_row), 1); + auto blobs = std::dynamic_pointer_cast(maps->items()); + ASSERT_TRUE(blobs); + ASSERT_OK_AND_ASSIGN(std::string value, + ReadBlobValue(blobs, maps->value_offset(last_row), + GetParam(), file_system)); + ASSERT_EQ(value, "world"); + } else { + auto arrays = std::dynamic_pointer_cast(actual->field(0)); + ASSERT_TRUE(arrays); + ASSERT_EQ(arrays->value_length(1), 4); + if (!with_selection) { + ASSERT_TRUE(arrays->IsNull(2)); + } + ASSERT_TRUE(arrays->IsNull(last_row)); + auto blobs = + std::dynamic_pointer_cast(arrays->values()); + ASSERT_TRUE(blobs); + ASSERT_OK_AND_ASSIGN( + std::string value, + ReadBlobValue(blobs, arrays->value_offset(1), GetParam(), file_system)); + ASSERT_EQ(value, "inline"); + } + } + } + } +} + TEST_F(BlobFileBatchReaderTest, RejectsNonCanonicalDecimalMapKey) { auto dir = paimon::test::UniqueTestDirectory::Create(); ASSERT_TRUE(dir); @@ -475,10 +664,11 @@ TEST_F(BlobFileBatchReaderTest, RejectsNonCanonicalDecimalMapKey) { ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input, /*batch_size=*/1, /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/1, /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_NOK_WITH_MSG(reader->NextBatch(), "non-canonical decimal key"); } @@ -499,10 +689,11 @@ TEST_F(BlobFileBatchReaderTest, RejectsInvalidUtf8StringMapKey) { ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input, /*batch_size=*/1, /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/1, /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_NOK_WITH_MSG(reader->NextBatch(), "invalid UTF-8"); } @@ -525,10 +716,11 @@ TEST_F(BlobFileBatchReaderTest, RejectsNullMapKey) { ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK_AND_ASSIGN(std::shared_ptr input, file_system->Open(file_path)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input, /*batch_size=*/1, /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input, /*batch_size=*/1, /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_NOK_WITH_MSG(reader->NextBatch(), "MAP<..., BLOB> keys cannot be null"); } @@ -662,11 +854,11 @@ TEST_F(BlobFileBatchReaderTest, TestRowNumbers) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN(auto reader, - BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + auto reader, BlobFileBatchReader::Create(input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_OK_AND_ASSIGN(auto number_of_rows, reader->GetNumberOfRows()); @@ -705,15 +897,13 @@ TEST_F(BlobFileBatchReaderTest, TestRowNumbersWithSelectionBitmap) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN(auto reader, - BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); - - RoaringBitmap32 selection; - selection.Add(0); - selection.Add(2); + ASSERT_OK_AND_ASSIGN( + auto reader, BlobFileBatchReader::Create(input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); + + RoaringBitmap32 selection = RoaringBitmap32::From({0, 2}); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, selection)); ASSERT_NOK_WITH_MSG(reader->GetPreviousBatchFileRowId(0), "No batch has been read yet"); ASSERT_OK_AND_ASSIGN(auto batch1, reader->NextBatch()); @@ -746,22 +936,23 @@ TEST_F(BlobFileBatchReaderTest, InvalidScenario) { BlobFileBatchReader::Create(input_stream, /*batch_size=*/0, /*blob_as_descriptor=*/true, /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_)), + /*blob_context=*/nullptr, GetArrowPool(pool_)), "blob file batch reader create failed: read batch size '0' should be larger than zero"); } { ASSERT_NOK_WITH_MSG(BlobFileBatchReader::Create( /*input_stream=*/nullptr, /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_)), + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_)), "blob file batch reader create failed: input stream is nullptr"); } { - ASSERT_OK_AND_ASSIGN( - auto reader, BlobFileBatchReader::Create(/*input_stream=*/input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + /*input_stream=*/input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_NOK_WITH_MSG(reader->GetFileSchema(), "blob file has no self-describing file schema"); ASSERT_TRUE(reader->GetReaderMetrics()); @@ -794,11 +985,11 @@ TEST_P(BlobFileBatchReaderTest, EmptyFile) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system->Open(dir->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(auto reader, - BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + auto reader, BlobFileBatchReader::Create(input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt)); ASSERT_OK_AND_ASSIGN(auto number_of_rows, reader->GetNumberOfRows()); @@ -820,11 +1011,11 @@ TEST_F(BlobFileBatchReaderTest, SetReadSchemaWithInvalidInputs) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN( - auto reader, BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_NOK_WITH_MSG(reader->SetReadSchema(/*read_schema=*/nullptr, /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt), "SetReadSchema failed: read schema cannot be nullptr"); @@ -846,11 +1037,11 @@ TEST_F(BlobFileBatchReaderTest, SetReadSchemaWithInvalidInputs) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN( - auto reader, BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_NOK_WITH_MSG(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt), "read schema field number 2 is not 1"); @@ -872,11 +1063,11 @@ TEST_F(BlobFileBatchReaderTest, SetReadSchemaWithInvalidInputs) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN( - auto reader, BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ASSERT_NOK_WITH_MSG(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt), "field my_blob_field: large_binary must be BLOB, ARRAY or " @@ -897,17 +1088,12 @@ TEST_F(BlobFileBatchReaderTest, SetReadSchemaWithInvalidInputs) { ASSERT_OK_AND_ASSIGN( std::shared_ptr input_stream, fs->Open(table_path + "/bucket-0/data-d7816e8e-6c6d-4e28-9137-837cdf706350-1.blob")); - ASSERT_OK_AND_ASSIGN( - auto reader, BlobFileBatchReader::Create(input_stream, - /*batch_size=*/1, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); - RoaringBitmap32 roaring; - roaring.Add(0); - roaring.Add(1); - roaring.Add(2); - roaring.Add(3); - roaring.Add(4); + ASSERT_OK_AND_ASSIGN(auto reader, BlobFileBatchReader::Create( + input_stream, + /*batch_size=*/1, /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); + RoaringBitmap32 roaring = RoaringBitmap32::From({0, 1, 2, 3, 4}); ASSERT_NOK_WITH_MSG( reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, /*selection_bitmap=*/roaring), "Invalid: row index 3 is out of bound of total row number 3"); diff --git a/src/paimon/format/blob/blob_format_writer_test.cpp b/src/paimon/format/blob/blob_format_writer_test.cpp index cf7ff928f..2d06d94b9 100644 --- a/src/paimon/format/blob/blob_format_writer_test.cpp +++ b/src/paimon/format/blob/blob_format_writer_test.cpp @@ -301,7 +301,8 @@ class BlobFormatWriterTestBase : public ::testing::Test { PAIMON_ASSIGN_OR_RAISE( std::unique_ptr reader, BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, blob_as_descriptor, - emit_placeholder_sentinel, pool_, GetArrowPool(pool_))); + emit_placeholder_sentinel, pool_, /*blob_context=*/nullptr, + GetArrowPool(pool_))); auto schema = arrow::schema(struct_type_->fields()); ::ArrowSchema c_schema; PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); @@ -402,10 +403,11 @@ TEST_P(BlobFormatWriterTest, TestSimple) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); ASSERT_TRUE(input_stream); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input_stream, /*batch_size=*/1024, blob_as_descriptor_, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, blob_as_descriptor_, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); auto schema = arrow::schema(struct_type_->fields()); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); @@ -628,10 +630,11 @@ TEST_P(BlobFormatWriterTest, TestLargeBlob) { // Verify we can read it back ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input_stream, /*batch_size=*/1024, blob_as_descriptor_, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, blob_as_descriptor_, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); auto schema = arrow::schema(struct_type_->fields()); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); @@ -678,10 +681,11 @@ TEST_P(BlobFormatWriterTest, TestAddBatchWithNullValues) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); ASSERT_TRUE(input_stream); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create( - input_stream, /*batch_size=*/1024, blob_as_descriptor_, - /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, blob_as_descriptor_, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); auto schema = arrow::schema(struct_type_->fields()); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); @@ -1217,11 +1221,12 @@ TEST_F(BlobFormatWriterPlaceholderTest, TestReadPlaceholderStrictAndAwareModes) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, - /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/false, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, + /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/false, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, @@ -1233,11 +1238,12 @@ TEST_F(BlobFormatWriterPlaceholderTest, TestReadPlaceholderStrictAndAwareModes) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, - /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/true, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, + /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/true, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, @@ -1257,11 +1263,12 @@ TEST_F(BlobFormatWriterPlaceholderTest, TestReadPlaceholderStrictAndAwareModes) { ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, - /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/true, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, + /*blob_as_descriptor=*/true, + /*emit_placeholder_sentinel=*/true, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, @@ -1288,17 +1295,16 @@ TEST_F(BlobFormatWriterPlaceholderTest, TestReadPlaceholderWithSelectionBitmap) ASSERT_OK_AND_ASSIGN(std::shared_ptr input_stream, file_system_->Open(dir_->Str() + "/file.blob")); - ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, - /*blob_as_descriptor=*/false, - /*emit_placeholder_sentinel=*/true, pool_, - GetArrowPool(pool_))); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, + /*blob_as_descriptor=*/false, + /*emit_placeholder_sentinel=*/true, pool_, + /*blob_context=*/nullptr, GetArrowPool(pool_))); auto schema = arrow::schema(struct_type_->fields()); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); - RoaringBitmap32 selection; - selection.Add(1); - selection.Add(2); + RoaringBitmap32 selection = RoaringBitmap32::From({1, 2}); ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, selection)); ASSERT_OK_AND_ASSIGN(auto chunked_array, paimon::test::ReadResultCollector::CollectResult(std::move(reader))); @@ -1331,7 +1337,7 @@ TEST_F(BlobFormatWriterPlaceholderTest, TestSentinelPrefixedValueVerbatim) { std::unique_ptr reader, BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, /*blob_as_descriptor=*/false, emit_placeholder_sentinel, - pool_, GetArrowPool(pool_))); + pool_, /*blob_context=*/nullptr, GetArrowPool(pool_))); ::ArrowSchema c_schema; ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok()); ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr, diff --git a/src/paimon/format/blob/blob_reader_builder.h b/src/paimon/format/blob/blob_reader_builder.h index 9f4460dc3..28003c049 100644 --- a/src/paimon/format/blob/blob_reader_builder.h +++ b/src/paimon/format/blob/blob_reader_builder.h @@ -22,6 +22,7 @@ #include #include +#include "paimon/common/reader/blob_read_context.h" #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/common/utils/options_utils.h" #include "paimon/format/blob/blob_file_batch_reader.h" @@ -31,7 +32,7 @@ namespace paimon::blob { -class BlobReaderBuilder : public ReaderBuilder { +class BlobReaderBuilder : public ReaderBuilder, public BlobReadContextAwareBuilder { public: BlobReaderBuilder(int32_t batch_size, const std::map& options) : batch_size_(batch_size), @@ -49,6 +50,10 @@ class BlobReaderBuilder : public ReaderBuilder { return this; } + void SetBlobReadContext(const std::shared_ptr& context) override { + blob_context_ = context; + } + Result> Build( const std::shared_ptr& input_stream) const override { if (pool_ == nullptr) { @@ -61,7 +66,8 @@ class BlobReaderBuilder : public ReaderBuilder { OptionsUtils::GetValueFromMap( options_, BlobDefs::kEmitPlaceholderSentinelKey, false)); return BlobFileBatchReader::Create(input_stream, batch_size_, blob_as_descriptor, - emit_placeholder_sentinel, pool_, arrow_pool_); + emit_placeholder_sentinel, pool_, blob_context_, + arrow_pool_); } private: @@ -69,6 +75,7 @@ class BlobReaderBuilder : public ReaderBuilder { std::shared_ptr pool_; std::shared_ptr arrow_pool_; std::map options_; + std::shared_ptr blob_context_; }; } // namespace paimon::blob diff --git a/src/paimon/format/blob/blob_stats_extractor.cpp b/src/paimon/format/blob/blob_stats_extractor.cpp index 5e71121f1..4b65830c9 100644 --- a/src/paimon/format/blob/blob_stats_extractor.cpp +++ b/src/paimon/format/blob/blob_stats_extractor.cpp @@ -57,7 +57,8 @@ BlobStatsExtractor::ExtractWithFileInfo(const std::shared_ptr& file_ std::unique_ptr blob_reader, BlobFileBatchReader::Create(input_stream, /*batch_size=*/1024, /*blob_as_descriptor=*/true, - /*emit_placeholder_sentinel=*/false, pool, GetArrowPool(pool))); + /*emit_placeholder_sentinel=*/false, pool, + /*blob_context=*/nullptr, GetArrowPool(pool))); ColumnStatsVector result_stats; result_stats.push_back( ColumnStats::CreateStringColumnStats(std::nullopt, std::nullopt, std::nullopt));