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 83cd9965a2 [core] Harden managed BLOB writes in primary-key tables
(#8654)
83cd9965a2 is described below
commit 83cd9965a2388ab86f6129d47d60c49bebf7f6af
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 18:22:18 2026 +0800
[core] Harden managed BLOB writes in primary-key tables (#8654)
Follow up on managed BLOB support for primary-key tables by covering the
postpone-bucket write path and tightening resource and reference
handling.
The postpone-bucket writer previously bypassed managed BLOB
externalization, so raw `BLOB` and `ARRAY<BLOB>` values could reach
descriptor-backed data-file writers. Pack finalization could also leave
the underlying output stream open when flush failed, while reference
collection repeatedly parsed identical descriptor URIs and created
redundant deduplication containers.
---
.../paimon/blob/ManagedBlobReferenceCollector.java | 16 +++-
.../paimon/blob/ManagedBlobReferenceFile.java | 12 ++-
.../paimon/blob/PrimaryKeyBlobExternalizer.java | 6 +-
.../paimon/io/KeyValueFileWriterFactory.java | 4 +
.../paimon/postpone/PostponeBucketWriter.java | 34 +++++--
.../blob/PrimaryKeyBlobExternalizerTest.java | 47 ++++++++++
.../operation/PrimaryKeyManagedBlobStoreTest.java | 102 ++++++++++++++++++++-
.../paimon/postpone/PostponeBucketWriterTest.java | 80 ++++++++++++++++
8 files changed, 280 insertions(+), 21 deletions(-)
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
index 38b898aea2..21fc29dead 100644
---
a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java
@@ -43,7 +43,7 @@ public class ManagedBlobReferenceCollector {
private final Path sidecar;
private final int[] blobFieldIndexes;
private final boolean[] blobArrayFields;
- private final Set<ManagedBlobReferenceFile.Reference> references;
+ private final Set<String> descriptorUris;
private boolean closed;
@@ -68,7 +68,7 @@ public class ManagedBlobReferenceCollector {
for (int i = 0; i < arrayFields.size(); i++) {
blobArrayFields[i] = arrayFields.get(i);
}
- this.references = new HashSet<>();
+ this.descriptorUris = new HashSet<>();
}
public void write(KeyValue keyValue) {
@@ -97,8 +97,7 @@ public class ManagedBlobReferenceCollector {
private void collect(Blob blob) {
if (blob instanceof BlobRef) {
-
ManagedBlobReferenceFile.fromDescriptorUri(blob.toDescriptor().uri())
- .ifPresent(references::add);
+ descriptorUris.add(blob.toDescriptor().uri());
}
}
@@ -107,7 +106,14 @@ public class ManagedBlobReferenceCollector {
return;
}
try {
- ManagedBlobReferenceFile.write(fileIO, sidecar, new
ArrayList<>(references));
+ List<ManagedBlobReferenceFile.Reference> references =
+ new ArrayList<>(descriptorUris.size());
+ for (String descriptorUri : descriptorUris) {
+ ManagedBlobReferenceFile.fromDescriptorUri(descriptorUri)
+ .ifPresent(references::add);
+ }
+ descriptorUris.clear();
+ ManagedBlobReferenceFile.write(fileIO, sidecar, references);
closed = true;
} catch (IOException e) {
abort();
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
index d04457ecc2..f7b10512c6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
@@ -26,7 +26,6 @@ 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;
@@ -61,10 +60,19 @@ public class ManagedBlobReferenceFile {
public static void write(FileIO fileIO, Path path, List<Reference>
references)
throws IOException {
- List<Reference> normalized = new ArrayList<>(new
HashSet<>(references));
+ List<Reference> normalized = new ArrayList<>(references);
normalized.sort(
Comparator.comparing(Reference::storageRootId)
.thenComparing(Reference::relativePath));
+ int uniqueCount = 0;
+ for (Reference reference : normalized) {
+ if (uniqueCount == 0 ||
!reference.equals(normalized.get(uniqueCount - 1))) {
+ normalized.set(uniqueCount++, reference);
+ }
+ }
+ if (uniqueCount < normalized.size()) {
+ normalized.subList(uniqueCount, normalized.size()).clear();
+ }
try (DataOutputStream out = new
DataOutputStream(fileIO.newOutputStream(path, false))) {
out.writeInt(MAGIC);
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
index b09c59b349..9216f7597c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java
@@ -279,10 +279,10 @@ public class PrimaryKeyBlobExternalizer {
if (writer == null) {
return;
}
- try {
+ PositionOutputStream currentOut = out;
+ try (PositionOutputStream ignored = currentOut) {
writer.close();
- out.flush();
- out.close();
+ currentOut.flush();
} finally {
writer = null;
out = null;
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 0e17ec5e4b..547d827a73 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
@@ -107,6 +107,10 @@ public class KeyValueFileWriterFactory {
return valueType;
}
+ public boolean hasBlobExternalizer() {
+ return blobExternalizer != null;
+ }
+
public InternalRow externalizeBlob(RowKind valueKind, InternalRow value)
throws IOException {
return blobExternalizer == null ? value :
blobExternalizer.externalize(valueKind, value);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
index 47256749b9..47068513a8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java
@@ -22,6 +22,7 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.KeyValueSerializer;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.CompactIncrement;
@@ -54,8 +55,6 @@ import java.util.List;
/** {@link RecordWriter} for {@code bucket = -2} tables. */
public class PostponeBucketWriter implements RecordWriter<KeyValue>,
MemoryOwner {
- private final FileIO fileIO;
- private final DataFilePathFactory pathFactory;
private final MergeFunction<KeyValue> mergeFunction;
private final KeyValueFileWriterFactory writerFactory;
private final List<DataFileMeta> files;
@@ -84,8 +83,6 @@ public class PostponeBucketWriter implements
RecordWriter<KeyValue>, MemoryOwner
this.mergeFunction = mergeFunction;
this.writerFactory = writerFactory;
this.fileRead = fileRead;
- this.fileIO = fileIO;
- this.pathFactory = pathFactory;
this.spillCompression = spillCompression;
this.maxDiskSize = maxDiskSize;
this.files = new ArrayList<>();
@@ -104,11 +101,25 @@ public class PostponeBucketWriter implements
RecordWriter<KeyValue>, MemoryOwner
@Override
public void write(KeyValue record) throws Exception {
- validateRetract(record);
- boolean success = sinkWriter.write(record);
+ KeyValue externalizedRecord = record;
+ if (writerFactory.hasBlobExternalizer()) {
+ InternalRow value =
writerFactory.externalizeBlob(record.valueKind(), record.value());
+ if (value != record.value()) {
+ externalizedRecord =
+ new KeyValue()
+ .replace(
+ record.key(),
+ record.sequenceNumber(),
+ record.valueKind(),
+ value)
+ .setLevel(record.level());
+ }
+ }
+ validateRetract(externalizedRecord);
+ boolean success = sinkWriter.write(externalizedRecord);
if (!success) {
flush();
- success = sinkWriter.write(record);
+ success = sinkWriter.write(externalizedRecord);
if (!success) {
// Should not get here, because writeBuffer will throw too big
exception out.
// But we throw again in case of something unexpected happens.
(like someone changed
@@ -205,7 +216,7 @@ public class PostponeBucketWriter implements
RecordWriter<KeyValue>, MemoryOwner
} finally {
// remove small files
for (DataFileMeta file : files) {
- fileIO.deleteQuietly(pathFactory.toPath(file));
+ writerFactory.deleteFile(file);
}
}
}
@@ -214,6 +225,7 @@ public class PostponeBucketWriter implements
RecordWriter<KeyValue>, MemoryOwner
@Override
public CommitIncrement prepareCommit(boolean waitCompaction) throws
Exception {
flush();
+ writerFactory.prepareCommit();
List<DataFileMeta> result = new ArrayList<>(files);
files.clear();
return new CommitIncrement(
@@ -237,6 +249,10 @@ public class PostponeBucketWriter implements
RecordWriter<KeyValue>, MemoryOwner
@Override
public void close() throws Exception {
- sinkWriter.close();
+ try {
+ writerFactory.abortManagedBlobWrites();
+ } finally {
+ sinkWriter.close();
+ }
}
}
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
index bb01bb4c06..5fe4686c9f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java
@@ -26,6 +26,8 @@ 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.PositionOutputStream;
+import org.apache.paimon.fs.PositionOutputStreamWrapper;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.types.DataTypes;
@@ -35,8 +37,10 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -46,6 +50,49 @@ class PrimaryKeyBlobExternalizerTest {
@TempDir java.nio.file.Path tempDir;
+ @Test
+ void testClosesPackStreamWhenFlushFails() throws Exception {
+ AtomicBoolean closed = new AtomicBoolean();
+ LocalFileIO fileIO =
+ new LocalFileIO() {
+ @Override
+ public PositionOutputStream newOutputStream(Path path,
boolean overwrite)
+ throws IOException {
+ return new PositionOutputStreamWrapper(
+ super.newOutputStream(path, overwrite)) {
+ @Override
+ public void flush() throws IOException {
+ throw new IOException("flush failure");
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed.set(true);
+ super.close();
+ }
+ };
+ }
+ };
+ 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.BLOB()),
+ Collections.singleton("f0"),
+ pathFactory,
+ 1024L);
+ externalizer.externalize(RowKind.INSERT,
GenericRow.of(Blob.fromData(new byte[] {1})));
+
+ assertThatThrownBy(externalizer::prepareCommit)
+ .isInstanceOf(IOException.class)
+ .hasMessage("flush failure");
+ assertThat(closed).isTrue();
+ }
+
@Test
void testRejectsNonBlobManagedField() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
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
index 333020d29c..7781189c0d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
@@ -27,6 +27,7 @@ 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.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
@@ -35,10 +36,12 @@ 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.postpone.PostponeBucketWriter;
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.BucketMode;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.DataField;
@@ -85,6 +88,96 @@ class PrimaryKeyManagedBlobStoreTest {
assertThat(read.value().getBlob(1).toData()).isEqualTo(expected);
}
+ @Test
+ void testPostponeBucketExternalizesBlobArray() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store =
+ createStore(
+ fileIO,
+ "payloads",
+ DataTypes.ARRAY(DataTypes.BLOB()),
+ BucketMode.POSTPONE_BUCKET);
+ byte[] expected =
"postpone-bucket-blob-array".getBytes(StandardCharsets.UTF_8);
+ KeyValue keyValue =
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ RowKind.INSERT,
+ GenericRow.of(
+ 1,
+ new GenericArray(new Object[]
{Blob.fromData(expected)})));
+
+ store.commitData(
+ Collections.singletonList(keyValue),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> BucketMode.POSTPONE_BUCKET);
+
+ ManifestEntry entry = store.newScan().plan().files().get(0);
+ assertThat(entry.bucket()).isEqualTo(BucketMode.POSTPONE_BUCKET);
+ assertThat(references(fileIO, store, entry)).hasSize(1);
+ KeyValue read =
+
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()).get(0);
+
assertThat(read.value().getArray(1).getBlob(0).toData()).isEqualTo(expected);
+ }
+
+ @Test
+ void testPostponeBucketBufferedRewriteKeepsManagedBlobReferences() throws
Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store =
+ createStore(fileIO, "payload", DataTypes.BLOB(),
BucketMode.POSTPONE_BUCKET);
+ AbstractFileStoreWrite<KeyValue> write = store.newWrite();
+ try (IOManager ioManager =
IOManager.create(tempDir.resolve("io").toString())) {
+ try {
+ write.withIOManager(ioManager);
+ write.write(
+ BinaryRow.EMPTY_ROW,
+ BucketMode.POSTPONE_BUCKET,
+ keyValue(
+ 1,
+ RowKind.INSERT,
+
"before-buffered-rewrite".getBytes(StandardCharsets.UTF_8)));
+ PostponeBucketWriter writer =
+ (PostponeBucketWriter)
+ write.getWriterWrapper(
+ BinaryRow.EMPTY_ROW,
BucketMode.POSTPONE_BUCKET)
+ .writer;
+ writer.toBufferedWriter();
+
+ assertThat(writer.useBufferedSinkWriter()).isTrue();
+ Path bucketPath =
+ store.pathFactory()
+ .bucketPath(BinaryRow.EMPTY_ROW,
BucketMode.POSTPONE_BUCKET);
+ assertThat(fileIO.listStatus(bucketPath))
+ .extracting(status -> status.getPath().getName())
+ .allMatch(
+ name ->
+ name.endsWith(
+
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX));
+
+ write.write(
+ BinaryRow.EMPTY_ROW,
+ BucketMode.POSTPONE_BUCKET,
+ keyValue(
+ 2,
+ RowKind.INSERT,
+
"after-buffered-rewrite".getBytes(StandardCharsets.UTF_8)));
+ List<CommitMessage> messages = write.prepareCommit(false, 1L);
+ try (FileStoreCommit commit = store.newCommit()) {
+ commit.commit(new ManifestCommittable(1L, null, messages),
false);
+ }
+ } finally {
+ write.close();
+ }
+ }
+
+ ManifestEntry entry = store.newScan().plan().files().get(0);
+ assertThat(references(fileIO, store, entry)).hasSize(2);
+
assertThat(store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()))
+ .extracting(
+ kv -> new String(kv.value().getBlob(1).toData(),
StandardCharsets.UTF_8))
+ .containsExactlyInAnyOrder("before-buffered-rewrite",
"after-buffered-rewrite");
+ }
+
@Test
void testExternalizeAndReadBlobArray() throws Exception {
FileIO fileIO = LocalFileIO.create();
@@ -213,6 +306,11 @@ class PrimaryKeyManagedBlobStoreTest {
private TestFileStore createStore(FileIO fileIO, String payloadName,
DataType payloadType)
throws Exception {
+ return createStore(fileIO, payloadName, payloadType, 1);
+ }
+
+ private TestFileStore createStore(
+ FileIO fileIO, String payloadName, DataType payloadType, int
bucket) throws Exception {
Path tablePath = new Path(tempDir.toUri());
List<DataField> valueFields =
Arrays.asList(
@@ -227,7 +325,7 @@ class PrimaryKeyManagedBlobStoreTest {
SpecialFields.KEY_FIELD_PREFIX + "id",
DataTypes.INT())));
Map<String, String> options = new HashMap<>();
- options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.BUCKET.key(), String.valueOf(bucket));
options.put(CoreOptions.BLOB_FIELD.key(), payloadName);
options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b");
TableSchema schema =
@@ -256,7 +354,7 @@ class PrimaryKeyManagedBlobStoreTest {
return new TestFileStore.Builder(
"avro",
tempDir.toString(),
- 1,
+ bucket,
RowType.of(),
keyType,
valueType,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/postpone/PostponeBucketWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/postpone/PostponeBucketWriterTest.java
new file mode 100644
index 0000000000..5b8994f5d6
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/postpone/PostponeBucketWriterTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.postpone;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.KeyValueFileWriterFactory;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.mergetree.compact.MergeFunction;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.types.RowKind;
+
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link PostponeBucketWriter}. */
+class PostponeBucketWriterTest {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testSkipBlobExternalizationWithoutExternalizer() throws Exception {
+ KeyValueFileWriterFactory writerFactory =
mock(KeyValueFileWriterFactory.class);
+ RollingFileWriter<KeyValue, DataFileMeta> rollingWriter =
mock(RollingFileWriter.class);
+ when(writerFactory.hasBlobExternalizer()).thenReturn(false);
+ when(writerFactory.createRollingMergeTreeFileWriter(anyInt(), any()))
+ .thenReturn(rollingWriter);
+ PostponeBucketWriter writer =
+ new PostponeBucketWriter(
+ mock(FileIO.class),
+ mock(DataFilePathFactory.class),
+ CompressOptions.defaultOptions(),
+ new MemorySize(1024),
+ null,
+ mock(MergeFunction.class),
+ writerFactory,
+ null,
+ false,
+ false,
+ null);
+ KeyValue record =
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ RowKind.INSERT,
+ GenericRow.of(1, "ordinary-value"));
+
+ writer.write(record);
+
+ verify(writerFactory).hasBlobExternalizer();
+ verify(writerFactory, never()).externalizeBlob(any(), any());
+ verify(rollingWriter).write(same(record));
+ }
+}
