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 db2c6183f0 [core][flink][spark] Support managed BLOBs in primary-key
tables (#8617)
db2c6183f0 is described below
commit db2c6183f05247e8a01e68da952e5b4b8b70ed78
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 10:18:25 2026 +0800
[core][flink][spark] Support managed BLOBs in primary-key tables (#8617)
Add table-managed storage for top-level `BLOB` and `ARRAY<BLOB>` fields
in primary-key tables without tying payload bytes to data-file row
positions. Values are externalized before MergeTree buffering, while
compaction preserves stable descriptors and rebuilds exact per-data-file
references.
---
docs/docs/primary-key-table/blob-storage.md | 137 +++++++++
docs/sidebars.js | 1 +
.../apache/paimon/data/columnar/ColumnarArray.java | 14 +-
.../apache/paimon/data/columnar/ColumnarRow.java | 6 +-
.../data/serializer/InternalArraySerializer.java | 14 +
.../serializer/InternalArraySerializerTest.java | 34 +++
.../apache/paimon/format/FormatReadWriteTest.java | 39 +++
.../paimon/blob/ManagedBlobReferenceCollector.java | 127 ++++++++
.../paimon/blob/ManagedBlobReferenceFile.java | 188 ++++++++++++
.../paimon/blob/PrimaryKeyBlobExternalizer.java | 301 ++++++++++++++++++
.../apache/paimon/io/KeyValueDataFileWriter.java | 68 ++++-
.../paimon/io/KeyValueDataFileWriterImpl.java | 7 +-
.../paimon/io/KeyValueFileWriterFactory.java | 51 +++-
.../paimon/io/KeyValueThinDataFileWriterImpl.java | 7 +-
.../apache/paimon/mergetree/MergeTreeWriter.java | 7 +-
.../paimon/operation/LocalOrphanFilesClean.java | 1 +
.../apache/paimon/operation/OrphanFilesClean.java | 9 +
.../org/apache/paimon/schema/SchemaValidation.java | 65 +++-
.../blob/ManagedBlobReferenceCollectorTest.java | 158 ++++++++++
.../paimon/blob/ManagedBlobReferenceFileTest.java | 111 +++++++
.../blob/PrimaryKeyBlobExternalizerTest.java | 336 +++++++++++++++++++++
.../paimon/io/PrimaryKeyBlobFileWriterTest.java | 148 +++++++++
.../operation/LocalOrphanFilesCleanTest.java | 24 ++
.../operation/PrimaryKeyManagedBlobStoreTest.java | 316 +++++++++++++++++++
.../apache/paimon/schema/SchemaValidationTest.java | 167 ++++++++++
.../java/org/apache/paimon/flink/FlinkCatalog.java | 2 +-
.../paimon/flink/orphan/FlinkOrphanFilesClean.java | 4 +-
.../org/apache/paimon/flink/BlobTableITCase.java | 40 +++
.../parquet/writer/ParquetRowDataWriter.java | 3 +-
.../format/avro/AvroFormatReadWriteTest.java | 7 +
.../paimon/format/orc/OrcFormatReadWriteTest.java | 5 +
.../format/parquet/ParquetFormatReadWriteTest.java | 5 +
.../spark/procedure/SparkOrphanFilesClean.scala | 1 +
.../org/apache/paimon/spark/sql/BlobTestBase.scala | 20 ++
34 files changed, 2396 insertions(+), 27 deletions(-)
diff --git a/docs/docs/primary-key-table/blob-storage.md
b/docs/docs/primary-key-table/blob-storage.md
new file mode 100644
index 0000000000..938129c3f2
--- /dev/null
+++ b/docs/docs/primary-key-table/blob-storage.md
@@ -0,0 +1,137 @@
+---
+title: "BLOB Storage"
+sidebar_position: 11
+---
+
+<!--
+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.
+-->
+
+# BLOB Storage
+
+Primary-key tables can store top-level `BLOB` and `ARRAY<BLOB>` payloads in
table-managed files. Unlike the positional
+BLOB files used by append tables, managed BLOB payloads have stable
descriptors. MergeTree sorting, deduplication, and
+compaction can therefore reorder or remove rows without rewriting the
surviving payload bytes.
+
+This mode stores:
+
+- a serialized `BlobDescriptor` for each scalar value or non-null array
element;
+- the payload in an immutable `.managed.blob` pack; and
+- one `.blobref` sidecar for every data file, containing the exact managed
packs referenced by that file.
+
+For general BLOB concepts and read options, see [BLOB
Storage](../multimodal-table/blob).
+
+## Create a Table
+
+Use `blob-field` to mark scalar or array fields whose payloads should be
stored in managed BLOB files.
+`blob-descriptor-field` and `blob-view-field` are inline forms: their
serialized descriptor or view metadata stays in
+the normal data file and is not materialized into a managed BLOB file.
+
+The following example accepts both a scalar value and an ordered array of
values:
+
+```sql
+CREATE TABLE media (
+ id BIGINT,
+ name STRING,
+ content BYTES COMMENT '__BLOB_FIELD; media content',
+ attachments ARRAY<BYTES> COMMENT '__BLOB_FIELD; related files',
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'merge-engine' = 'deduplicate',
+ 'changelog-producer' = 'none',
+ 'blob.target-file-size' = '128 mb'
+);
+
+INSERT INTO media VALUES
+ (1, 'logo', X'89504E470D0A1A0A', ARRAY[X'25504446', NULL]);
+```
+
+For a primary-key table, every non-null `Blob` value in a `blob-field` is
externalized before it enters the MergeTree
+sort buffer. Its payload is copied into a new table-managed BLOB pack
regardless of the value's backing representation.
+Reads return the payload bytes by default; the existing `blob-as-descriptor`
read option can expose descriptors instead.
+A `blob-descriptor-field` is written inline to the normal data file and does
not participate in managed storage or its
+reference sidecars.
+
+`ARRAY<BLOB>` is externalized element by element. Every non-null `Blob`
element is copied into managed storage, while
+array order, a null array, and null elements are preserved. An empty array
writes no payload. `ARRAY<BLOB>` uses
+`blob-field`; `blob-descriptor-field` and `blob-view-field` remain scalar-only
declarations.
+
+`blob.target-file-size` controls when a writer rolls to a new managed payload
pack. A pack can contain payloads from
+multiple rows, and a row descriptor records its URI, offset, and length.
+
+:::note
+
+On append tables, `blob-descriptor-field` is descriptor-only storage and
writes must provide a descriptor. Append-table
+`blob-field` storage still requires row tracking and data evolution. The
managed raw-byte externalization described here
+is specific to supported primary-key tables.
+
+:::
+
+## Requirements and Limitations
+
+Primary-key managed BLOB storage has the following requirements:
+
+| Item | Requirement |
+|------|-------------|
+| Managed BLOB declaration | Scalar `BLOB` and `ARRAY<BLOB>` use `blob-field`;
`blob-descriptor-field` remains inline |
+| Merge engine | `deduplicate` only |
+| Changelog producer | `none` only |
+| Key usage | A managed BLOB column cannot be a primary, partition, bucket, or
sequence key |
+| External data paths | `data-file.external-paths` is not supported |
+| PK clustering override | `pk-clustering-override` is not supported |
+
+`row-tracking.enabled` and `data-evolution.enabled` are not required for this
primary-key mode.
+
+## Update, Delete, and Compaction
+
+An update writes a new descriptor and managed payload when a scalar value or
array element changes. A delete record does
+not write a new payload. Deduplication determines the final rows of each data
file, and its `.blobref` sidecar contains
+only the managed packs referenced by those rows.
+
+Compaction preserves descriptors for surviving values and creates new
`.blobref` sidecars from the compacted output.
+It does not copy the referenced payload bytes into new `.managed.blob` packs.
This keeps ordinary compaction cost
+proportional to row metadata instead of BLOB size.
+
+The `.blobref` file is owned by its data file through
`DataFileMeta.extraFiles`, so it follows the data file through
+snapshot, tag, branch, rollback, expiration, and deletion lifecycles. Shared
`.managed.blob` packs are deliberately not
+extra files because more than one retained data file can reference the same
pack.
+
+## Garbage Collection
+
+Garbage collection of unreferenced `.managed.blob` packs is not implemented
yet. Updates, deletes, compaction, or an
+ambiguous writer failure can therefore leave payload packs that are no longer
reachable from current rows.
+
+The ordinary orphan-file cleaner intentionally preserves all `.managed.blob`
files. This fail-safe behavior prevents it
+from deleting a payload that is still reachable from a snapshot, tag, branch,
or another retained root, but it also
+means unused BLOB storage can grow until a root-aware BLOB garbage collector
is available.
+
+A future collector must compute reachability across all retained roots and
treat a missing, corrupt, or unsupported
+`.blobref` sidecar as unsafe to delete. An empty, valid sidecar is different
from a missing sidecar: it explicitly states
+that the data file references no managed payload pack.
+
+## Reference Metadata
+
+The BLOB reference set is recorded in a data-file-owned `.blobref` sidecar
rather than `IndexManifest`. BLOB
+reachability is an exact dependency of one data file, and
`DataFileMeta.extraFiles` already provides the required
+lifecycle for such metadata. Using `IndexManifest` would require a new
snapshot-level index lifecycle and would be a
+larger compatibility change.
+
+Each sidecar is immutable, versioned, checksummed, and deterministic. It
stores only managed payload identities;
+descriptor and view fields remain inline and are not added to the managed
reference set. Managed pack identity is
+derived from the descriptor URI and its reserved `.managed.blob` suffix, so a
valid reference is retained even when the
+pack and the data file are in different directories.
diff --git a/docs/sidebars.js b/docs/sidebars.js
index e88793b7fe..ed0db484e7 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -87,6 +87,7 @@ const sidebars = {
"primary-key-table/global-index",
"primary-key-table/chain-table",
"primary-key-table/pk-clustering-override",
+ "primary-key-table/blob-storage",
{
type: "category",
"label": "Merge Engine",
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarArray.java
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarArray.java
index a6d0c8f704..ae8663f7d9 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarArray.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarArray.java
@@ -29,6 +29,7 @@ import org.apache.paimon.data.InternalVector;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.variant.GenericVariant;
import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.fs.FileIO;
import java.io.Serializable;
import java.util.Arrays;
@@ -41,6 +42,7 @@ public final class ColumnarArray implements InternalArray,
DataSetters, Serializ
private final ColumnVector data;
private final int offset;
private final int numElements;
+ private FileIO fileIO;
public ColumnarArray(ColumnVector data, int offset, int numElements) {
this.data = data;
@@ -48,6 +50,10 @@ public final class ColumnarArray implements InternalArray,
DataSetters, Serializ
this.numElements = numElements;
}
+ public void setFileIO(FileIO fileIO) {
+ this.fileIO = fileIO;
+ }
+
@Override
public int size() {
return numElements;
@@ -135,12 +141,16 @@ public final class ColumnarArray implements
InternalArray, DataSetters, Serializ
@Override
public Blob getBlob(int pos) {
- return Blob.fromBytes(getBinary(pos), null, null);
+ return Blob.fromBytes(getBinary(pos), null, fileIO, false);
}
@Override
public InternalArray getArray(int pos) {
- return ((ArrayColumnVector) data).getArray(offset + pos);
+ InternalArray array = ((ArrayColumnVector) data).getArray(offset +
pos);
+ if (array instanceof ColumnarArray) {
+ ((ColumnarArray) array).setFileIO(fileIO);
+ }
+ return array;
}
@Override
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRow.java
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRow.java
index e6d798078a..6ccbe5ee7a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRow.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRow.java
@@ -166,7 +166,11 @@ public final class ColumnarRow implements InternalRow,
DataSetters, Serializable
@Override
public InternalArray getArray(int pos) {
- return vectorizedColumnBatch.getArray(rowId, pos);
+ InternalArray array = vectorizedColumnBatch.getArray(rowId, pos);
+ if (array instanceof ColumnarArray) {
+ ((ColumnarArray) array).setFileIO(fileIO);
+ }
+ return array;
}
@Override
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalArraySerializer.java
b/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalArraySerializer.java
index b3126e4df5..c792d3a597 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalArraySerializer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalArraySerializer.java
@@ -29,6 +29,7 @@ import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.memory.MemorySegmentUtils;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
import java.io.IOException;
import java.lang.reflect.Array;
@@ -66,6 +67,8 @@ public class InternalArraySerializer implements
Serializer<InternalArray> {
return copyGenericArray((GenericArray) from);
} else if (from instanceof BinaryArray) {
return ((BinaryArray) from).copy();
+ } else if (eleType.getTypeRoot() == DataTypeRoot.BLOB) {
+ return copyObjectArray(from);
} else {
return toBinaryArray(from).copy();
}
@@ -104,6 +107,17 @@ public class InternalArraySerializer implements
Serializer<InternalArray> {
}
}
+ private GenericArray copyObjectArray(InternalArray array) {
+ Object[] newArray =
+ (Object[])
Array.newInstance(InternalRow.getDataClass(eleType), array.size());
+ for (int i = 0; i < array.size(); i++) {
+ if (!array.isNullAt(i)) {
+ newArray[i] =
eleSer.copy(elementGetter.getElementOrNull(array, i));
+ }
+ }
+ return new GenericArray(newArray);
+ }
+
@Override
public void serialize(InternalArray record, DataOutputView target) throws
IOException {
BinaryArray binaryArray = toBinaryArray(record);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/serializer/InternalArraySerializerTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/serializer/InternalArraySerializerTest.java
index fea4aa6789..e45db0bd56 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/serializer/InternalArraySerializerTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/serializer/InternalArraySerializerTest.java
@@ -21,17 +21,27 @@ package org.apache.paimon.data.serializer;
import org.apache.paimon.data.BinaryArray;
import org.apache.paimon.data.BinaryArrayWriter;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.columnar.ColumnarArray;
import org.apache.paimon.data.columnar.heap.HeapBytesVector;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.types.DataTypes;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** A test for the {@link InternalArraySerializer}. */
class InternalArraySerializerTest extends SerializerTestBase<InternalArray> {
@@ -82,6 +92,30 @@ class InternalArraySerializerTest extends
SerializerTestBase<InternalArray> {
return Arrays.copyOfRange(testData, 0, testData.length - 1);
}
+ @Test
+ void testCopyColumnarBlobArrayPreservesReader(@TempDir java.nio.file.Path
tempDir)
+ throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ byte[] payload = "blob-payload".getBytes(StandardCharsets.UTF_8);
+ Path path = new Path(tempDir.resolve("blob.data").toUri());
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ out.write(payload);
+ }
+
+ byte[] descriptor = new BlobDescriptor(path.toString(), 0,
payload.length).serialize();
+ HeapBytesVector vector = new HeapBytesVector(2);
+ vector.putByteArray(0, descriptor, 0, descriptor.length);
+ vector.setNullAt(1);
+ ColumnarArray array = new ColumnarArray(vector, 0, 2);
+ array.setFileIO(fileIO);
+
+ InternalArray copied = new
InternalArraySerializer(DataTypes.BLOB()).copy(array);
+
+ assertThat(copied).isInstanceOf(GenericArray.class);
+ assertThat(copied.getBlob(0).toData()).isEqualTo(payload);
+ assertThat(copied.isNullAt(1)).isTrue();
+ }
+
static BinaryArray copyNewOffset(BinaryArray array) {
BinaryArray newArray = new BinaryArray();
byte[] bytes = array.toBytes();
diff --git
a/paimon-common/src/test/java/org/apache/paimon/format/FormatReadWriteTest.java
b/paimon-common/src/test/java/org/apache/paimon/format/FormatReadWriteTest.java
index a07141937f..4b498cbee8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/format/FormatReadWriteTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/format/FormatReadWriteTest.java
@@ -20,6 +20,8 @@ package org.apache.paimon.format;
import org.apache.paimon.data.BinaryArray;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
@@ -48,6 +50,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -115,6 +118,42 @@ public abstract class FormatReadWriteTest {
assertThat(result.get(2).isNullAt(1)).isTrue();
}
+ protected void testArrayBlobDescriptorRoundTrip() throws IOException {
+ FileFormat format = fileFormat();
+ RowType rowType = DataTypes.ROW(DataTypes.ARRAY(DataTypes.BLOB()));
+ byte[] expected =
"managed-array-blob".getBytes(StandardCharsets.UTF_8);
+ Path payload = new Path(parent, "payload.managed.blob");
+ try (PositionOutputStream out = fileIO.newOutputStream(payload,
false)) {
+ out.write(expected);
+ }
+ BlobDescriptor first = new BlobDescriptor(payload.toString(), 0,
expected.length);
+ BlobDescriptor second = new BlobDescriptor("file:/blob/second", 7, 11);
+
+ write(
+ format.createWriterFactory(rowType),
+ file,
+ GenericRow.of(
+ new GenericArray(
+ new Object[] {
+ Blob.fromDescriptor(uri -> null, first),
+ null,
+ Blob.fromDescriptor(uri -> null, second)
+ })));
+
+ try (RecordReader<InternalRow> reader =
+ format.createReaderFactory(rowType, rowType, new ArrayList<>())
+ .createReader(
+ new FormatReaderContext(fileIO, file,
fileIO.getFileSize(file)))) {
+ InternalRow row = reader.readBatch().next();
+ InternalArray blobs = row.getArray(0);
+ assertThat(blobs.size()).isEqualTo(3);
+ assertThat(blobs.getBlob(0).toDescriptor()).isEqualTo(first);
+ assertThat(blobs.getBlob(0).toData()).isEqualTo(expected);
+ assertThat(blobs.isNullAt(1)).isTrue();
+ assertThat(blobs.getBlob(2).toDescriptor()).isEqualTo(second);
+ }
+ }
+
@Test
public void testFullTypes() throws IOException {
FileFormat format = fileFormat();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java
new file mode 100644
index 0000000000..38b898aea2
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java
@@ -0,0 +1,127 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobRef;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.paimon.types.BlobType.isBlobFileField;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Collects exact managed BLOB dependencies from the final rows of one data
file. */
+public class ManagedBlobReferenceCollector {
+
+ private final FileIO fileIO;
+ private final Path sidecar;
+ private final int[] blobFieldIndexes;
+ private final boolean[] blobArrayFields;
+ private final Set<ManagedBlobReferenceFile.Reference> references;
+
+ private boolean closed;
+
+ public ManagedBlobReferenceCollector(
+ FileIO fileIO, Path dataFile, RowType valueType, Set<String>
managedBlobFields) {
+ this.fileIO = fileIO;
+ this.sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ List<Integer> indexes = new ArrayList<>();
+ List<Boolean> arrayFields = new ArrayList<>();
+ for (int i = 0; i < valueType.getFieldCount(); i++) {
+ if (!managedBlobFields.contains(valueType.getFieldNames().get(i)))
{
+ continue;
+ }
+ DataTypeRoot typeRoot = valueType.getTypeAt(i).getTypeRoot();
+ if (isBlobFileField(valueType.getTypeAt(i))) {
+ indexes.add(i);
+ arrayFields.add(typeRoot == DataTypeRoot.ARRAY);
+ }
+ }
+ this.blobFieldIndexes =
indexes.stream().mapToInt(Integer::intValue).toArray();
+ this.blobArrayFields = new boolean[arrayFields.size()];
+ for (int i = 0; i < arrayFields.size(); i++) {
+ blobArrayFields[i] = arrayFields.get(i);
+ }
+ this.references = new HashSet<>();
+ }
+
+ public void write(KeyValue keyValue) {
+ checkState(!closed, "Managed BLOB reference collector is already
closed.");
+ if (keyValue.valueKind().isRetract()) {
+ return;
+ }
+
+ for (int i = 0; i < blobFieldIndexes.length; i++) {
+ int fieldIndex = blobFieldIndexes[i];
+ if (keyValue.value().isNullAt(fieldIndex)) {
+ continue;
+ }
+ if (blobArrayFields[i]) {
+ InternalArray array = keyValue.value().getArray(fieldIndex);
+ for (int j = 0; j < array.size(); j++) {
+ if (!array.isNullAt(j)) {
+ collect(array.getBlob(j));
+ }
+ }
+ } else {
+ collect(keyValue.value().getBlob(fieldIndex));
+ }
+ }
+ }
+
+ private void collect(Blob blob) {
+ if (blob instanceof BlobRef) {
+
ManagedBlobReferenceFile.fromDescriptorUri(blob.toDescriptor().uri())
+ .ifPresent(references::add);
+ }
+ }
+
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ try {
+ ManagedBlobReferenceFile.write(fileIO, sidecar, new
ArrayList<>(references));
+ closed = true;
+ } catch (IOException e) {
+ abort();
+ throw e;
+ }
+ }
+
+ public void abort() {
+ fileIO.deleteQuietly(sidecar);
+ closed = true;
+ }
+
+ public String result() {
+ checkState(closed, "Managed BLOB reference collector must be closed
before result.");
+ return sidecar.getName();
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
new file mode 100644
index 0000000000..d04457ecc2
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
@@ -0,0 +1,188 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.zip.CRC32;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Versioned metadata containing the managed BLOB packs referenced by one
data file. */
+public class ManagedBlobReferenceFile {
+
+ private static final int MAGIC = 0x50424C52;
+ private static final int VERSION = 1;
+
+ public static final String MANAGED_BLOB_SUFFIX = ".managed.blob";
+ public static final String REFERENCE_FILE_SUFFIX = ".blobref";
+
+ private ManagedBlobReferenceFile() {}
+
+ public static Optional<Reference> fromDescriptorUri(String descriptorUri) {
+ Path blobPath = new Path(descriptorUri);
+ if (!blobPath.getName().endsWith(MANAGED_BLOB_SUFFIX)) {
+ return Optional.empty();
+ }
+ return Optional.of(new Reference(blobPath.getParent().toString(),
blobPath.getName()));
+ }
+
+ public static Path sidecarPath(Path dataFile) {
+ return new Path(dataFile.getParent(), dataFile.getName() +
REFERENCE_FILE_SUFFIX);
+ }
+
+ public static void write(FileIO fileIO, Path path, List<Reference>
references)
+ throws IOException {
+ List<Reference> normalized = new ArrayList<>(new
HashSet<>(references));
+ normalized.sort(
+ Comparator.comparing(Reference::storageRootId)
+ .thenComparing(Reference::relativePath));
+
+ try (DataOutputStream out = new
DataOutputStream(fileIO.newOutputStream(path, false))) {
+ out.writeInt(MAGIC);
+ CRC32 checksum = new CRC32();
+ DataOutputStream payload = new DataOutputStream(new
CheckedOutputStream(out, checksum));
+ payload.writeByte(VERSION);
+ payload.writeInt(normalized.size());
+ for (Reference reference : normalized) {
+ payload.writeUTF(reference.storageRootId());
+ payload.writeUTF(reference.relativePath());
+ }
+ payload.flush();
+ out.writeInt((int) checksum.getValue());
+ } catch (IOException e) {
+ fileIO.deleteQuietly(path);
+ throw e;
+ }
+ }
+
+ public static List<Reference> read(FileIO fileIO, Path path) throws
IOException {
+ try (DataInputStream in = new
DataInputStream(fileIO.newInputStream(path))) {
+ int magic = in.readInt();
+ if (magic != MAGIC) {
+ throw new IOException("Invalid managed BLOB reference file
magic: " + magic);
+ }
+
+ CRC32 checksum = new CRC32();
+ DataInputStream payload = new DataInputStream(new
CheckedInputStream(in, checksum));
+ int version = payload.readUnsignedByte();
+ if (version != VERSION) {
+ throw new IOException(
+ "Unsupported managed BLOB reference file version: " +
version);
+ }
+
+ int count = payload.readInt();
+ if (count < 0) {
+ throw new IOException("Invalid managed BLOB reference count: "
+ count);
+ }
+
+ List<Reference> references = new ArrayList<>(Math.min(count,
1024));
+ for (int i = 0; i < count; i++) {
+ String storageRootId = payload.readUTF();
+ String relativePath = payload.readUTF();
+ try {
+ references.add(new Reference(storageRootId, relativePath));
+ } catch (IllegalArgumentException e) {
+ throw new IOException("Invalid managed BLOB reference
entry.", e);
+ }
+ }
+
+ int actualChecksum = (int) checksum.getValue();
+ int expectedChecksum = payload.readInt();
+ if (expectedChecksum != actualChecksum) {
+ throw new IOException(
+ "Invalid managed BLOB reference file checksum.
Expected "
+ + Integer.toUnsignedLong(expectedChecksum)
+ + " but computed "
+ + Integer.toUnsignedLong(actualChecksum)
+ + ".");
+ }
+ if (payload.read() != -1) {
+ throw new IOException("Unexpected trailing bytes in managed
BLOB reference file.");
+ }
+ return references;
+ }
+ }
+
+ /** Exact identity of a managed BLOB payload pack. */
+ public static class Reference {
+
+ private final String storageRootId;
+ private final String relativePath;
+
+ public Reference(String storageRootId, String relativePath) {
+ checkArgument(
+ storageRootId != null && !storageRootId.isEmpty(),
+ "Managed BLOB storage root must not be empty.");
+ checkArgument(
+ relativePath != null
+ && !relativePath.isEmpty()
+ && relativePath.equals(new
Path(relativePath).getName())
+ && !".".equals(relativePath)
+ && !"..".equals(relativePath),
+ "Managed BLOB relative path must be a file name: %s.",
+ relativePath);
+ this.storageRootId = storageRootId;
+ this.relativePath = relativePath;
+ }
+
+ public String storageRootId() {
+ return storageRootId;
+ }
+
+ public String relativePath() {
+ return relativePath;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Reference reference = (Reference) o;
+ return Objects.equals(storageRootId, reference.storageRootId)
+ && Objects.equals(relativePath, reference.relativePath);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(storageRootId, relativePath);
+ }
+
+ @Override
+ public String toString() {
+ return storageRootId + "/" + relativePath;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java
b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java
new file mode 100644
index 0000000000..b09c59b349
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java
@@ -0,0 +1,301 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.blob.BlobFormatWriter;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.RowDataToObjectArrayConverter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Externalizes primary-key BLOB values before they enter the MergeTree write
buffer. */
+public class PrimaryKeyBlobExternalizer {
+
+ private final FileIO fileIO;
+ private final RowDataToObjectArrayConverter rowConverter;
+ private final int[] blobFieldIndexes;
+ private final boolean[] blobArrayFields;
+ private final ManagedBlobPackWriter[] packWriters;
+ private final List<Path> uncommittedPacks;
+
+ public PrimaryKeyBlobExternalizer(
+ FileIO fileIO,
+ RowType valueType,
+ Set<String> managedBlobFields,
+ DataFilePathFactory pathFactory,
+ long targetFileSize) {
+ checkArgument(targetFileSize > 0, "Managed BLOB target file size must
be positive.");
+ this.fileIO = fileIO;
+ this.rowConverter = new RowDataToObjectArrayConverter(valueType);
+ this.uncommittedPacks = new ArrayList<>();
+
+ List<Integer> indexes = new ArrayList<>();
+ List<Boolean> arrayFields = new ArrayList<>();
+ List<ManagedBlobPackWriter> writers = new ArrayList<>();
+ Set<String> unknownFields = new TreeSet<>(managedBlobFields);
+ for (int i = 0; i < valueType.getFieldCount(); i++) {
+ DataField field = valueType.getFields().get(i);
+ if (!managedBlobFields.contains(field.name())) {
+ continue;
+ }
+ unknownFields.remove(field.name());
+ boolean blob = field.type().getTypeRoot() == DataTypeRoot.BLOB;
+ boolean blobArray =
+ field.type().getTypeRoot() == DataTypeRoot.ARRAY
+ && ((ArrayType)
field.type()).getElementType().getTypeRoot()
+ == DataTypeRoot.BLOB;
+ checkArgument(
+ blob || blobArray,
+ "Managed BLOB field '%s' must be BLOB or ARRAY<BLOB>, but
was %s.",
+ field.name(),
+ field.type());
+ indexes.add(i);
+ arrayFields.add(blobArray);
+ writers.add(
+ new ManagedBlobPackWriter(
+ fileIO,
+ blobArray
+ ? RowType.of(((ArrayType)
field.type()).getElementType())
+ : new
RowType(Collections.singletonList(field)),
+ pathFactory,
+ targetFileSize,
+ uncommittedPacks));
+ }
+ checkArgument(
+ unknownFields.isEmpty(),
+ "Managed BLOB fields do not exist in value type: %s.",
+ unknownFields);
+ this.blobFieldIndexes =
indexes.stream().mapToInt(Integer::intValue).toArray();
+ this.blobArrayFields = new boolean[arrayFields.size()];
+ for (int i = 0; i < arrayFields.size(); i++) {
+ blobArrayFields[i] = arrayFields.get(i);
+ }
+ this.packWriters = writers.toArray(new ManagedBlobPackWriter[0]);
+ }
+
+ public boolean enabled() {
+ return blobFieldIndexes.length > 0;
+ }
+
+ public InternalRow externalize(RowKind valueKind, InternalRow value)
throws IOException {
+ if (!enabled()) {
+ return value;
+ }
+
+ GenericRow result = null;
+ try {
+ for (int i = 0; i < blobFieldIndexes.length; i++) {
+ int fieldIndex = blobFieldIndexes[i];
+ if (value.isNullAt(fieldIndex)) {
+ continue;
+ }
+
+ if (valueKind.isRetract()) {
+ if (result == null) {
+ result = copy(value);
+ }
+ result.setField(fieldIndex, null);
+ } else if (blobArrayFields[i]) {
+ InternalArray array = value.getArray(fieldIndex);
+ GenericArray externalized = externalizeArray(array,
packWriters[i]);
+ if (externalized != null) {
+ if (result == null) {
+ result = copy(value);
+ }
+ result.setField(fieldIndex, externalized);
+ }
+ } else {
+ Blob blob = value.getBlob(fieldIndex);
+ if (result == null) {
+ result = copy(value);
+ }
+ BlobDescriptor descriptor = packWriters[i].write(blob);
+ result.setField(
+ fieldIndex,
+ Blob.fromFile(
+ fileIO,
+ descriptor.uri(),
+ descriptor.offset(),
+ descriptor.length()));
+ }
+ }
+ } catch (IOException | RuntimeException e) {
+ abort();
+ throw e;
+ }
+ return result == null ? value : result;
+ }
+
+ private GenericArray externalizeArray(InternalArray array,
ManagedBlobPackWriter packWriter)
+ throws IOException {
+ Object[] elements = null;
+ for (int i = 0; i < array.size(); i++) {
+ if (array.isNullAt(i)) {
+ continue;
+ }
+
+ Blob blob = array.getBlob(i);
+
+ if (elements == null) {
+ elements = new Object[array.size()];
+ for (int j = 0; j < array.size(); j++) {
+ elements[j] = array.isNullAt(j) ? null : array.getBlob(j);
+ }
+ }
+ BlobDescriptor descriptor = packWriter.write(blob);
+ elements[i] =
+ Blob.fromFile(
+ fileIO, descriptor.uri(), descriptor.offset(),
descriptor.length());
+ }
+ return elements == null ? null : new GenericArray(elements);
+ }
+
+ public void prepareCommit() throws IOException {
+ try {
+ for (ManagedBlobPackWriter packWriter : packWriters) {
+ packWriter.closeCurrent();
+ }
+ uncommittedPacks.clear();
+ } catch (IOException e) {
+ abort();
+ throw e;
+ }
+ }
+
+ public void abort() {
+ for (ManagedBlobPackWriter packWriter : packWriters) {
+ packWriter.abortCurrent();
+ }
+ for (Path path : uncommittedPacks) {
+ fileIO.deleteQuietly(path);
+ }
+ uncommittedPacks.clear();
+ }
+
+ private GenericRow copy(InternalRow value) {
+ GenericRow row = rowConverter.toGenericRow(value);
+ row.setRowKind(value.getRowKind());
+ return row;
+ }
+
+ private static class ManagedBlobPackWriter {
+
+ private final FileIO fileIO;
+ private final RowType blobType;
+ private final DataFilePathFactory pathFactory;
+ private final long targetFileSize;
+ private final List<Path> uncommittedPacks;
+
+ private Path currentPath;
+ private PositionOutputStream out;
+ private BlobFormatWriter writer;
+ private BlobDescriptor lastDescriptor;
+
+ private ManagedBlobPackWriter(
+ FileIO fileIO,
+ RowType blobType,
+ DataFilePathFactory pathFactory,
+ long targetFileSize,
+ List<Path> uncommittedPacks) {
+ this.fileIO = fileIO;
+ this.blobType = blobType;
+ this.pathFactory = pathFactory;
+ this.targetFileSize = targetFileSize;
+ this.uncommittedPacks = uncommittedPacks;
+ }
+
+ private BlobDescriptor write(Blob blob) throws IOException {
+ if (writer == null) {
+ openCurrent();
+ }
+
+ lastDescriptor = null;
+ writer.addElement(GenericRow.of(blob));
+ BlobDescriptor descriptor = lastDescriptor;
+ if (descriptor == null) {
+ throw new IOException("Managed BLOB writer did not produce a
descriptor.");
+ }
+ if (writer.reachTargetSize(true, targetFileSize)) {
+ closeCurrent();
+ }
+ return descriptor;
+ }
+
+ private void openCurrent() throws IOException {
+ currentPath =
+
pathFactory.newPathFromExtension(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ uncommittedPacks.add(currentPath);
+ out = fileIO.newOutputStream(currentPath, false);
+ writer =
+ new BlobFormatWriter(
+ out,
+ (fieldName, descriptor) -> {
+ lastDescriptor = descriptor;
+ return false;
+ },
+ blobType);
+ writer.setFile(currentPath);
+ }
+
+ private void closeCurrent() throws IOException {
+ if (writer == null) {
+ return;
+ }
+ try {
+ writer.close();
+ out.flush();
+ out.close();
+ } finally {
+ writer = null;
+ out = null;
+ currentPath = null;
+ }
+ }
+
+ private void abortCurrent() {
+ IOUtils.closeQuietly(writer);
+ IOUtils.closeQuietly(out);
+ writer = null;
+ out = null;
+ currentPath = null;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java
index 111f3dcf7e..34b6ee34f1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java
@@ -20,6 +20,8 @@ package org.apache.paimon.io;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.blob.ManagedBlobReferenceCollector;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.RowHelper;
@@ -39,8 +41,11 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
+import java.util.Set;
import java.util.function.Function;
import static org.apache.paimon.io.DataFilePathFactory.dataFileToFileIndexPath;
@@ -68,6 +73,7 @@ public abstract class KeyValueDataFileWriter
private final RowHelper keyKeeper;
private final FileSource fileSource;
@Nullable private final DataFileIndexWriter dataFileIndexWriter;
+ @Nullable private final ManagedBlobReferenceCollector
blobReferenceCollector;
private BinaryRow minKey = null;
private long minSeqNumber = Long.MAX_VALUE;
@@ -87,7 +93,8 @@ public abstract class KeyValueDataFileWriter
CoreOptions options,
FileSource fileSource,
FileIndexOptions fileIndexOptions,
- boolean isExternalPath) {
+ boolean isExternalPath,
+ Set<String> managedBlobFields) {
super(fileIO, context, path, converter, writeRowType,
options.asyncFileWrite());
this.keyType = keyType;
@@ -103,6 +110,11 @@ public abstract class KeyValueDataFileWriter
this.dataFileIndexWriter =
DataFileIndexWriter.create(
fileIO, dataFileToFileIndexPath(path), valueType,
fileIndexOptions);
+ this.blobReferenceCollector =
+ managedBlobFields.isEmpty()
+ ? null
+ : new ManagedBlobReferenceCollector(
+ fileIO, path, valueType, managedBlobFields);
}
@Override
@@ -112,6 +124,9 @@ public abstract class KeyValueDataFileWriter
if (dataFileIndexWriter != null) {
dataFileIndexWriter.write(kv.value());
}
+ if (blobReferenceCollector != null) {
+ blobReferenceCollector.write(kv);
+ }
keyKeeper.copyInto(kv.key());
if (minKey == null) {
@@ -159,6 +174,14 @@ public abstract class KeyValueDataFileWriter
: dataFileIndexWriter.result();
String externalPath = isExternalPath ? path.toString() : null;
+ List<String> extraFiles = new ArrayList<>();
+ if (indexResult.independentIndexFile() != null) {
+ extraFiles.add(indexResult.independentIndexFile());
+ }
+ if (blobReferenceCollector != null) {
+ extraFiles.add(blobReferenceCollector.result());
+ }
+
return DataFileMeta.create(
path.getName(),
fileSize,
@@ -171,9 +194,7 @@ public abstract class KeyValueDataFileWriter
maxSeqNumber,
schemaId,
level,
- indexResult.independentIndexFile() == null
- ? Collections.emptyList()
- :
Collections.singletonList(indexResult.independentIndexFile()),
+ extraFiles.isEmpty() ? Collections.emptyList() : extraFiles,
deleteRecordCount,
indexResult.embeddedIndexBytes(),
fileSource,
@@ -187,9 +208,42 @@ public abstract class KeyValueDataFileWriter
@Override
public void close() throws IOException {
- if (dataFileIndexWriter != null) {
- dataFileIndexWriter.close();
+ try {
+ if (dataFileIndexWriter != null) {
+ dataFileIndexWriter.close();
+ }
+ super.close();
+ if (blobReferenceCollector != null) {
+ blobReferenceCollector.close();
+ }
+ } catch (IOException e) {
+ abort();
+ throw e;
+ }
+ }
+
+ @Override
+ public void abort() {
+ if (blobReferenceCollector != null) {
+ blobReferenceCollector.abort();
+ }
+ super.abort();
+ }
+
+ @Override
+ public Optional<FileWriterAbortExecutor> abortExecutor() {
+ Optional<FileWriterAbortExecutor> mainExecutor = super.abortExecutor();
+ if (blobReferenceCollector == null) {
+ return mainExecutor;
}
- super.close();
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(path);
+ return Optional.of(
+ new FileWriterAbortExecutor(fileIO, path) {
+ @Override
+ public void abort() {
+ mainExecutor.ifPresent(FileWriterAbortExecutor::abort);
+ fileIO.deleteQuietly(sidecar);
+ }
+ });
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java
index 798c70ed23..628fdf5cfe 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java
@@ -30,6 +30,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;
import java.util.Arrays;
+import java.util.Set;
import java.util.function.Function;
/** Write data files containing {@link KeyValue}s. */
@@ -47,7 +48,8 @@ public class KeyValueDataFileWriterImpl extends
KeyValueDataFileWriter {
CoreOptions options,
FileSource fileSource,
FileIndexOptions fileIndexOptions,
- boolean isExternalPath) {
+ boolean isExternalPath,
+ Set<String> managedBlobFields) {
super(
fileIO,
context,
@@ -61,7 +63,8 @@ public class KeyValueDataFileWriterImpl extends
KeyValueDataFileWriter {
options,
fileSource,
fileIndexOptions,
- isExternalPath);
+ isExternalPath,
+ managedBlobFields);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
index b1ae451e59..0e17ec5e4b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
@@ -23,7 +23,9 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.KeyValueSerializer;
import org.apache.paimon.KeyValueThinSerializer;
import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.blob.PrimaryKeyBlobExternalizer;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatWriterFactory;
@@ -36,6 +38,7 @@ import
org.apache.paimon.statistics.NoneSimpleColStatsCollector;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Pair;
@@ -43,6 +46,7 @@ import org.apache.paimon.utils.StatsCollectorFactories;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -53,6 +57,8 @@ import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.paimon.types.BlobType.fieldNamesInBlobFile;
+
/** A factory to create {@link FileWriter}s for writing {@link KeyValue}
files. */
public class KeyValueFileWriterFactory {
@@ -64,6 +70,8 @@ public class KeyValueFileWriterFactory {
private final long suggestedFileSize;
private final CoreOptions options;
private final FileIndexOptions fileIndexOptions;
+ private final Set<String> managedBlobFields;
+ @Nullable private final PrimaryKeyBlobExternalizer blobExternalizer;
private KeyValueFileWriterFactory(
FileIO fileIO,
@@ -79,6 +87,16 @@ public class KeyValueFileWriterFactory {
this.suggestedFileSize = suggestedFileSize;
this.options = options;
this.fileIndexOptions = options.indexColumnsOptions();
+ this.managedBlobFields = managedBlobFields();
+ this.blobExternalizer =
+ managedBlobFields.isEmpty()
+ ? null
+ : new PrimaryKeyBlobExternalizer(
+ fileIO,
+ valueType,
+ managedBlobFields,
+ formatContext.pathFactory(new
WriteFormatKey(0, false)),
+ options.blobTargetFileSize());
}
public RowType keyType() {
@@ -89,6 +107,22 @@ public class KeyValueFileWriterFactory {
return valueType;
}
+ public InternalRow externalizeBlob(RowKind valueKind, InternalRow value)
throws IOException {
+ return blobExternalizer == null ? value :
blobExternalizer.externalize(valueKind, value);
+ }
+
+ public void prepareCommit() throws IOException {
+ if (blobExternalizer != null) {
+ blobExternalizer.prepareCommit();
+ }
+ }
+
+ public void abortManagedBlobWrites() {
+ if (blobExternalizer != null) {
+ blobExternalizer.abort();
+ }
+ }
+
@VisibleForTesting
public DataFilePathFactory pathFactory(int level) {
return formatContext.pathFactory(new WriteFormatKey(level, false));
@@ -151,6 +185,8 @@ public class KeyValueFileWriterFactory {
Path path, WriteFormatKey key, FileSource fileSource, boolean
isExternalPath) {
// Changelog is sequentially consumed, file index is unnecessary.
FileIndexOptions indexOptions = key.isChangelog ? new
FileIndexOptions() : fileIndexOptions;
+ Set<String> dataFileManagedBlobFields =
+ key.isChangelog ? Collections.emptySet() : managedBlobFields;
return formatContext.thinModeEnabled
? new KeyValueThinDataFileWriterImpl(
fileIO,
@@ -164,7 +200,8 @@ public class KeyValueFileWriterFactory {
options,
fileSource,
indexOptions,
- isExternalPath)
+ isExternalPath,
+ dataFileManagedBlobFields)
: new KeyValueDataFileWriterImpl(
fileIO,
formatContext.fileWriterContext(key),
@@ -177,14 +214,20 @@ public class KeyValueFileWriterFactory {
options,
fileSource,
indexOptions,
- isExternalPath);
+ isExternalPath,
+ dataFileManagedBlobFields);
+ }
+
+ private Set<String> managedBlobFields() {
+ return fieldNamesInBlobFile(valueType, options.blobInlineField());
}
public void deleteFile(DataFileMeta file) {
// this path factory is only for path generation, so we don't care
about the true or false
// in WriteFormatKey
- fileIO.deleteQuietly(
- formatContext.pathFactory(new WriteFormatKey(file.level(),
false)).toPath(file));
+ DataFilePathFactory pathFactory =
+ formatContext.pathFactory(new WriteFormatKey(file.level(),
false));
+ file.collectFiles(pathFactory).forEach(fileIO::deleteQuietly);
}
public FileIO getFileIO() {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java
index 86d8df0ce7..fd05478533 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java
@@ -33,6 +33,7 @@ import org.apache.paimon.utils.Pair;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import java.util.function.Function;
/**
@@ -55,7 +56,8 @@ public class KeyValueThinDataFileWriterImpl extends
KeyValueDataFileWriter {
CoreOptions options,
FileSource fileSource,
FileIndexOptions fileIndexOptions,
- boolean isExternalPath) {
+ boolean isExternalPath,
+ Set<String> managedBlobFields) {
super(
fileIO,
context,
@@ -69,7 +71,8 @@ public class KeyValueThinDataFileWriterImpl extends
KeyValueDataFileWriter {
options,
fileSource,
fileIndexOptions,
- isExternalPath);
+ isExternalPath,
+ managedBlobFields);
Map<Integer, Integer> idToIndex = new
HashMap<>(valueType.getFieldCount());
for (int i = 0; i < valueType.getFieldCount(); i++) {
idToIndex.put(valueType.getFields().get(i).id(), i);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index beb2651f1f..cbd92ea0fb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -163,10 +163,11 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
@Override
public void write(KeyValue kv) throws Exception {
long sequenceNumber = newSequenceNumber();
- boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(),
kv.key(), kv.value());
+ InternalRow value = writerFactory.externalizeBlob(kv.valueKind(),
kv.value());
+ boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(),
kv.key(), value);
if (!success) {
flushWriteBuffer(false, false);
- success = writeBuffer.put(sequenceNumber, kv.valueKind(),
kv.key(), kv.value());
+ success = writeBuffer.put(sequenceNumber, kv.valueKind(),
kv.key(), value);
if (!success) {
throw new RuntimeException("Mem table is too small to hold a
single element.");
}
@@ -263,6 +264,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
waitCompaction = true;
}
trySyncLatestCompaction(waitCompaction);
+ writerFactory.prepareCommit();
return drainIncrement();
}
@@ -341,6 +343,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
@Override
public void close() throws Exception {
+ writerFactory.abortManagedBlobWrites();
// cancel compaction so that it does not block job cancelling
compactManager.cancelCompaction();
sync();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
b/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
index fb9f9f7589..a630d8543a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
@@ -259,6 +259,7 @@ public class LocalOrphanFilesClean extends OrphanFilesClean
{
return files.stream()
.filter(status -> !status.isDir())
+ .filter(status -> !isManagedBlobPack(status.getPath()))
.filter(this::oldEnough)
.map(status -> Pair.of(status.getPath(), status.getLen()))
.collect(Collectors.toList());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
index b146ffbac7..4245460225 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -19,6 +19,7 @@
package org.apache.paimon.operation;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
@@ -218,6 +219,10 @@ public abstract class OrphanFilesClean implements
Serializable {
}
protected void cleanFile(Path path) {
+ if (isManagedBlobPack(path)) {
+ return;
+ }
+
if (!dryRun) {
try {
if (fileIO.isDir(path)) {
@@ -234,6 +239,10 @@ public abstract class OrphanFilesClean implements
Serializable {
}
}
+ protected boolean isManagedBlobPack(Path path) {
+ return
path.getName().endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ }
+
protected Set<Snapshot> safelyGetAllSnapshots(String branch) throws
IOException {
FileStoreTable branchTable = table.switchToBranch(branch);
SnapshotManager snapshotManager = branchTable.snapshotManager();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index f9dd882b70..fb0eb2e6f2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -216,6 +216,7 @@ public class SchemaValidation {
Set<String> blobDescriptorFields =
validateBlobDescriptorFields(tableRowType, options);
Set<String> blobViewFields =
validateBlobViewFields(tableRowType, options,
blobDescriptorFields);
+ validatePrimaryKeyBlobConfiguration(schema, options);
Set<String> blobInlineFields = new HashSet<>(blobDescriptorFields);
blobInlineFields.addAll(blobViewFields);
@@ -1208,9 +1209,12 @@ public class SchemaValidation {
.map(DataField::name)
.collect(Collectors.toList());
if (!blobNames.isEmpty()) {
- checkArgument(
- options.dataEvolutionEnabled(),
- "Data evolution config must enabled for table with BLOB or
ARRAY<BLOB> type column.");
+ boolean primaryKeyManagedBlob = !schema.primaryKeys().isEmpty();
+ if (!primaryKeyManagedBlob) {
+ checkArgument(
+ options.dataEvolutionEnabled(),
+ "Data evolution config must enabled for table with
BLOB or ARRAY<BLOB> type column.");
+ }
checkArgument(
fields.size() > blobNames.size(),
"Table with BLOB or ARRAY<BLOB> type column must have
other normal columns.");
@@ -1300,6 +1304,61 @@ public class SchemaValidation {
return configured;
}
+ private static void validatePrimaryKeyBlobConfiguration(
+ TableSchema schema, CoreOptions options) {
+ if (schema.primaryKeys().isEmpty()) {
+ return;
+ }
+
+ Set<String> managedBlobFields =
+ fieldNamesInBlobFile(new RowType(schema.fields()),
options.blobInlineField());
+ if (managedBlobFields.isEmpty()) {
+ return;
+ }
+
+ List<String> primaryKeyBlobFields =
+ managedBlobFields.stream()
+ .filter(schema.primaryKeys()::contains)
+ .collect(Collectors.toList());
+ checkArgument(
+ primaryKeyBlobFields.isEmpty(),
+ "Managed BLOB fields cannot be primary keys: %s.",
+ primaryKeyBlobFields);
+
+ List<String> bucketKeyBlobFields =
+ managedBlobFields.stream()
+ .filter(schema.bucketKeys()::contains)
+ .collect(Collectors.toList());
+ checkArgument(
+ bucketKeyBlobFields.isEmpty(),
+ "Managed BLOB fields cannot be bucket keys: %s.",
+ bucketKeyBlobFields);
+
+ List<String> sequenceBlobFields =
+ managedBlobFields.stream()
+ .filter(options.sequenceField()::contains)
+ .collect(Collectors.toList());
+ checkArgument(
+ sequenceBlobFields.isEmpty(),
+ "Managed BLOB fields cannot be sequence fields: %s.",
+ sequenceBlobFields);
+
+ checkArgument(
+ options.mergeEngine() == MergeEngine.DEDUPLICATE,
+ "Primary-key managed BLOB tables only support the deduplicate
merge engine.");
+ checkArgument(
+ options.changelogProducer() == ChangelogProducer.NONE,
+ "Primary-key managed BLOB tables only support
changelog-producer 'none'.");
+ checkArgument(
+ options.dataFileExternalPaths() == null,
+ "Primary-key managed BLOB tables do not support '%s'.",
+ CoreOptions.DATA_FILE_EXTERNAL_PATHS.key());
+ checkArgument(
+ !options.pkClusteringOverride(),
+ "Primary-key managed BLOB tables do not support '%s'.",
+ CoreOptions.PK_CLUSTERING_OVERRIDE.key());
+ }
+
private static void validateIncrementalClustering(TableSchema schema,
CoreOptions options) {
if (options.clusteringIncrementalEnabled()) {
checkArgument(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceCollectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceCollectorTest.java
new file mode 100644
index 0000000000..4cefdd1b52
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceCollectorTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ManagedBlobReferenceCollector}. */
+class ManagedBlobReferenceCollectorTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testCollectExactManagedReferencesFromFinalRows() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ Path dataFile = new Path(bucketPath, "data-a.avro");
+ Path managedBlob = new Path(bucketPath, "data-b.managed.blob");
+ Path externalBlob = new
Path(tempDir.resolve("external/data-c.managed.blob").toUri());
+ ManagedBlobReferenceCollector collector =
+ new ManagedBlobReferenceCollector(
+ fileIO,
+ dataFile,
+ RowType.of(DataTypes.INT(), DataTypes.BLOB()),
+ Collections.singleton("f1"));
+
+ collector.write(keyValue(RowKind.INSERT, Blob.fromFile(fileIO,
managedBlob.toString())));
+ collector.write(
+ keyValue(RowKind.UPDATE_AFTER, Blob.fromFile(fileIO,
managedBlob.toString())));
+ collector.write(keyValue(RowKind.INSERT, Blob.fromFile(fileIO,
externalBlob.toString())));
+ collector.write(keyValue(RowKind.DELETE, Blob.fromFile(fileIO,
managedBlob.toString())));
+ collector.close();
+
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ assertThat(collector.result()).isEqualTo(sidecar.getName());
+ assertThat(ManagedBlobReferenceFile.read(fileIO, sidecar))
+ .containsExactly(
+ new ManagedBlobReferenceFile.Reference(
+ bucketPath.toString(), managedBlob.getName()),
+ new ManagedBlobReferenceFile.Reference(
+ externalBlob.getParent().toString(),
externalBlob.getName()));
+ }
+
+ @Test
+ void testCollectManagedReferencesFromBlobArray() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ Path otherBucketPath = new Path(tempDir.resolve("bucket-1").toUri());
+ fileIO.mkdirs(bucketPath);
+ Path dataFile = new Path(bucketPath, "data-a.avro");
+ Path first = new Path(bucketPath, "data-b.managed.blob");
+ Path second = new Path(otherBucketPath, "data-c.managed.blob");
+ Path external = new
Path(tempDir.resolve("external/data-d.blob").toUri());
+ ManagedBlobReferenceCollector collector =
+ new ManagedBlobReferenceCollector(
+ fileIO,
+ dataFile,
+ RowType.of(DataTypes.INT(),
DataTypes.ARRAY(DataTypes.BLOB())),
+ Collections.singleton("f1"));
+
+ collector.write(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ new GenericArray(
+ new Object[] {
+ Blob.fromFile(fileIO,
first.toString()),
+ null,
+ Blob.fromFile(fileIO,
external.toString()),
+ Blob.fromFile(fileIO,
second.toString())
+ }))));
+ collector.close();
+
+ assertThat(
+ ManagedBlobReferenceFile.read(
+ fileIO,
ManagedBlobReferenceFile.sidecarPath(dataFile)))
+ .containsExactly(
+ new ManagedBlobReferenceFile.Reference(
+ bucketPath.toString(), first.getName()),
+ new ManagedBlobReferenceFile.Reference(
+ otherBucketPath.toString(), second.getName()));
+ }
+
+ @Test
+ void testCollectsOnlyDeclaredFields() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ Path dataFile = new Path(bucketPath, "data-a.avro");
+ Path managed = new Path(bucketPath, "data-b.managed.blob");
+ Path unmanaged = new Path(bucketPath, "data-c.managed.blob");
+ RowType valueType =
+ RowType.of(
+ new org.apache.paimon.types.DataType[] {
+ DataTypes.INT(), DataTypes.BLOB(), DataTypes.BLOB()
+ },
+ new String[] {"id", "managed", "unmanaged"});
+ ManagedBlobReferenceCollector collector =
+ new ManagedBlobReferenceCollector(
+ fileIO, dataFile, valueType,
Collections.singleton("managed"));
+
+ collector.write(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ Blob.fromFile(fileIO,
managed.toString()),
+ Blob.fromFile(fileIO,
unmanaged.toString()))));
+ collector.close();
+
+ assertThat(
+ ManagedBlobReferenceFile.read(
+ fileIO,
ManagedBlobReferenceFile.sidecarPath(dataFile)))
+ .containsExactly(
+ new ManagedBlobReferenceFile.Reference(
+ bucketPath.toString(), managed.getName()));
+ }
+
+ private KeyValue keyValue(RowKind kind, Blob blob) {
+ return new KeyValue().replace(GenericRow.of(1), kind, GenericRow.of(1,
blob));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceFileTest.java
new file mode 100644
index 0000000000..bec0c403ca
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReferenceFileTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.DataOutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ManagedBlobReferenceFile}. */
+class ManagedBlobReferenceFileTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testRoundTripAndDeduplicateReferences() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.resolve("data.avro.blobref").toUri());
+ ManagedBlobReferenceFile.Reference first =
+ new ManagedBlobReferenceFile.Reference(
+ tempDir.resolve("bucket-0").toUri().toString(),
"data-a.managed.blob");
+ ManagedBlobReferenceFile.Reference second =
+ new ManagedBlobReferenceFile.Reference(
+ tempDir.resolve("bucket-0").toUri().toString(),
"data-b.managed.blob");
+
+ ManagedBlobReferenceFile.write(fileIO, path, Arrays.asList(second,
first, second));
+
+ assertThat(ManagedBlobReferenceFile.read(fileIO,
path)).containsExactly(first, second);
+
+ Path emptyPath = new
Path(tempDir.resolve("empty.avro.blobref").toUri());
+ ManagedBlobReferenceFile.write(fileIO, emptyPath,
Collections.emptyList());
+ assertThat(ManagedBlobReferenceFile.read(fileIO, emptyPath)).isEmpty();
+ }
+
+ @Test
+ void testClassifyManagedBlobPathAcrossDirectories() {
+ Path dataFile = new
Path(tempDir.resolve("bucket-0/data-a.avro").toUri());
+ Path managedBlob = new
Path(tempDir.resolve("bucket-0/data-b.managed.blob").toUri());
+ Path externalManagedBlob =
+ new
Path(tempDir.resolve("external/data-c.managed.blob").toUri());
+ Path ordinaryBlob = new
Path(tempDir.resolve("bucket-0/data-d.blob").toUri());
+
+
assertThat(ManagedBlobReferenceFile.fromDescriptorUri(managedBlob.toString()))
+ .contains(
+ new ManagedBlobReferenceFile.Reference(
+ dataFile.getParent().toString(),
managedBlob.getName()));
+
assertThat(ManagedBlobReferenceFile.fromDescriptorUri(externalManagedBlob.toString()))
+ .contains(
+ new ManagedBlobReferenceFile.Reference(
+ externalManagedBlob.getParent().toString(),
+ externalManagedBlob.getName()));
+
assertThat(ManagedBlobReferenceFile.fromDescriptorUri(ordinaryBlob.toString())).isEmpty();
+ assertThat(ManagedBlobReferenceFile.sidecarPath(dataFile).getName())
+ .isEqualTo("data-a.avro.blobref");
+ }
+
+ @Test
+ void testRejectUnsupportedVersion() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new
Path(tempDir.resolve("unsupported.avro.blobref").toUri());
+ try (DataOutputStream out = new
DataOutputStream(fileIO.newOutputStream(path, false))) {
+ out.writeInt(0x50424C52);
+ out.writeByte(99);
+ out.writeInt(0);
+ }
+
+ assertThatThrownBy(() -> ManagedBlobReferenceFile.read(fileIO, path))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessage("Unsupported managed BLOB reference file version:
99");
+ }
+
+ @Test
+ void testRejectCorruptChecksum() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.resolve("corrupt.avro.blobref").toUri());
+ try (DataOutputStream out = new
DataOutputStream(fileIO.newOutputStream(path, false))) {
+ out.writeInt(0x50424C52);
+ out.writeByte(1);
+ out.writeInt(0);
+ out.writeInt(12345);
+ }
+
+ assertThatThrownBy(() -> ManagedBlobReferenceFile.read(fileIO, path))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessageContaining("Invalid managed BLOB reference file
checksum");
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java
b/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java
new file mode 100644
index 0000000000..bb01bb4c06
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java
@@ -0,0 +1,336 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobArrayPlaceholder;
+import org.apache.paimon.data.BlobRef;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PrimaryKeyBlobExternalizer}. */
+class PrimaryKeyBlobExternalizerTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testRejectsNonBlobManagedField() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+
+ assertThatThrownBy(
+ () ->
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT()),
+ Collections.singleton("f0"),
+ pathFactory,
+ 1024L))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Managed BLOB field 'f0' must be BLOB or
ARRAY<BLOB>, but was INT.");
+ }
+
+ @Test
+ void testRejectsUnknownManagedField() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+
+ assertThatThrownBy(
+ () ->
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.BLOB()),
+ Collections.singleton("missing"),
+ pathFactory,
+ 1024L))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Managed BLOB fields do not exist in value type:
[missing].");
+ }
+
+ @Test
+ void testExternalizeRawBlobBeforeBuffering() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ RowType valueType = RowType.of(DataTypes.INT(), DataTypes.BLOB());
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO, valueType, Collections.singleton("f1"),
pathFactory, 1024L);
+ byte[] expected = "managed-blob".getBytes(StandardCharsets.UTF_8);
+
+ InternalRow result =
+ externalizer.externalize(RowKind.INSERT, GenericRow.of(1,
Blob.fromData(expected)));
+ Blob blob = result.getBlob(1);
+
+ assertThat(blob).isInstanceOf(BlobRef.class);
+ assertThat(blob.toDescriptor().uri())
+ .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+
+ externalizer.prepareCommit();
+
+ assertThat(blob.toData()).isEqualTo(expected);
+ assertThat(fileIO.exists(new
Path(blob.toDescriptor().uri()))).isTrue();
+ }
+
+ @Test
+ void testRematerializesBlobRefInput() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT(), DataTypes.BLOB()),
+ Collections.singleton("f1"),
+ pathFactory,
+ 1024L);
+ byte[] expected = "external-blob".getBytes(StandardCharsets.UTF_8);
+ Path source = new Path(tempDir.resolve("external.blob").toUri());
+ try (org.apache.paimon.fs.PositionOutputStream out =
+ fileIO.newOutputStream(source, false)) {
+ out.write(expected);
+ }
+ BlobRef blobRef = (BlobRef) Blob.fromFile(fileIO, source.toString());
+
+ Blob result =
+ externalizer.externalize(RowKind.INSERT, GenericRow.of(1,
blobRef)).getBlob(1);
+
+ assertThat(result).isInstanceOf(BlobRef.class);
+ assertThat(result.toDescriptor().uri())
+ .isNotEqualTo(source.toString())
+ .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ externalizer.prepareCommit();
+ assertThat(result.toData()).isEqualTo(expected);
+ }
+
+ @Test
+ void testExternalizesOnlyDeclaredFields() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ RowType valueType =
+ RowType.of(
+ new org.apache.paimon.types.DataType[] {
+ DataTypes.INT(), DataTypes.BLOB(), DataTypes.BLOB()
+ },
+ new String[] {"id", "managed", "unmanaged"});
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO, valueType, Collections.singleton("managed"),
pathFactory, 1024L);
+ Blob unmanaged = Blob.fromData(new byte[] {2});
+
+ InternalRow result =
+ externalizer.externalize(
+ RowKind.INSERT, GenericRow.of(1, Blob.fromData(new
byte[] {1}), unmanaged));
+
+ assertThat(result.getBlob(1)).isInstanceOf(BlobRef.class);
+ assertThat(result.getBlob(2)).isSameAs(unmanaged);
+ }
+
+ @Test
+ void testRetractSkipsPayloadAndAbortDeletesPrivatePack() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT(), DataTypes.BLOB()),
+ Collections.singleton("f1"),
+ pathFactory,
+ 1024L);
+
+ InternalRow retract =
+ externalizer.externalize(
+ RowKind.DELETE, GenericRow.of(1, Blob.fromData(new
byte[] {1})));
+ assertThat(retract.isNullAt(1)).isTrue();
+ assertThat(fileIO.listStatus(bucketPath)).isEmpty();
+
+ Blob privateBlob =
+ externalizer
+ .externalize(
+ RowKind.INSERT, GenericRow.of(2,
Blob.fromData(new byte[] {2})))
+ .getBlob(1);
+ Path privatePack = new Path(privateBlob.toDescriptor().uri());
+ assertThat(fileIO.exists(privatePack)).isTrue();
+
+ externalizer.abort();
+
+ assertThat(fileIO.exists(privatePack)).isFalse();
+ }
+
+ @Test
+ void testExternalizeBlobArrayElements() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT(),
DataTypes.ARRAY(DataTypes.BLOB())),
+ Collections.singleton("f1"),
+ pathFactory,
+ 1024L);
+ byte[] expected = "array-element".getBytes(StandardCharsets.UTF_8);
+ byte[] second =
"second-array-element".getBytes(StandardCharsets.UTF_8);
+
+ InternalRow result =
+ externalizer.externalize(
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ new GenericArray(
+ new Object[] {
+ Blob.fromData(expected), null,
Blob.fromData(second)
+ })));
+ InternalArray blobs = result.getArray(1);
+
+ assertThat(blobs.size()).isEqualTo(3);
+ assertThat(blobs.getBlob(0)).isInstanceOf(BlobRef.class);
+ assertThat(blobs.getBlob(0).toDescriptor().uri())
+ .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ assertThat(blobs.isNullAt(1)).isTrue();
+ assertThat(blobs.getBlob(2)).isInstanceOf(BlobRef.class);
+
+ externalizer.prepareCommit();
+ assertThat(blobs.getBlob(0).toData()).isEqualTo(expected);
+ assertThat(blobs.getBlob(2).toData()).isEqualTo(second);
+ }
+
+ @Test
+ void testRematerializesBlobRefArrayElement() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT(),
DataTypes.ARRAY(DataTypes.BLOB())),
+ Collections.singleton("f1"),
+ pathFactory,
+ 1024L);
+ byte[] expected =
"external-array-blob".getBytes(StandardCharsets.UTF_8);
+ Path source = new Path(tempDir.resolve("external-array.blob").toUri());
+ try (org.apache.paimon.fs.PositionOutputStream out =
+ fileIO.newOutputStream(source, false)) {
+ out.write(expected);
+ }
+ BlobRef blobRef = (BlobRef) Blob.fromFile(fileIO, source.toString());
+
+ InternalArray result =
+ externalizer
+ .externalize(
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ new GenericArray(
+ new Object[] {
+ Blob.fromData(new byte[]
{1}), blobRef
+ })))
+ .getArray(1);
+
+ assertThat(result.getBlob(1)).isInstanceOf(BlobRef.class);
+ assertThat(result.getBlob(1).toDescriptor().uri())
+ .isNotEqualTo(source.toString())
+ .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ externalizer.prepareCommit();
+ assertThat(result.getBlob(1).toData()).isEqualTo(expected);
+ }
+
+ @Test
+ void testBlobArrayNullEmptyRetractAndPlaceholder() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path bucketPath = new Path(tempDir.resolve("bucket-0").toUri());
+ fileIO.mkdirs(bucketPath);
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ bucketPath, "avro", "data-", "changelog-", false,
null, null);
+ PrimaryKeyBlobExternalizer externalizer =
+ new PrimaryKeyBlobExternalizer(
+ fileIO,
+ RowType.of(DataTypes.INT(),
DataTypes.ARRAY(DataTypes.BLOB())),
+ Collections.singleton("f1"),
+ pathFactory,
+ 1024L);
+
+ GenericRow nullArray = GenericRow.of(1, null);
+ GenericRow emptyArray = GenericRow.of(2, new GenericArray(new
Object[0]));
+ GenericRow nullElement = GenericRow.of(3, new GenericArray(new
Object[] {null}));
+ assertThat(externalizer.externalize(RowKind.INSERT,
nullArray)).isSameAs(nullArray);
+ assertThat(externalizer.externalize(RowKind.INSERT,
emptyArray)).isSameAs(emptyArray);
+ assertThat(externalizer.externalize(RowKind.INSERT,
nullElement)).isSameAs(nullElement);
+
+ InternalRow retract =
+ externalizer.externalize(
+ RowKind.DELETE, GenericRow.of(4,
BlobArrayPlaceholder.INSTANCE));
+ assertThat(retract.isNullAt(1)).isTrue();
+
+ assertThatThrownBy(
+ () ->
+ externalizer.externalize(
+ RowKind.INSERT,
+ GenericRow.of(5,
BlobArrayPlaceholder.INSTANCE)))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("placeholder blob array");
+ assertThat(fileIO.listStatus(bucketPath)).isEmpty();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java
new file mode 100644
index 0000000000..22c549bf3f
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobRef;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FlushingFileFormat;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+import java.util.function.Function;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for managed BLOB references produced by {@link
KeyValueFileWriterFactory}. */
+class PrimaryKeyBlobFileWriterTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testAttachReferenceSidecarToDataFile() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ RowType keyType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ SpecialFields.KEY_FIELD_ID_START,
+ SpecialFields.KEY_FIELD_PREFIX + "id",
+ DataTypes.INT())));
+ RowType valueType =
+ new RowType(
+ java.util.Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()),
+ new DataField(2, "descriptor",
DataTypes.BLOB())));
+ Function<String, FileStorePathFactory> pathFactories =
+ format ->
+ new FileStorePathFactory(
+ tablePath,
+ RowType.of(),
+
CoreOptions.PARTITION_DEFAULT_NAME.defaultValue(),
+ format,
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+
CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null);
+ Options options = new Options();
+ options.set(CoreOptions.BLOB_FIELD, "payload");
+ options.set(CoreOptions.BLOB_DESCRIPTOR_FIELD, "descriptor");
+ KeyValueFileWriterFactory factory =
+ KeyValueFileWriterFactory.builder(
+ fileIO,
+ 0,
+ keyType,
+ valueType,
+ new FlushingFileFormat("avro"),
+ pathFactories,
+ 1024 * 1024)
+ .build(BinaryRow.EMPTY_ROW, 0, new
CoreOptions(options));
+ DataFilePathFactory pathFactory = factory.pathFactory(0);
+ Path descriptorPath =
+
pathFactory.newPathFromExtension(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ Blob descriptor = Blob.fromFile(fileIO, descriptorPath.toString());
+ InternalRow externalized =
+ factory.externalizeBlob(
+ RowKind.INSERT,
+ GenericRow.of(1, Blob.fromData(new byte[] {1}),
descriptor));
+ assertThat(externalized.getBlob(1)).isInstanceOf(BlobRef.class);
+ assertThat(externalized.getBlob(2)).isSameAs(descriptor);
+ Path managedBlob = new
Path(externalized.getBlob(1).toDescriptor().uri());
+
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ factory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
+
+ writer.write(new KeyValue().replace(GenericRow.of(1), 0,
RowKind.INSERT, externalized));
+ writer.close();
+ factory.prepareCommit();
+
+ DataFileMeta meta = writer.result().get(0);
+
assertThat(meta.extraFiles()).singleElement().asString().endsWith(".blobref");
+ Path sidecar = pathFactory.toAlignedPath(meta.extraFiles().get(0),
meta);
+ assertThat(ManagedBlobReferenceFile.read(fileIO, sidecar))
+ .containsExactly(
+ new ManagedBlobReferenceFile.Reference(
+ sidecar.getParent().toString(),
managedBlob.getName()));
+
+ Path dataPath = pathFactory.toPath(meta);
+ factory.deleteFile(meta);
+ assertThat(fileIO.exists(dataPath)).isFalse();
+ assertThat(fileIO.exists(sidecar)).isFalse();
+
+ RollingFileWriter<KeyValue, DataFileMeta> emptyWriter =
+ factory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
+ emptyWriter.write(
+ new KeyValue()
+ .replace(
+ GenericRow.of(2), 1, RowKind.INSERT,
GenericRow.of(2, null, null)));
+ emptyWriter.close();
+
+ DataFileMeta emptyMeta = emptyWriter.result().get(0);
+
assertThat(emptyMeta.extraFiles()).singleElement().asString().endsWith(".blobref");
+ Path emptySidecar =
pathFactory.toAlignedPath(emptyMeta.extraFiles().get(0), emptyMeta);
+ assertThat(ManagedBlobReferenceFile.read(fileIO,
emptySidecar)).isEmpty();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
index b690246101..cd6480d7e7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.Changelog;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.ExternalPathStrategy;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.DataFormatTestUtil;
import org.apache.paimon.data.GenericRow;
@@ -152,6 +153,29 @@ public class LocalOrphanFilesCleanTest {
normallyRemoving(tablePath);
}
+ @Test
+ public void testKeepManagedBlobPack() throws Exception {
+ commit(Collections.singletonList(TestPojo.next()));
+
+ Path part1 = listSubDirs(tablePath, p ->
p.getName().contains("=")).get(0);
+ Path part2 = listSubDirs(part1, p -> p.getName().contains("=")).get(0);
+ Path bucket = listSubDirs(part2, p ->
p.getName().startsWith(BUCKET_PATH_PREFIX)).get(0);
+ Path managedBlob =
+ new Path(bucket, "orphan" +
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ Path ordinaryOrphan = new Path(bucket, "orphan.avro");
+ fileIO.newOutputStream(managedBlob, false).close();
+ fileIO.newOutputStream(ordinaryOrphan, false).close();
+
+ LocalOrphanFilesClean cleaner =
+ new LocalOrphanFilesClean(
+ table, System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(2));
+ List<Path> deleted = cleaner.clean().getDeletedFilesPath();
+
+ assertThat(fileIO.exists(managedBlob)).isTrue();
+ assertThat(fileIO.exists(ordinaryOrphan)).isFalse();
+ assertThat(deleted).doesNotContain(managedBlob);
+ }
+
public void normallyRemoving(Path dataPath) throws Throwable {
int commitTimes = 30;
List<List<TestPojo>> committedData = new ArrayList<>();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
new file mode 100644
index 0000000000..333020d29c
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
@@ -0,0 +1,316 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.TestFileStore;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
+import org.apache.paimon.schema.KeyValueFieldsExtractor;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for managed BLOB storage in primary-key tables. */
+class PrimaryKeyManagedBlobStoreTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testExternalizeRawBlobBeforeMergeTreeBuffer() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store = createStore(fileIO);
+ byte[] expected = "raw-before-buffer".getBytes(StandardCharsets.UTF_8);
+ KeyValue keyValue = keyValue(1, RowKind.INSERT, expected);
+
+ store.commitData(
+ Collections.singletonList(keyValue), ignored ->
BinaryRow.EMPTY_ROW, ignored -> 0);
+
+ ManifestEntry entry = store.newScan().plan().files().get(0);
+ List<ManagedBlobReferenceFile.Reference> references =
references(fileIO, store, entry);
+ assertThat(references).hasSize(1);
+ assertThat(references.get(0).relativePath())
+ .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+
+ KeyValue read =
+
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()).get(0);
+ assertThat(read.value().getBlob(1).toData()).isEqualTo(expected);
+ }
+
+ @Test
+ void testExternalizeAndReadBlobArray() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store = createArrayStore(fileIO);
+ byte[] expected = "array-payload".getBytes(StandardCharsets.UTF_8);
+ byte[] second =
"second-array-payload".getBytes(StandardCharsets.UTF_8);
+ Path source = new Path(tempDir.resolve("external-array.blob").toUri());
+ try (org.apache.paimon.fs.PositionOutputStream out =
+ fileIO.newOutputStream(source, false)) {
+ out.write(second);
+ }
+ Blob external = Blob.fromFile(fileIO, source.toString());
+ KeyValue keyValue =
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ new GenericArray(
+ new Object[] {
+ Blob.fromData(expected),
null, external
+ })));
+
+ store.commitData(
+ Collections.singletonList(keyValue), ignored ->
BinaryRow.EMPTY_ROW, ignored -> 0);
+
+ ManifestEntry entry = store.newScan().plan().files().get(0);
+ assertThat(references(fileIO, store, entry)).hasSize(2);
+ InternalArray blobs =
+
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId())
+ .get(0)
+ .value()
+ .getArray(1);
+ assertThat(blobs.size()).isEqualTo(3);
+ assertThat(blobs.getBlob(0).toData()).isEqualTo(expected);
+ assertThat(blobs.isNullAt(1)).isTrue();
+
assertThat(blobs.getBlob(2).toDescriptor().uri()).isNotEqualTo(source.toString());
+ assertThat(blobs.getBlob(2).toData()).isEqualTo(second);
+ }
+
+ @Test
+ void testCompactionRebuildsExactBlobReferences() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store = createStore(fileIO);
+
+ store.commitData(
+ Collections.singletonList(
+ keyValue(1, RowKind.INSERT,
"old".getBytes(StandardCharsets.UTF_8))),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> 0);
+ ManagedBlobReferenceFile.Reference oldReference =
+ references(fileIO, store,
store.newScan().plan().files().get(0)).get(0);
+
+ store.commitData(
+ Arrays.asList(
+ keyValue(1, RowKind.UPDATE_AFTER,
"new".getBytes(StandardCharsets.UTF_8)),
+ keyValue(2, RowKind.INSERT,
"second".getBytes(StandardCharsets.UTF_8))),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> 0);
+ List<ManagedBlobReferenceFile.Reference> survivingReferences = new
java.util.ArrayList<>();
+ for (ManifestEntry file : store.newScan().plan().files()) {
+ survivingReferences.addAll(references(fileIO, store, file));
+ }
+ survivingReferences.remove(oldReference);
+ assertThat(survivingReferences).hasSize(2);
+
+ forceFullCompaction(store);
+
+ List<ManifestEntry> files = store.newScan().plan().files();
+ assertThat(files).hasSize(1);
+ List<ManagedBlobReferenceFile.Reference> compactedReferences =
+ references(fileIO, store, files.get(0));
+ assertThat(compactedReferences)
+ .containsExactlyInAnyOrderElementsOf(survivingReferences)
+ .doesNotContain(oldReference);
+
assertThat(store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()))
+ .extracting(
+ kv -> new String(kv.value().getBlob(1).toData(),
StandardCharsets.UTF_8))
+ .containsExactlyInAnyOrder("new", "second");
+ }
+
+ @Test
+ void testBlobArrayCompactionRebuildsExactReferences() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store = createArrayStore(fileIO);
+
+ store.commitData(
+ Arrays.asList(
+ arrayKeyValue(1, RowKind.INSERT, "old"),
+ arrayKeyValue(2, RowKind.INSERT, "deleted")),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> 0);
+ List<ManagedBlobReferenceFile.Reference> oldReferences = new
java.util.ArrayList<>();
+ for (ManifestEntry file : store.newScan().plan().files()) {
+ oldReferences.addAll(references(fileIO, store, file));
+ }
+ assertThat(oldReferences).hasSize(2);
+
+ store.commitData(
+ Arrays.asList(
+ arrayKeyValue(1, RowKind.UPDATE_AFTER, "new"),
+ arrayKeyValue(2, RowKind.DELETE,
"must-not-be-written")),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> 0);
+ forceFullCompaction(store);
+
+ List<ManifestEntry> files = store.newScan().plan().files();
+ assertThat(files).hasSize(1);
+ assertThat(references(fileIO, store, files.get(0)))
+ .hasSize(1)
+ .doesNotContainAnyElementsOf(oldReferences);
+ List<KeyValue> rows =
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId());
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).value().getArray(1).getBlob(0).toData())
+ .isEqualTo("new".getBytes(StandardCharsets.UTF_8));
+ }
+
+ private TestFileStore createStore(FileIO fileIO) throws Exception {
+ return createStore(fileIO, "payload", DataTypes.BLOB());
+ }
+
+ private TestFileStore createArrayStore(FileIO fileIO) throws Exception {
+ return createStore(fileIO, "payloads",
DataTypes.ARRAY(DataTypes.BLOB()));
+ }
+
+ private TestFileStore createStore(FileIO fileIO, String payloadName,
DataType payloadType)
+ throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ List<DataField> valueFields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, payloadName, payloadType));
+ RowType valueType = new RowType(valueFields);
+ RowType keyType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ SpecialFields.KEY_FIELD_ID_START,
+ SpecialFields.KEY_FIELD_PREFIX + "id",
+ DataTypes.INT())));
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), payloadName);
+ options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b");
+ TableSchema schema =
+ new SchemaManager(fileIO, tablePath)
+ .createTable(
+ new Schema(
+ valueFields,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ ""));
+ KeyValueFieldsExtractor extractor =
+ new KeyValueFieldsExtractor() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public List<DataField> keyFields(TableSchema ignored) {
+ return keyType.getFields();
+ }
+
+ @Override
+ public List<DataField> valueFields(TableSchema
tableSchema) {
+ return tableSchema.fields();
+ }
+ };
+ return new TestFileStore.Builder(
+ "avro",
+ tempDir.toString(),
+ 1,
+ RowType.of(),
+ keyType,
+ valueType,
+ extractor,
+ DeduplicateMergeFunction.factory(),
+ schema)
+ .build();
+ }
+
+ private List<ManagedBlobReferenceFile.Reference> references(
+ FileIO fileIO, TestFileStore store, ManifestEntry entry) throws
Exception {
+ DataFileMeta dataFile = entry.file();
+ String referenceFile =
+ dataFile.extraFiles().stream()
+ .filter(
+ file ->
+ file.endsWith(
+
ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Missing managed
BLOB sidecar."));
+ DataFilePathFactory pathFactory =
+
store.pathFactory().createDataFilePathFactory(entry.partition(),
entry.bucket());
+ Path sidecar = pathFactory.toAlignedPath(referenceFile, dataFile);
+ return ManagedBlobReferenceFile.read(fileIO, sidecar);
+ }
+
+ private void forceFullCompaction(TestFileStore store) throws Exception {
+ AbstractFileStoreWrite<KeyValue> write = store.newWrite();
+ try {
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ List<CommitMessage> messages = write.prepareCommit(true, 1000L);
+ try (FileStoreCommit commit = store.newCommit()) {
+ commit.commit(new ManifestCommittable(1000L, null, messages),
false);
+ }
+ } finally {
+ write.close();
+ }
+ }
+
+ private KeyValue keyValue(int id, RowKind kind, byte[] bytes) {
+ return new KeyValue()
+ .replace(GenericRow.of(id), kind, GenericRow.of(id,
Blob.fromData(bytes)));
+ }
+
+ private KeyValue arrayKeyValue(int id, RowKind kind, String value) {
+ return new KeyValue()
+ .replace(
+ GenericRow.of(id),
+ kind,
+ GenericRow.of(
+ id,
+ new GenericArray(
+ new Object[] {
+
Blob.fromData(value.getBytes(StandardCharsets.UTF_8))
+ })));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 86a568e875..d13ae60bff 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -173,6 +173,173 @@ class SchemaValidationTest {
"Table with BLOB or ARRAY<BLOB> type column must have
other normal columns.");
}
+ @Test
+ public void testPrimaryKeyArrayBlobField() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payloads",
DataTypes.ARRAY(DataTypes.BLOB())));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payloads");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPrimaryKeyArrayBlobManagedByType() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payloads",
DataTypes.ARRAY(DataTypes.BLOB())));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPrimaryKeyBlobFileField() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), "payload");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPrimaryKeyBlobViewCoexistsWithManagedBlob() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()),
+ new DataField(2, "view", DataTypes.BLOB()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+ options.put(CoreOptions.BLOB_VIEW_FIELD.key(), "view");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPrimaryKeyBlobRejectsUnsupportedSemantics() {
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ options,
singletonList("payload"), emptyList())))
+ .hasMessage("Managed BLOB fields cannot be primary keys:
[payload].");
+
+ Map<String, String> sequenceOptions = new HashMap<>(options);
+ sequenceOptions.put(CoreOptions.SEQUENCE_FIELD.key(), "payload");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ sequenceOptions,
singletonList("id"), emptyList())))
+ .hasMessage("Managed BLOB fields cannot be sequence fields:
[payload].");
+
+ Map<String, String> mergeOptions = new HashMap<>(options);
+ mergeOptions.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ mergeOptions,
singletonList("id"), emptyList())))
+ .hasMessage(
+ "Primary-key managed BLOB tables only support the
deduplicate merge engine.");
+
+ Map<String, String> changelogOptions = new HashMap<>(options);
+ changelogOptions.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ changelogOptions,
+ singletonList("id"),
+ emptyList())))
+ .hasMessage(
+ "Primary-key managed BLOB tables only support
changelog-producer 'none'.");
+
+ Map<String, String> externalPathOptions = new HashMap<>(options);
+ externalPathOptions.put(CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(),
"file:///tmp/data");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ externalPathOptions,
+ singletonList("id"),
+ emptyList())))
+ .hasMessage(
+ "Primary-key managed BLOB tables do not support
'data-file.external-paths'.");
+
+ Map<String, String> clusteringOptions = new HashMap<>(options);
+ clusteringOptions.put(CoreOptions.PK_CLUSTERING_OVERRIDE.key(),
"true");
+ clusteringOptions.put(CoreOptions.CLUSTERING_COLUMNS.key(), "id");
+ clusteringOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ clusteringOptions,
+ singletonList("id"),
+ emptyList())))
+ .hasMessage(
+ "Primary-key managed BLOB tables do not support
'pk-clustering-override'.");
+ }
+
+ private TableSchema primaryKeyBlobSchema(
+ Map<String, String> options, List<String> primaryKeys,
List<String> partitionKeys) {
+ return new TableSchema(
+ 1,
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB())),
+ 10,
+ partitionKeys,
+ primaryKeys,
+ options,
+ "");
+ }
+
@Test
public void testPartialUpdateTableAggregateFunctionWithoutSequenceGroup() {
Map<String, String> options = new HashMap<>(2);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index c99f538b0a..12b18e5d70 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -1098,7 +1098,7 @@ public class FlinkCatalog extends AbstractCatalog {
List<String> blobFields = CoreOptions.blobField(options);
Set<String> blobFileFields = new
HashSet<>(CoreOptions.blobField(options));
Set<String> blobInlineFields = blobInlineFields(options);
- if (!blobFields.isEmpty()) {
+ if (!blobFields.isEmpty() && !schema.getPrimaryKey().isPresent()) {
checkArgument(
options.containsKey(CoreOptions.DATA_EVOLUTION_ENABLED.key()),
"When setting '"
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
index a30cc3e0dd..3ce2bf82f8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
@@ -282,7 +282,9 @@ public class FlinkOrphanFilesClean extends OrphanFilesClean
{
Path dirPath = new Path(dir);
List<FileStatus> files =
tryBestListingDirs(dirPath);
for (FileStatus file : files) {
- if (!file.isDir() &&
oldEnough(file)) {
+ if (!file.isDir()
+ &&
!isManagedBlobPack(file.getPath())
+ && oldEnough(file)) {
out.collect(
Tuple2.of(
file.getPath().toString(),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
index 3d63a3ffb4..e2ad151fdc 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
@@ -147,6 +147,46 @@ public class BlobTableITCase extends CatalogITCaseBase {
.containsExactly(Row.of(true));
}
+ @Test
+ public void testPrimaryKeyArrayBlobField() throws Exception {
+ batchSql(
+ "CREATE TABLE pk_array_blob_table ("
+ + "id INT, pictures ARRAY<BYTES>, "
+ + "PRIMARY KEY (id) NOT ENFORCED) "
+ + "WITH ('bucket'='1', 'blob-field'='pictures')");
+ String dataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(
+ 1,
+ new byte[][] {
+ new byte[] {72, 101, 108, 108,
111},
+ null,
+ new byte[] {89, 69}
+ })));
+ tEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE pk_array_blob_source "
+ + "(id INT, pictures ARRAY<BYTES>) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ dataId))
+ .await();
+
+ batchSql("INSERT INTO pk_array_blob_table SELECT * FROM
pk_array_blob_source");
+
+ assertThat(
+ batchSql(
+ "SELECT id, CARDINALITY(pictures),
pictures[1], pictures[2], pictures[3] "
+ + "FROM pk_array_blob_table"))
+ .containsExactly(
+ Row.of(
+ 1,
+ 3,
+ new byte[] {72, 101, 108, 108, 111},
+ null,
+ new byte[] {89, 69}));
+ }
+
@Test
public void testArrayBlobFieldWithDescriptorElements() throws Exception {
byte[] blobData = new byte[1024];
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
index a0e197e1da..89a45c38c6 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
@@ -316,8 +316,7 @@ public class ParquetRowDataWriter {
@Override
public void write(InternalArray arrayData, int ordinal) {
- // Currently we don't support BLOB inside arrays/maps.
- throw new UnsupportedOperationException("BLOB in array is not
supported.");
+ writeBlob(arrayData.getBlob(ordinal));
}
private void writeBlob(Blob blob) {
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFormatReadWriteTest.java
index 62c9da064e..4a7d724635 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFormatReadWriteTest.java
@@ -28,6 +28,8 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.junit.jupiter.api.Test;
+
import java.util.ArrayList;
import java.util.List;
@@ -43,6 +45,11 @@ public class AvroFormatReadWriteTest extends
FormatReadWriteTest {
return new AvroFileFormat(new FileFormatFactory.FormatContext(new
Options(), 1024, 1024));
}
+ @Test
+ public void testArrayBlobDescriptors() throws Exception {
+ testArrayBlobDescriptorRoundTrip();
+ }
+
@Override
protected RowType rowTypeForFullTypesTest() {
RowType rowWithoutVector = super.rowTypeForFullTypesTest();
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
index e6e1dc0a14..65711a67c0 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
@@ -57,6 +57,11 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** An orc {@link FormatReadWriteTest}. */
public class OrcFormatReadWriteTest extends FormatReadWriteTest {
+ @Test
+ public void testArrayBlobDescriptors() throws Exception {
+ testArrayBlobDescriptorRoundTrip();
+ }
+
private final FileFormat legacyFormat =
new OrcFileFormat(
new FileFormatFactory.FormatContext(
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
index fd1552189f..c320a766d5 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
@@ -63,6 +63,11 @@ public class ParquetFormatReadWriteTest extends
FormatReadWriteTest {
new FileFormatFactory.FormatContext(new Options(), 1024,
1024));
}
+ @Test
+ public void testArrayBlobDescriptors() throws Exception {
+ testArrayBlobDescriptorRoundTrip();
+ }
+
@Test
public void testWriteMetadata() throws Exception {
ParquetFileFormat format =
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala
index 98d049d619..428ac6e097 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala
@@ -127,6 +127,7 @@ case class SparkOrphanFilesClean(
dir =>
tryBestListingDirs(new Path(dir)).asScala
.filter(file => !file.isDir())
+ .filter(file => !isManagedBlobPack(file.getPath))
.filter(oldEnough)
.map {
file =>
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala
index 72867b7ff0..feb8acfbfb 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala
@@ -129,6 +129,26 @@ class BlobTestBase extends PaimonSparkTestBase {
}
}
+ test("Blob: test primary-key array blob") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, pictures ARRAY<BINARY>) TBLPROPERTIES (" +
+ "'primary-key'='id', 'bucket'='1', 'blob-field'='pictures')")
+ sql(
+ "INSERT INTO t VALUES " +
+ "(1, array(X'48656C6C6F', CAST(NULL AS BINARY), X'5945'))")
+
+ val row =
+ sql("SELECT id, size(pictures), pictures[0], pictures[1], pictures[2]
FROM t")
+ .collect()(0)
+ assert(row.getInt(0) == 1)
+ assert(row.getInt(1) == 3)
+ assert(util.Arrays.equals(row.getAs[Array[Byte]](2), Array[Byte](72,
101, 108, 108, 111)))
+ assert(row.isNullAt(3))
+ assert(util.Arrays.equals(row.getAs[Array[Byte]](4), Array[Byte](89,
69)))
+ }
+ }
+
test("Blob: array blob writes descriptor elements and reads descriptors") {
withTable("t") {
val blobData = new Array[Byte](1024)
