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 exactly equal to
the internal reserved marker _PAIMON_BLOB_PLACEHOLDER is persisted as
a bin_length -2 entry; 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. Placeholders are identified by exact byte equality only; a
user blob colliding with the marker inside the two internal channels
is accepted as negligibly improbable.

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.

BlobFileBatchReader::GetPreviousBatchFileRowId now maps batch
positions back to original file row indexes through
target_blob_row_indexes_, so _ROW_ID completion keeps working when a
selection bitmap removed rows. CompleteRowTrackingFieldsBatchReader
takes the physical field names of a format without a self-describing
schema (blob) from DataFileMeta::write_cols instead of treating
NotImplemented as an empty file schema.

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

Author: 2 people

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

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.