feat(blob): support placeholder fallback for partial updates

A data-evolution partial update rewrites only the touched rows of a
blob column and records every untouched row as a placeholder entry
(bin_length -2, no data bytes).

The blob format now writes and reads such entries, BlobBunch keeps all
max-sequence layers of a bunch, and the new BlobFallbackBatchReader
resolves each row to the newest layer holding a real value: an
explicitly written null wins over older layers, a row that is a
placeholder in every layer degrades to null, and row-range pushdown is
honored including the row id ranges a layer does not cover.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: SteNicholas

src/paimon/CMakeLists.txt

@@ -130,6 +130,7 @@ set(PAIMON_COMMON_SRCS
     common/reader/predicate_batch_reader.cpp
     common/reader/prefetch_file_batch_reader_impl.cpp
     common/reader/reader_utils.cpp
+    common/reader/blob_fallback_batch_reader.cpp
     common/reader/blob_view_resolving_batch_reader.cpp
     common/reader/complete_row_kind_batch_reader.cpp
     common/reader/data_evolution_file_reader.cpp
@@ -540,6 +541,7 @@ if(PAIMON_BUILD_TESTS)
                     common/reader/prefetch_file_batch_reader_impl_test.cpp
                     common/reader/reader_utils_test.cpp
                     common/reader/complete_row_kind_batch_reader_test.cpp
+                    common/reader/blob_fallback_batch_reader_test.cpp
                     common/reader/blob_view_resolving_batch_reader_test.cpp
                     common/reader/data_evolution_file_reader_test.cpp
                     common/reader/data_evolution_array_test.cpp

src/paimon/common/data/blob_defs.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <cstdint>
+#include <cstring>
 
 namespace paimon {
 
@@ -45,6 +46,28 @@ class BlobDefs {
 
     /// A bin_length value of -1 in the index indicates a null blob entry.
     static constexpr int64_t kNullBinLength = -1;
+    /// A bin_length value of -2 in the index indicates a placeholder blob entry, written by
+    /// data-evolution partial updates for rows whose blob value is not updated. A placeholder
+    /// entry occupies no file space; readers must fall back to an older blob file covering the
+    /// same row to resolve the value. Aligned with Java's BlobFormatWriter.PLACE_HOLDER_LENGTH.
+    static constexpr int64_t kPlaceholderBinLength = -2;
+    /// In-band sentinel bytes representing a placeholder blob value. The write path persists a
+    /// blob equal to these bytes as a placeholder entry, the same way a serialized
+    /// BlobDescriptor is detected by its magic header; a placeholder-aware reader (see
+    /// kEmitPlaceholderSentinelKey) emits them for placeholder entries so that fallback merging
+    /// can still identify placeholders after the batch has passed through schema-mapping
+    /// readers. Layout: version(1) + magic "BLOBPLHD"(8).
+    static constexpr char kPlaceholderSentinel[] = {0x01, 'B', 'L', 'O', 'B', 'P', 'L', 'H', 'D'};
+    static constexpr int32_t kPlaceholderSentinelLength = 9;
+    /// Internal (non user-facing) format option: when "true", the blob reader emits
+    /// kPlaceholderSentinel for placeholder entries instead of failing on them. Only the
+    /// data-evolution blob fallback read path sets this.
+    static constexpr char kEmitPlaceholderSentinelKey[] = "blob.internal.emit-placeholder-sentinel";
+
+    static bool IsPlaceholderSentinel(const char* data, size_t size) {
+        return size == static_cast<size_t>(kPlaceholderSentinelLength) &&
+               memcmp(data, kPlaceholderSentinel, kPlaceholderSentinelLength) == 0;
+    }
     /// Blob file format version.
     static constexpr int8_t kFileVersion = 1;
     /// Magic number identifying the start of each blob bin.

src/paimon/common/reader/blob_fallback_batch_reader.cpp

@@ -0,0 +1,325 @@
+/*
+ * Copyright 2024-present Alibaba Inc.
+ *
+ * Licensed 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/blob_fallback_batch_reader.h"
+
+#include <algorithm>
+#include <string_view>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "fmt/format.h"
+#include "paimon/common/data/blob_defs.h"
+#include "paimon/common/data/blob_utils.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"
+
+namespace paimon {
+
+Result<std::unique_ptr<BlobFallbackBatchReader>> BlobFallbackBatchReader::Create(
+    std::vector<std::vector<Segment>>&& sequence_groups,
+    const std::shared_ptr<arrow::Schema>& read_schema, int32_t read_batch_size,
+    const std::shared_ptr<MemoryPool>& pool) {
+    if (sequence_groups.size() < 2) {
+        return Status::Invalid(
+            "Blob fallback needs at least two sequence groups; a single group should be read "
+            "sequentially.");
+    }
+    if (read_schema == nullptr) {
+        return Status::Invalid("Blob fallback read schema cannot be nullptr.");
+    }
+    if (read_batch_size <= 0) {
+        return Status::Invalid(fmt::format(
+            "Blob fallback read batch size '{}' should be larger than zero", read_batch_size));
+    }
+    int32_t blob_field_idx = -1;
+    for (int32_t i = 0; i < read_schema->num_fields(); i++) {
+        if (BlobUtils::IsBlobField(read_schema->field(i))) {
+            if (blob_field_idx != -1) {
+                return Status::Invalid(
+                    "Blob fallback read schema should contain exactly one blob field.");
+            }
+            blob_field_idx = i;
+        }
+    }
+    if (blob_field_idx == -1) {
+        return Status::Invalid("Blob fallback read schema should contain a blob field.");
+    }
+    std::vector<GroupCursor> groups;
+    groups.reserve(sequence_groups.size());
+    for (auto& segments : sequence_groups) {
+        if (segments.empty()) {
+            return Status::Invalid("Blob fallback sequence group should not be empty.");
+        }
+        GroupCursor cursor;
+        cursor.segments = std::move(segments);
+        groups.push_back(std::move(cursor));
+    }
+    return std::unique_ptr<BlobFallbackBatchReader>(new BlobFallbackBatchReader(
+        std::move(groups), read_schema, blob_field_idx, read_batch_size, pool));
+}
+
+BlobFallbackBatchReader::BlobFallbackBatchReader(std::vector<GroupCursor>&& groups,
+                                                 const std::shared_ptr<arrow::Schema>& read_schema,
+                                                 int32_t blob_field_idx, int32_t read_batch_size,
+                                                 const std::shared_ptr<MemoryPool>& pool)
+    : groups_(std::move(groups)),
+      read_schema_(read_schema),
+      blob_field_idx_(blob_field_idx),
+      read_batch_size_(read_batch_size),
+      arrow_pool_(GetArrowPool(pool)) {}
+
+Result<int64_t> BlobFallbackBatchReader::FillWindow(size_t group_idx, int64_t want,
+                                                    std::vector<Chunk>* chunks) {
+    GroupCursor& cursor = groups_[group_idx];
+    int64_t collected = 0;
+    while (collected < want) {
+        if (!cursor.pending.empty()) {
+            const std::shared_ptr<arrow::StructArray>& 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});
+            cursor.pending_pos += take;
+            collected += take;
+            if (cursor.pending_pos == front->length()) {
+                cursor.pending.pop_front();
+                cursor.pending_pos = 0;
+            }
+            continue;
+        }
+        if (cursor.segment_idx >= cursor.segments.size()) {
+            // group exhausted; only the first group may define a shorter window
+            break;
+        }
+        Segment& segment = cursor.segments[cursor.segment_idx];
+        if (segment.reader == nullptr) {
+            // gap segment: all rows are placeholders
+            if (cursor.gap_remaining < 0) {
+                cursor.gap_remaining = segment.gap_selected_row_count;
+            }
+            if (cursor.gap_remaining == 0) {
+                cursor.segment_idx++;
+                cursor.gap_remaining = -1;
+                continue;
+            }
+            int64_t take = std::min(cursor.gap_remaining, want - collected);
+            chunks->push_back(Chunk{nullptr, 0, take});
+            cursor.gap_remaining -= take;
+            collected += take;
+            if (cursor.gap_remaining == 0) {
+                cursor.segment_idx++;
+                cursor.gap_remaining = -1;
+            }
+            continue;
+        }
+        PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap batch_with_bitmap,
+                               segment.reader->NextBatchWithBitmap());
+        if (BatchReader::IsEofBatch(batch_with_bitmap)) {
+            cursor.segment_idx++;
+            continue;
+        }
+        auto& [read_batch, bitmap] = batch_with_bitmap;
+        auto& [c_array, c_schema] = read_batch;
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> src_array,
+                                          arrow::ImportArray(c_array.get(), c_schema.get()));
+        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;
+            }
+            auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(selected_array);
+            if (struct_array == nullptr) {
+                return Status::Invalid("Blob fallback expects file readers to emit struct arrays.");
+            }
+            cursor.pending.push_back(std::move(struct_array));
+        }
+    }
+    return collected;
+}
+
+Result<std::vector<bool>> BlobFallbackBatchReader::ComputePlaceholderFlags(
+    const std::vector<Chunk>& chunks, int64_t row_count) const {
+    std::vector<bool> flags(row_count, false);
+    int64_t pos = 0;
+    for (const auto& chunk : chunks) {
+        if (chunk.array == nullptr) {
+            // gap rows stand for placeholders
+            std::fill(flags.begin() + pos, flags.begin() + pos + chunk.length, true);
+        } else {
+            std::shared_ptr<arrow::Array> blob_col = chunk.array->field(blob_field_idx_);
+            auto binary_col = std::dynamic_pointer_cast<arrow::LargeBinaryArray>(blob_col);
+            if (binary_col == nullptr) {
+                return Status::Invalid(fmt::format(
+                    "Blob fallback expects the blob column to be large binary, but got {}",
+                    blob_col->type()->ToString()));
+            }
+            for (int64_t k = 0; k < chunk.length; k++) {
+                int64_t idx = chunk.offset + k;
+                if (!binary_col->IsNull(idx)) {
+                    std::string_view value = binary_col->GetView(idx);
+                    flags[pos + k] = BlobDefs::IsPlaceholderSentinel(value.data(), value.size());
+                }
+            }
+        }
+        pos += chunk.length;
+    }
+    return flags;
+}
+
+Result<std::shared_ptr<arrow::Array>> BlobFallbackBatchReader::AssembleColumn(
+    int32_t field_idx, const std::vector<int32_t>& group_choice,
+    const std::vector<std::vector<Chunk>>& group_chunks) const {
+    const auto row_count = static_cast<int64_t>(group_choice.size());
+    arrow::ArrayVector pieces;
+    int64_t run_start = 0;
+    while (run_start < row_count) {
+        const int32_t group = group_choice[run_start];
+        int64_t run_end = run_start + 1;
+        while (run_end < row_count && group_choice[run_end] == group) {
+            run_end++;
+        }
+        if (group < 0) {
+            // placeholder in every layer: no value survives, the row degrades to null
+            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+                std::shared_ptr<arrow::Array> null_piece,
+                arrow::MakeArrayOfNull(read_schema_->field(field_idx)->type(), run_end - run_start,
+                                       arrow_pool_.get()));
+            pieces.push_back(std::move(null_piece));
+        } else {
+            int64_t pos = 0;
+            for (const auto& chunk : group_chunks[group]) {
+                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.");
+                    }
+                    std::shared_ptr<arrow::Array> column = chunk.array->field(field_idx);
+                    pieces.push_back(column->Slice(chunk.offset + (overlap_start - pos),
+                                                   overlap_end - overlap_start));
+                }
+                pos += chunk.length;
+                if (pos >= run_end) {
+                    break;
+                }
+            }
+        }
+        run_start = run_end;
+    }
+    if (pieces.size() == 1 && pieces[0]->offset() == 0) {
+        return pieces[0];
+    }
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> concat_array,
+                                      arrow::Concatenate(pieces, arrow_pool_.get()));
+    return concat_array;
+}
+
+Result<BatchReader::ReadBatch> BlobFallbackBatchReader::NextBatch() {
+    if (closed_) {
+        return Status::Invalid("blob fallback batch reader is closed");
+    }
+    std::vector<std::vector<Chunk>> group_chunks(groups_.size());
+    // 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++) {
+        int64_t expect = row_count == 0 ? 1 : row_count;
+        PAIMON_ASSIGN_OR_RAISE(int64_t got, FillWindow(g, expect, &group_chunks[g]));
+        int64_t expected = row_count == 0 ? 0 : row_count;
+        if (got != expected) {
+            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, expected));
+        }
+    }
+    if (row_count == 0) {
+        return BatchReader::MakeEofBatch();
+    }
+
+    std::vector<std::vector<bool>> placeholder_flags(groups_.size());
+    for (size_t g = 0; g < groups_.size(); g++) {
+        PAIMON_ASSIGN_OR_RAISE(placeholder_flags[g],
+                               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<int32_t> 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]) {
+                group_choice[r] = static_cast<int32_t>(g);
+                break;
+            }
+        }
+    }
+
+    arrow::ArrayVector columns;
+    columns.reserve(read_schema_->num_fields());
+    for (int32_t field_idx = 0; field_idx < read_schema_->num_fields(); field_idx++) {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> column,
+                               AssembleColumn(field_idx, group_choice, group_chunks));
+        columns.push_back(std::move(column));
+    }
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> target_array,
+                                      arrow::StructArray::Make(columns, read_schema_->fields()));
+    std::unique_ptr<ArrowArray> c_array = std::make_unique<ArrowArray>();
+    std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(
+        arrow::ExportArray(*target_array, c_array.get(), c_schema.get()));
+    return std::make_pair(std::move(c_array), std::move(c_schema));
+}
+
+Result<BatchReader::ReadBatchWithBitmap> BlobFallbackBatchReader::NextBatchWithBitmap() {
+    PAIMON_ASSIGN_OR_RAISE(ReadBatch batch, NextBatch());
+    if (BatchReader::IsEofBatch(batch)) {
+        return BatchReader::MakeEofBatchWithBitmap();
+    }
+    return ReaderUtils::AddAllValidBitmap(std::move(batch));
+}
+
+void BlobFallbackBatchReader::Close() {
+    for (auto& group : groups_) {
+        group.pending.clear();
+        for (auto& segment : group.segments) {
+            if (segment.reader) {
+                segment.reader->Close();
+            }
+        }
+    }
+    closed_ = true;
+}
+
+std::shared_ptr<Metrics> BlobFallbackBatchReader::GetReaderMetrics() const {
+    auto metrics = std::make_shared<MetricsImpl>();
+    for (const auto& group : groups_) {
+        for (const auto& segment : group.segments) {
+            if (segment.reader) {
+                auto reader_metrics = segment.reader->GetReaderMetrics();
+                if (reader_metrics) {
+                    metrics->Merge(reader_metrics);
+                }
+            }
+        }
+    }
+    return metrics;
+}
+
+}  // namespace paimon