This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 30a13d69e6 [flink] Support storing BlobDescriptor in lookup join for
BLOB tables (#8391)
30a13d69e6 is described below
commit 30a13d69e6947419b2914276686d7c079454cf42
Author: wangwj <[email protected]>
AuthorDate: Fri Jul 3 13:27:39 2026 +0800
[flink] Support storing BlobDescriptor in lookup join for BLOB tables
(#8391)
When performing lookup join against a Paimon BLOB table (e.g., storing
images/videos), the full blob content is materialized via
`BlobSerializer.serialize() → blob.toData()` and written into local
RocksDB during bootstrap. For tables with large blob fields, this
causes:
- Extremely high local disk usage (e.g., 2 billion images × 200KB =
~400TB total, ~400GB per subtask)
- Page-size overflow errors (single records exceeding the 64KB default
page)
- Prolonged bootstrap time leading to TaskManager heartbeat timeouts
This PR introduces a new table option `lookup.blob-as-descriptor`
(default `false`). When enabled:
1. BLOB fields are stored as their lightweight `BlobDescriptor` bytes
(~130 bytes containing file URI, offset, and length) instead of the full
blob content.
2. The RowType used for value serialization replaces BLOB with
VARBINARY, so `InternalSerializers.create()` produces `BinarySerializer`
(calls `getBinary()`) instead of `BlobSerializer` (calls
`getBlob().toData()`).
3. `BlobRef.toDescriptor()` is a pure in-memory operation (no I/O), so
bootstrap performance is not affected.
The downstream consumer receives the serialized BlobDescriptor bytes and
can resolve the actual blob content on demand via a UDF reading from
DFS.
Impact: Per-subtask RocksDB storage drops from ~400GB to ~3.6GB (over
100x reduction), making lookup join feasible for tables with large BLOB
columns.
Tests
Existing unit tests pass (no behavioral change when the option is
disabled).
TODO: Add integration test for NoPrimaryKeyLookupTable with
lookup.blob-as-descriptor = true to verify BlobDescriptor bytes are
correctly stored and returned.
**Example usage:**
```sql
CREATE TABLE dim_images (
url STRING,
image BLOB
) WITH (
'lookup.blob-as-descriptor' = 'true'
);
SELECT s.*, resolve_blob(d.image) AS image_data
FROM stream_table s
LEFT JOIN dim_images FOR SYSTEM_TIME AS OF s.proc_time AS d
ON s.url = d.url;
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 13 ++
.../paimon/flink/lookup/BlobAsDescriptorRow.java | 215 +++++++++++++++++++++
.../paimon/flink/lookup/FullCacheLookupTable.java | 56 +++++-
.../flink/lookup/NoPrimaryKeyLookupTable.java | 4 +-
.../org/apache/paimon/flink/LookupJoinITCase.java | 58 ++++++
6 files changed, 347 insertions(+), 5 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 33d038ba43..f75de298d6 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -849,6 +849,12 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td>Boolean</td>
<td>When need to lookup, commit will wait for compaction by
lookup.</td>
</tr>
+ <tr>
+ <td><h5>lookup.blob-as-descriptor</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>When enabled, the lookup join stores only the BlobDescriptor
(a lightweight reference containing file URI, offset, and length) for BLOB
fields instead of the full blob bytes. This dramatically reduces local disk and
memory usage for tables with large BLOB columns (e.g., images, videos). The
downstream consumer receives the serialized BlobDescriptor bytes and can
resolve the actual blob content on demand.</td>
+ </tr>
<tr>
<td><h5>lookup.cache-file-retention</h5></td>
<td style="word-wrap: break-word;">1 h</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 55c25b409b..ee1db88285 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1422,6 +1422,19 @@ public class CoreOptions implements Serializable {
.defaultValue(MemorySize.parse("256 mb"))
.withDescription("Max memory size for lookup cache.");
+ public static final ConfigOption<Boolean> LOOKUP_CACHE_BLOB_DESCRIPTOR =
+ key("lookup.blob-as-descriptor")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When enabled, the lookup join stores only the
BlobDescriptor "
+ + "(a lightweight reference containing
file URI, offset, and length) "
+ + "for BLOB fields instead of the full
blob bytes. This dramatically "
+ + "reduces local disk and memory usage for
tables with large BLOB "
+ + "columns (e.g., images, videos). The
downstream consumer receives "
+ + "the serialized BlobDescriptor bytes and
can resolve the actual "
+ + "blob content on demand.");
+
public static final ConfigOption<Double> LOOKUP_CACHE_HIGH_PRIO_POOL_RATIO
=
key("lookup.cache.high-priority-pool-ratio")
.doubleType()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/BlobAsDescriptorRow.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/BlobAsDescriptorRow.java
new file mode 100644
index 0000000000..b79e46d4f8
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/BlobAsDescriptorRow.java
@@ -0,0 +1,215 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.flink.lookup;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalVector;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarBinaryType;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * An {@link InternalRow} wrapper that converts BLOB fields to their
BlobDescriptor serialized
+ * bytes.
+ *
+ * <p>When the lookup cache stores blob descriptors instead of full blob data,
this wrapper
+ * intercepts access to BLOB fields and returns the serialized BlobDescriptor
bytes via {@link
+ * #getBinary(int)}. This allows the lookup cache to store only lightweight
references (~100-150
+ * bytes) instead of the full blob content (which can be megabytes per record).
+ *
+ * <p>Non-blob fields are delegated to the wrapped row unchanged.
+ */
+public class BlobAsDescriptorRow implements InternalRow {
+
+ private final InternalRow wrapped;
+ private final Set<Integer> blobFieldPositions;
+
+ public BlobAsDescriptorRow(InternalRow wrapped, Set<Integer>
blobFieldPositions) {
+ this.wrapped = wrapped;
+ this.blobFieldPositions = blobFieldPositions;
+ }
+
+ /**
+ * Returns the set of field positions that are BLOB type in the given row
type.
+ *
+ * @param rowType the row type to inspect
+ * @return set of positions with BLOB type fields, empty if none
+ */
+ public static Set<Integer> blobFieldPositions(RowType rowType) {
+ Set<Integer> positions = new HashSet<>();
+ List<DataType> fieldTypes = rowType.getFieldTypes();
+ for (int i = 0; i < fieldTypes.size(); i++) {
+ if (fieldTypes.get(i).getTypeRoot() == DataTypeRoot.BLOB) {
+ positions.add(i);
+ }
+ }
+ return positions;
+ }
+
+ /**
+ * Creates a new RowType where BLOB fields are replaced with VARBINARY.
This is used to create
+ * the correct serializer that uses BinarySerializer instead of
BlobSerializer for cached
+ * values.
+ *
+ * @param rowType the original row type
+ * @param blobPositions the positions of BLOB fields
+ * @return a new RowType with BLOB fields replaced by VARBINARY
+ */
+ public static RowType replaceBlobWithVarBinary(RowType rowType,
Set<Integer> blobPositions) {
+ if (blobPositions.isEmpty()) {
+ return rowType;
+ }
+ List<DataType> newTypes = new ArrayList<>(rowType.getFieldTypes());
+ for (int pos : blobPositions) {
+ newTypes.set(pos, new VarBinaryType(VarBinaryType.MAX_LENGTH));
+ }
+ RowType.Builder builder = RowType.builder();
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ builder.field(rowType.getFields().get(i).name(), newTypes.get(i));
+ }
+ return builder.build();
+ }
+
+ @Override
+ public int getFieldCount() {
+ return wrapped.getFieldCount();
+ }
+
+ @Override
+ public RowKind getRowKind() {
+ return wrapped.getRowKind();
+ }
+
+ @Override
+ public void setRowKind(RowKind kind) {
+ wrapped.setRowKind(kind);
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return wrapped.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return wrapped.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return wrapped.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return wrapped.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return wrapped.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return wrapped.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return wrapped.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return wrapped.getDouble(pos);
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return wrapped.getString(pos);
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ return wrapped.getDecimal(pos, precision, scale);
+ }
+
+ @Override
+ public Timestamp getTimestamp(int pos, int precision) {
+ return wrapped.getTimestamp(pos, precision);
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ if (blobFieldPositions.contains(pos)) {
+ // Convert blob to descriptor bytes for caching
+ Blob blob = wrapped.getBlob(pos);
+ if (blob == null) {
+ return null;
+ }
+ return blob.toDescriptor().serialize();
+ }
+ return wrapped.getBinary(pos);
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ return wrapped.getArray(pos);
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ return wrapped.getMap(pos);
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ return wrapped.getRow(pos, numFields);
+ }
+
+ @Override
+ public Blob getBlob(int pos) {
+ return wrapped.getBlob(pos);
+ }
+
+ @Override
+ public Variant getVariant(int pos) {
+ return wrapped.getVariant(pos);
+ }
+
+ @Override
+ public InternalVector getVector(int pos) {
+ return wrapped.getVector(pos);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index 6b02084d8a..520021fa4d 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -55,6 +55,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -81,6 +82,8 @@ public abstract class FullCacheLookupTable implements
LookupTable {
protected final Context context;
protected final RowType projectedType;
protected final boolean refreshAsync;
+ protected final boolean blobAsDescriptor;
+ protected final Set<Integer> blobFieldPositions;
@Nullable protected final FieldsComparator userDefinedSeqComparator;
protected final int appendUdsFieldNumber;
@@ -135,6 +138,8 @@ public abstract class FullCacheLookupTable implements
LookupTable {
Options options = Options.fromMap(context.table.options());
this.projectedType = projectedType;
this.refreshAsync = options.get(LOOKUP_REFRESH_ASYNC);
+ this.blobAsDescriptor =
options.get(CoreOptions.LOOKUP_CACHE_BLOB_DESCRIPTOR);
+ this.blobFieldPositions =
BlobAsDescriptorRow.blobFieldPositions(projectedType);
this.cachedException = new AtomicReference<>();
this.maxPendingSnapshotCount =
options.get(LOOKUP_REFRESH_ASYNC_PENDING_SNAPSHOT_COUNT);
}
@@ -181,9 +186,23 @@ public abstract class FullCacheLookupTable implements
LookupTable {
protected void bootstrap() throws Exception {
Predicate scanPredicate =
PredicateBuilder.andNullable(context.tablePredicate,
partitionFilter);
+
+ // When lookup.blob-as-descriptor is enabled. Force the format reader
to return
+ // BlobRef (descriptor-backed) instead of BlobData for BLOB fields.
This ensures
+ // blob.toDescriptor() succeeds during cache serialization, even when
the table
+ // was not originally written with blob-as-descriptor=true.
+ LookupFileStoreTable readerTable = context.table;
+ if (blobAsDescriptor && !blobFieldPositions.isEmpty()) {
+ readerTable =
+ (LookupFileStoreTable)
+ context.table.copy(
+ Collections.singletonMap(
+
CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true"));
+ }
+
this.reader =
new LookupStreamingReader(
- context.table,
+ readerTable,
context.projection,
scanPredicate,
context.requiredCachedBucketIds,
@@ -194,16 +213,22 @@ public abstract class FullCacheLookupTable implements
LookupTable {
return;
}
+ // Parallel bootstrap read serializes rows with BlobSerializer, which
materializes
+ // BlobRef into BlobData. Disable parallelism when caching blob
descriptors.
+ boolean useParallelBootstrapRead = !(blobAsDescriptor &&
!blobFieldPositions.isEmpty());
+
BinaryExternalSortBuffer bulkLoadSorter =
RocksDBState.createBulkLoadSorter(
IOManager.create(context.tempPath.toString()),
context.table.coreOptions());
Predicate predicate = projectedPredicate();
try (RecordReaderIterator<InternalRow> batch =
- new
RecordReaderIterator<>(reader.toRecordReader(reader.nextSplits(), true))) {
+ new RecordReaderIterator<>(
+ reader.toRecordReader(reader.nextSplits(),
useParallelBootstrapRead))) {
while (batch.hasNext()) {
InternalRow row = batch.next();
if (predicate == null || predicate.test(row)) {
- bulkLoadSorter.write(GenericRow.of(toKeyBytes(row),
toValueBytes(row)));
+ InternalRow valueRow = wrapForCache(row);
+ bulkLoadSorter.write(GenericRow.of(toKeyBytes(row),
toValueBytes(valueRow)));
}
}
}
@@ -332,6 +357,31 @@ public abstract class FullCacheLookupTable implements
LookupTable {
return context.projectedPredicate;
}
+ /**
+ * Wraps the given row for caching. When {@code cacheBlobDescriptor} is
enabled and the row
+ * contains BLOB fields, wraps it with {@link BlobAsDescriptorRow} so that
BLOB fields are
+ * stored as lightweight BlobDescriptor bytes instead of full blob content.
+ */
+ protected InternalRow wrapForCache(InternalRow row) {
+ if (blobAsDescriptor && !blobFieldPositions.isEmpty()) {
+ return new BlobAsDescriptorRow(row, blobFieldPositions);
+ }
+ return row;
+ }
+
+ /**
+ * Returns the RowType used for value serialization in the cache. When
{@code
+ * cacheBlobDescriptor} is enabled, BLOB fields are replaced with
VARBINARY so that the
+ * serializer uses {@code BinarySerializer} (calls {@code getBinary()})
instead of {@code
+ * BlobSerializer} (calls {@code getBlob().toData()}).
+ */
+ protected RowType cacheValueRowType() {
+ if (blobAsDescriptor && !blobFieldPositions.isEmpty()) {
+ return BlobAsDescriptorRow.replaceBlobWithVarBinary(projectedType,
blobFieldPositions);
+ }
+ return projectedType;
+ }
+
public abstract byte[] toKeyBytes(InternalRow row) throws IOException;
public abstract byte[] toValueBytes(InternalRow row) throws IOException;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
index 546bc5a60a..4fd1dc1ae0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
@@ -59,7 +59,7 @@ public class NoPrimaryKeyLookupTable extends
FullCacheLookupTable {
"join-key-index",
InternalSerializers.create(
TypeUtils.project(projectedType,
joinKeyRow.indexMapping())),
- InternalSerializers.create(projectedType),
+ InternalSerializers.create(cacheValueRowType()),
lruCacheSize);
bootstrap();
}
@@ -83,7 +83,7 @@ public class NoPrimaryKeyLookupTable extends
FullCacheLookupTable {
joinKeyRow.replaceRow(row);
if (row.getRowKind() == RowKind.INSERT || row.getRowKind() ==
RowKind.UPDATE_AFTER) {
if (predicate == null || predicate.test(row)) {
- state.add(joinKeyRow, row);
+ state.add(joinKeyRow, wrapForCache(row));
}
} else {
throw new RuntimeException(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
index 32a4948501..ebdec9bc3e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink;
+import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.flink.FlinkConnectorOptions.LookupCacheMode;
import org.apache.paimon.utils.BlockingIterator;
@@ -1563,4 +1564,61 @@ public class LookupJoinITCase extends CatalogITCaseBase {
iterator.close();
}
+
+ @ParameterizedTest
+ @EnumSource(
+ value = LookupCacheMode.class,
+ names = {"FULL", "MEMORY"})
+ public void testLookupBlobAsDescriptorOnNormalBlobTable(LookupCacheMode
mode) throws Exception {
+ // Test that lookup.blob-as-descriptor works correctly even when the
table was NOT
+ // written with blob-as-descriptor=true. Previously this would fail
with
+ // "Blob data can not convert to descriptor" because BlobFormatReader
returned BlobData
+ // which cannot be converted to a descriptor. Blob tables cannot
define primary keys
+ // (row-tracking requirement), so use an append-only table with id as
the lookup key.
+ sql(
+ "CREATE TABLE BLOB_DIM (id INT, name STRING, picture BYTES)
WITH ("
+ + "'row-tracking.enabled'='true', "
+ + "'data-evolution.enabled'='true', "
+ + "'blob-field'='picture', "
+ + "'lookup.blob-as-descriptor'='true', "
+ + "'lookup.cache'='%s', "
+ + "'continuous.discovery-interval'='1 ms')",
+ mode);
+
+ // Write raw blob data (NOT as descriptor) — this is the normal write
path.
+ sql("INSERT INTO BLOB_DIM VALUES (1, 'cat', X'48656C6C6F'), (2, 'dog',
X'576F726C64')");
+
+ // Lookup join with lookup.blob-as-descriptor=true.
+ // The fix forces the reader to use blob-as-descriptor mode so that
BlobFormatReader
+ // returns BlobRef (which supports toDescriptor()) instead of BlobData.
+ String query =
+ "SELECT T.i, D.name, D.picture FROM T LEFT JOIN BLOB_DIM "
+ + "for system_time as of T.proctime AS D ON T.i =
D.id";
+ BlockingIterator<Row, Row> iterator =
BlockingIterator.of(sEnv.executeSql(query).collect());
+
+ sql("INSERT INTO T VALUES (1), (2), (3)");
+ List<Row> result = iterator.collect(3);
+
+ // With lookup.blob-as-descriptor=true, the BLOB fields should be
returned as
+ // serialized BlobDescriptor bytes (not the original raw data).
+ assertThat(result).hasSize(3);
+
+ // Verify non-blob fields are correct
+ assertThat(result.stream().map(r -> r.getField(1)))
+ .containsExactlyInAnyOrder("cat", "dog", null);
+
+ // For matched rows, the picture field should contain valid
BlobDescriptor bytes
+ for (Row row : result) {
+ if (row.getField(1) != null) {
+ byte[] descriptorBytes = (byte[]) row.getField(2);
+ assertThat(descriptorBytes).isNotNull();
+ // Verify it's a valid BlobDescriptor by deserializing
+ BlobDescriptor descriptor =
BlobDescriptor.deserialize(descriptorBytes);
+ assertThat(descriptor.uri()).isNotEmpty();
+ assertThat(descriptor.length()).isGreaterThan(0);
+ }
+ }
+
+ iterator.close();
+ }
}
