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 writes such entries only for partial updates: a
data-evolution blob-only column write enables the placeholder protocol
(BlobDefs::kWritePlaceholderKey), under which a value equal to the
placeholder sentinel is persisted as a bin_length -2 entry and a
sentinel-prefixed real value is unescaped; any other write stores blob
bytes verbatim, so user data can never be turned into a placeholder.
The reader fails on placeholder entries unless the internal option
BlobDefs::kEmitPlaceholderSentinelKey switches it to emit the sentinel
for them, escaping stored values that start with the sentinel bytes so
the fallback merge can never mistake real data for placeholders.

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 a null blob while
keeping its _ROW_ID and reporting -1 as its _SEQUENCE_NUMBER, resolved
rows report their layer's sequence number, 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
@@ -542,6 +543,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,8 @@
 #pragma once
 
 #include <cstdint>
+#include <cstring>
+#include <string_view>
 
 namespace paimon {
 
@@ -45,6 +47,60 @@ 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;
+    /// Sentinel bytes standing for a placeholder blob value in two internal channels:
+    ///
+    /// - Write channel: a data-evolution partial update (a blob-only column write, see
+    ///   kWritePlaceholderKey) marks a not-updated row with these bytes, and the blob format
+    ///   writer persists it as a bin_length -2 entry. Use PlaceholderSentinelView() to build
+    ///   such write arrays. Outside that mode the writer never interprets values, so arbitrary
+    ///   user bytes can never be turned into a placeholder entry.
+    /// - Read channel: a placeholder-aware reader (see kEmitPlaceholderSentinelKey) emits these
+    ///   bytes for -2 entries so the fallback merge can identify placeholders after the batch
+    ///   has passed through schema-mapping readers.
+    ///
+    /// Both channels escape legitimate values so user bytes can never collide with the
+    /// sentinel: a real value whose bytes start with the sentinel travels with one extra copy
+    /// of the sentinel prepended, and the consuming end strips it (an exact match is a
+    /// placeholder, a longer match is unescaped). Sentinel bytes are never stored in blob files
+    /// and never returned to users. 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, "false" by default: when "true", the blob
+    /// reader emits kPlaceholderSentinel for placeholder entries (escaping real values as
+    /// described above) instead of failing on them. Only the data-evolution blob fallback read
+    /// path sets this.
+    static constexpr char kEmitPlaceholderSentinelKey[] = "blob.internal.emit-placeholder-sentinel";
+    /// Internal (non user-facing) format option, "false" by default: when "true", the blob
+    /// format writer applies the placeholder protocol to incoming values (an exact
+    /// kPlaceholderSentinel match becomes a bin_length -2 entry, a longer match is unescaped).
+    /// Only set for data-evolution partial updates, i.e. blob-only column writes of a table
+    /// with data evolution enabled; all other writes store bytes verbatim.
+    static constexpr char kWritePlaceholderKey[] = "blob.internal.write-placeholder";
+
+    /// The sentinel bytes for building a data-evolution partial-update write array: a row equal
+    /// to this view is persisted as a placeholder entry (see kWritePlaceholderKey).
+    static std::string_view PlaceholderSentinelView() {
+        return {kPlaceholderSentinel, static_cast<size_t>(kPlaceholderSentinelLength)};
+    }
+
+    static bool IsPlaceholderSentinel(const char* data, size_t size) {
+        return size == static_cast<size_t>(kPlaceholderSentinelLength) &&
+               memcmp(data, kPlaceholderSentinel, kPlaceholderSentinelLength) == 0;
+    }
+
+    /// True when the bytes start with the placeholder sentinel (an exact match included). A
+    /// longer match is an escaped real value: strip the leading sentinel to recover it. A
+    /// producer for either placeholder channel must prepend one sentinel copy to any real value
+    /// this predicate matches.
+    static bool HasPlaceholderSentinelPrefix(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.