feat(blob): support placeholder fallback for partial updates

src/paimon/CMakeLists.txt

@@ -131,6 +131,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
@@ -803,6 +805,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/table_read_test.cpp
                     core/table/source/append_count_reader_test.cpp
                     core/table/source/pk_count_reader_test.cpp
+                    core/table/source/data_evolution_batch_scan_test.cpp
                     core/table/source/data_split_test.cpp
                     core/table/source/deletion_file_test.cpp
                     core/table/source/split_generator_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,50 @@ 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 identify a placeholder by exact byte equality with this internal reserved
+    /// value (IsPlaceholderSentinel). A user blob whose bytes exactly equal the marker would
+    /// collide with it inside these channels; the marker is distinctive enough that this is
+    /// accepted as negligibly improbable. Sentinel bytes are never stored in blob files.
+    static constexpr char kPlaceholderSentinel[] = "_PAIMON_BLOB_PLACEHOLDER";
+    /// Byte length of kPlaceholderSentinel, excluding the literal's terminating NUL.
+    static constexpr int32_t kPlaceholderSentinelLength = sizeof(kPlaceholderSentinel) - 1;
+    /// Internal (non user-facing) format option, "false" by default: 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";
+    /// Internal (non user-facing) format option, "false" by default: when "true", the blob
+    /// format writer persists a value exactly equal to kPlaceholderSentinel as a bin_length -2
+    /// entry. 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)};
+    }
+
+    /// True when the bytes are exactly the 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.