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 1ef01ddc2c [core] Add coverage for dedicated-format bundle fallback
(#7598)
1ef01ddc2c is described below
commit 1ef01ddc2c7cfe2d80f6af961ea54c57d5f59df3
Author: QuakeWang <[email protected]>
AuthorDate: Tue Jun 23 15:24:49 2026 +0800
[core] Add coverage for dedicated-format bundle fallback (#7598)
---
.../DedicatedFormatRollingFileWriterTest.java | 169 +++++++++++++++++++--
...DedicatedFormatRollingFileWriterVectorTest.java | 156 +++++++++++++++++++
2 files changed, 314 insertions(+), 11 deletions(-)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
index c0bd2814ce..ffd6baedb6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
@@ -43,9 +43,11 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.nio.file.Files;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -146,8 +148,80 @@ public class DedicatedFormatRollingFileWriterTest {
}
@Test
- public void testBundleWriting() throws IOException {
- // Create a bundle of records
+ public void testBundleWritingPreservesMainFileIndexSideEffects() throws
IOException {
+ Options options = new Options();
+ options.set("file-index.bloom-filter.columns", "f0");
+ options.set("file-index.in-manifest-threshold", "1 MB");
+ CoreOptions coreOptions = new CoreOptions(options);
+ writer =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ null,
+ TARGET_FILE_SIZE,
+ TARGET_FILE_SIZE,
+ TARGET_FILE_SIZE,
+ SCHEMA,
+ pathFactory,
+ () -> seqNumCounter,
+ COMPRESSION,
+ new StatsCollectorFactories(coreOptions),
+ new FileIndexOptions(coreOptions),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, coreOptions));
+
+ List<InternalRow> rows =
+ Arrays.asList(
+ GenericRow.of(
+ 1, BinaryString.fromString("test1"), new
BlobData(testBlobData)),
+ GenericRow.of(
+ 2, BinaryString.fromString("test2"), new
BlobData(testBlobData)),
+ GenericRow.of(
+ 3, BinaryString.fromString("test3"), new
BlobData(testBlobData)));
+
+ writer.writeBundle(new SingleUseBundleRecords(rows));
+ writer.close();
+
+ DataFileMeta mainFile =
+ writer.result().stream()
+ .filter(file -> "parquet".equals(file.fileFormat()))
+ .findFirst()
+ .get();
+
+ assertThat(mainFile.rowCount()).isEqualTo(rows.size());
+ assertThat(mainFile.embeddedIndex()).isNotNull();
+ assertThat(mainFile.embeddedIndex()).isNotEmpty();
+ assertThat(mainFile.extraFiles()).isEmpty();
+ }
+
+ @Test
+ public void testBundleWritingWithExternalStorageFallback() throws
IOException {
+ Options options = new Options();
+ options.set(CoreOptions.BLOB_DESCRIPTOR_FIELD, "f2");
+ options.set(CoreOptions.BLOB_EXTERNAL_STORAGE_FIELD, "f2");
+ java.nio.file.Path externalStoragePath =
tempDir.resolve("external-storage-blob-path");
+ options.set(CoreOptions.BLOB_EXTERNAL_STORAGE_PATH,
externalStoragePath.toString());
+ writer =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ null,
+ TARGET_FILE_SIZE,
+ TARGET_FILE_SIZE,
+ TARGET_FILE_SIZE,
+ SCHEMA,
+ pathFactory,
+ () -> seqNumCounter,
+ COMPRESSION,
+ new StatsCollectorFactories(new CoreOptions(options)),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, new
CoreOptions(options)));
+
List<InternalRow> rows =
Arrays.asList(
GenericRow.of(
@@ -157,10 +231,17 @@ public class DedicatedFormatRollingFileWriterTest {
GenericRow.of(
3, BinaryString.fromString("test3"), new
BlobData(testBlobData)));
- // Write bundle
- writer.writeBundle(new TestBundleRecords(rows));
+ writer.writeBundle(new SingleUseBundleRecords(rows));
+ writer.close();
+ List<DataFileMeta> metasResult = writer.result();
assertThat(writer.recordCount()).isEqualTo(3);
+ assertThat(metasResult).hasSize(1);
+ assertThat(metasResult.get(0).rowCount()).isEqualTo(3);
+ assertThat(Files.exists(externalStoragePath)).isTrue();
+ try (Stream<java.nio.file.Path> stream =
Files.list(externalStoragePath)) {
+ assertThat(stream.anyMatch(p ->
p.getFileName().toString().endsWith(".blob"))).isTrue();
+ }
}
@Test
@@ -253,6 +334,64 @@ public class DedicatedFormatRollingFileWriterTest {
assertThat(blobSizeTestWriter.recordCount()).isEqualTo(400);
}
+ @Test
+ public void testBundleWritingRespectsBlobTargetFileSize() throws
IOException {
+ long blobTargetFileSize = 2 * 1024 * 1024L;
+ DedicatedFormatRollingFileWriter bundleWriter =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ null,
+ 128 * 1024 * 1024,
+ blobTargetFileSize,
+ 128 * 1024 * 1024,
+ SCHEMA,
+ new DataFilePathFactory(
+ new Path(tempDir + "/bundle-blob-size-test"),
+ "parquet",
+ "data-",
+ "changelog",
+ false,
+ null,
+ null),
+ () -> new LongCounter(),
+ COMPRESSION,
+ new StatsCollectorFactories(new CoreOptions(new
Options())),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, new CoreOptions(new
Options())));
+
+ byte[] blobData = new byte[1024 * 1024];
+ new Random(321).nextBytes(blobData);
+ List<InternalRow> rows = new java.util.ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ rows.add(
+ GenericRow.of(
+ i,
+ BinaryString.fromString("bundle-blob-test-" + i),
+ new BlobData(blobData)));
+ }
+
+ bundleWriter.writeBundle(new SingleUseBundleRecords(rows));
+ bundleWriter.close();
+
+ List<DataFileMeta> blobFiles =
+ bundleWriter.result().stream()
+ .filter(file -> "blob".equals(file.fileFormat()))
+ .collect(java.util.stream.Collectors.toList());
+ assertThat(blobFiles)
+ .as("Bundle writes should still roll blob files inside the
bundle.")
+ .hasSizeGreaterThan(1);
+ for (DataFileMeta blobFile : blobFiles.subList(0, blobFiles.size() -
1)) {
+ assertThat(blobFile.fileSize())
+ .isGreaterThanOrEqualTo(blobTargetFileSize)
+ .isLessThanOrEqualTo(blobTargetFileSize + blobData.length);
+ }
+ assertThat(bundleWriter.recordCount()).isEqualTo(rows.size());
+ }
+
@Test
public void testSchemaValidation() throws IOException {
// Test that the writer correctly handles the schema with blob field
@@ -494,18 +633,21 @@ public class DedicatedFormatRollingFileWriterTest {
@Test
void testSequenceNumberIncrementInBlobWritePathBatch() throws IOException {
- // Write multiple rows as a batch and verify sequence-number
continuity in blob files
int numRows = 10;
+ List<InternalRow> rows = new java.util.ArrayList<>();
for (int i = 0; i < numRows; i++) {
- InternalRow row =
+ rows.add(
GenericRow.of(
- i, BinaryString.fromString("test" + i), new
BlobData(testBlobData));
- writer.write(row);
+ i, BinaryString.fromString("test" + i), new
BlobData(testBlobData)));
}
+ writer.writeBundle(new SingleUseBundleRecords(rows));
writer.close();
List<DataFileMeta> metasResult = writer.result();
+ assertThat(metasResult).hasSize(2);
+
assertThat(metasResult.get(0).rowCount()).isEqualTo(metasResult.get(1).rowCount());
+
// Extract blob files (skip the first normal file)
List<DataFileMeta> blobFiles =
metasResult.stream()
@@ -616,16 +758,21 @@ public class DedicatedFormatRollingFileWriterTest {
assertThat(writer.recordCount()).isEqualTo(3);
}
- /** Simple implementation of BundleRecords for testing. */
- private static class TestBundleRecords implements BundleRecords {
+ /** Bundle implementation that can only be iterated once. */
+ private static class SingleUseBundleRecords implements BundleRecords {
private final List<InternalRow> rows;
+ private boolean iterated;
- public TestBundleRecords(List<InternalRow> rows) {
+ private SingleUseBundleRecords(List<InternalRow> rows) {
this.rows = rows;
}
@Override
public java.util.Iterator<InternalRow> iterator() {
+ if (iterated) {
+ throw new IllegalStateException("Bundle should only be
consumed once.");
+ }
+ iterated = true;
return rows.iterator();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
index 826d2c0c79..156c79672e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
@@ -25,18 +25,24 @@ import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
+import org.apache.paimon.fileindex.FileIndexPredicate;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.blob.BlobFileFormat;
+import org.apache.paimon.fs.ByteArraySeekableStream;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.BundleRecords;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.operation.BlobFileContext;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.LongCounter;
+import org.apache.paimon.utils.RoaringBitmap32;
import org.apache.paimon.utils.StatsCollectorFactories;
import org.junit.jupiter.api.BeforeEach;
@@ -47,6 +53,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Random;
@@ -172,6 +179,122 @@ public class DedicatedFormatRollingFileWriterVectorTest {
assertThat(writer.recordCount()).isEqualTo(rowNum);
}
+ @Test
+ public void testBundleWriting() throws Exception {
+ List<InternalRow> rows = makeRows(8, 10);
+ writer.writeBundle(new SingleUseBundleRecords(rows));
+ writer.close();
+ List<DataFileMeta> metasResult = writer.result();
+
+ assertThat(writer.recordCount()).isEqualTo(rows.size());
+ assertThat(metasResult).hasSize(3);
+ assertThat(metasResult.get(0).rowCount()).isEqualTo(rows.size());
+ assertThat(metasResult.get(1).rowCount()).isEqualTo(rows.size());
+ assertThat(metasResult.get(2).rowCount()).isEqualTo(rows.size());
+ assertBundleSequenceSideEffects(metasResult, rows.size());
+ }
+
+ @Test
+ public void testBundleWritingRespectsVectorTargetFileSize() throws
Exception {
+ long vectorTargetFileSize = 16 * 1024L;
+ writer =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ FileFormat.fromIdentifier("json", new Options()),
+ 128 * 1024 * 1024,
+ 128 * 1024 * 1024,
+ vectorTargetFileSize,
+ SCHEMA,
+ new DataFilePathFactory(
+ new Path(tempDir + "/bundle-vector-size-test"),
+ "parquet",
+ "data-",
+ "changelog",
+ false,
+ null,
+ null),
+ () -> new LongCounter(),
+ COMPRESSION,
+ new StatsCollectorFactories(new CoreOptions(new
Options())),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, new CoreOptions(new
Options())));
+
+ List<InternalRow> rows = makeRows(2000, 1);
+ writer.writeBundle(new SingleUseBundleRecords(rows));
+ writer.close();
+
+ List<DataFileMeta> vectorStoreFiles =
+ writer.result().stream()
+ .filter(file -> isVectorStoreFile(file.fileName()))
+ .collect(java.util.stream.Collectors.toList());
+ assertThat(vectorStoreFiles)
+ .as("Bundle writes should still roll vector files inside the
bundle.")
+ .hasSizeGreaterThan(1);
+ for (DataFileMeta file : vectorStoreFiles.subList(0,
vectorStoreFiles.size() - 1)) {
+
assertThat(file.fileSize()).isGreaterThanOrEqualTo(vectorTargetFileSize);
+ }
+ assertThat(writer.recordCount()).isEqualTo(rows.size());
+ }
+
+ @Test
+ public void testBundleWritingPreservesMainFileIndexSideEffects() throws
Exception {
+ Options options = new Options();
+ options.set("file-index.bitmap.columns", "f0");
+ options.set("file-index.in-manifest-threshold", "1 MB");
+ CoreOptions coreOptions = new CoreOptions(options);
+ writer =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ FileFormat.fromIdentifier("json", new Options()),
+ TARGET_FILE_SIZE,
+ TARGET_FILE_SIZE,
+ VECTOR_TARGET_FILE_SIZE,
+ SCHEMA,
+ pathFactory,
+ () -> seqNumCounter,
+ COMPRESSION,
+ new StatsCollectorFactories(coreOptions),
+ new FileIndexOptions(coreOptions),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, coreOptions));
+
+ List<InternalRow> rows = makeRows(4, 10);
+ writer.writeBundle(new SingleUseBundleRecords(rows));
+ writer.close();
+
+ DataFileMeta mainFile =
+ writer.result().stream()
+ .filter(file -> "parquet".equals(file.fileFormat()))
+ .findFirst()
+ .get();
+
+ assertThat(mainFile.rowCount()).isEqualTo(rows.size());
+ assertThat(mainFile.embeddedIndex()).isNotEmpty();
+ assertThat(mainFile.extraFiles()).isEmpty();
+
+ RowType normalFileSchema =
+ RowType.builder()
+ .field("f0", DataTypes.INT())
+ .field("f1", DataTypes.STRING())
+ .field("f4", DataTypes.INT())
+ .build();
+ PredicateBuilder predicateBuilder = new
PredicateBuilder(normalFileSchema);
+ try (FileIndexPredicate index =
+ new FileIndexPredicate(
+ new ByteArraySeekableStream(mainFile.embeddedIndex()),
normalFileSchema)) {
+ assertThat(((BitmapIndexResult)
index.evaluate(predicateBuilder.equal(0, 2))).get())
+ .isEqualTo(RoaringBitmap32.bitmapOf(2));
+ assertThat(index.evaluate(predicateBuilder.equal(0,
99)).remain()).isFalse();
+ }
+ }
+
@Test
void testVectorStoreFileNameFormatWithSharedUuid() throws Exception {
// 100k vector-store data would create 1 normal, 1 blob, and 3
vector-store files
@@ -310,4 +433,37 @@ public class DedicatedFormatRollingFileWriterVectorTest {
}
return rows;
}
+
+ private void assertBundleSequenceSideEffects(List<DataFileMeta> metas,
long rowCount) {
+ long expectedCounter = rowCount * metas.size();
+ assertThat(seqNumCounter.getValue()).isEqualTo(expectedCounter);
+ for (DataFileMeta meta : metas) {
+ assertThat(meta.minSequenceNumber()).isEqualTo(expectedCounter -
rowCount);
+ assertThat(meta.maxSequenceNumber()).isEqualTo(expectedCounter -
1);
+ }
+ }
+
+ private static class SingleUseBundleRecords implements BundleRecords {
+
+ private final List<InternalRow> rows;
+ private boolean iterated;
+
+ private SingleUseBundleRecords(List<InternalRow> rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ public Iterator<InternalRow> iterator() {
+ if (iterated) {
+ throw new IllegalStateException("Bundle should only be
consumed once.");
+ }
+ iterated = true;
+ return rows.iterator();
+ }
+
+ @Override
+ public long rowCount() {
+ return rows.size();
+ }
+ }
}
