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 f842658251 [core] Support MAP shared-shredding for primary-key tables
and rewrites (#8709)
f842658251 is described below
commit f8426582514a29bf3eb99547c80f8755f0edcd4d
Author: lxy <[email protected]>
AuthorDate: Sat Jul 18 09:29:43 2026 +0800
[core] Support MAP shared-shredding for primary-key tables and rewrites
(#8709)
---
.../paimon/operation/BaseAppendFileStoreWrite.java | 10 -
.../org/apache/paimon/schema/SchemaValidation.java | 37 +-
.../apache/paimon/append/AppendOnlyWriterTest.java | 71 ++-
.../BucketedAppendFileStoreWriteTest.java | 105 ++++
.../apache/paimon/schema/SchemaValidationTest.java | 49 +-
.../paimon/table/MapSharedShreddingTableTest.java | 610 +++++++++++++++++++++
6 files changed, 834 insertions(+), 48 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 901fbe2fe0..7dceb12e5c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -26,7 +26,6 @@ import org.apache.paimon.compact.CompactManager;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.fileindex.FileIndexOptions;
@@ -228,7 +227,6 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
if (toCompact.isEmpty()) {
return Collections.emptyList();
}
- checkNoSharedShreddingRewrite("Compaction rewrite");
Exception collectedExceptions = null;
RowDataRollingFileWriter rewriter =
createRollingFileWriter(
@@ -261,7 +259,6 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
public List<DataFileMeta> clusterRewrite(
BinaryRow partition, int bucket, List<DataFileMeta> toCluster)
throws Exception {
- checkNoSharedShreddingRewrite("Cluster rewrite");
RecordReaderIterator<InternalRow> reader =
createFilesIterator(partition, bucket, toCluster, null);
@@ -298,13 +295,6 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
return rewriter.result();
}
- private void checkNoSharedShreddingRewrite(String rewriteName) {
- if (!MapSharedShreddingUtils.detectShreddingColumns(writeType,
options).isEmpty()) {
- throw new UnsupportedOperationException(
- rewriteName + " is not supported for MAP
shared-shredding.");
- }
- }
-
private RowDataRollingFileWriter createRollingFileWriter(
BinaryRow partition, int bucket, Supplier<LongCounter>
seqNumCounterSupplier) {
return new RowDataRollingFileWriter(
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 4b81ba6ae4..d145704fef 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
@@ -682,35 +682,40 @@ public class SchemaValidation {
if (hasSharedShredding) {
validateMapSharedShreddingFileFormats(options);
validateMapSharedShreddingCompressions(options);
- validateUnsupportedTypesWithMapSharedShredding(schema);
- if (!schema.primaryKeys().isEmpty()) {
+ validateUnsupportedTypesWithMapSharedShredding(schema, options);
+ if (options.bucket() == BucketMode.POSTPONE_BUCKET) {
throw new IllegalArgumentException(
- "MAP shared-shredding currently only supports
append-only tables.");
- }
- if (options.bucket() != -1 && !options.writeOnly()) {
- throw new IllegalArgumentException(
- "MAP shared-shredding currently requires bucket = -1
or write-only = true because rewrite/compaction is not supported.");
+ "MAP shared-shredding currently does not support
postpone bucket mode.");
}
}
}
- private static void
validateUnsupportedTypesWithMapSharedShredding(TableSchema schema) {
+ private static void validateUnsupportedTypesWithMapSharedShredding(
+ TableSchema schema, CoreOptions options) {
RowType rowType = new RowType(schema.fields());
if (containsType(rowType, type -> type instanceof VariantType)) {
throw new IllegalArgumentException(
"MAP shared-shredding currently cannot be used with
Variant fields.");
}
- if (containsType(rowType, type -> type.is(DataTypeRoot.BLOB))) {
- throw new IllegalArgumentException(
- "MAP shared-shredding currently cannot be used with BLOB
fields.");
- }
if (containsType(rowType, type -> type instanceof MultisetType)) {
throw new IllegalArgumentException(
"MAP shared-shredding currently cannot be used with
MULTISET fields.");
}
- if (containsType(rowType, type -> type instanceof VectorType)) {
- throw new IllegalArgumentException(
- "MAP shared-shredding currently cannot be used with VECTOR
fields.");
+
+ for (DataField field : schema.fields()) {
+ if (options.mapStorageLayout(field.name()) !=
MapStorageLayout.SHARED_SHREDDING) {
+ continue;
+ }
+
+ DataType valueType = ((MapType) field.type()).getValueType();
+ if (containsType(valueType, type -> type.is(DataTypeRoot.BLOB))) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot contain BLOB
fields.");
+ }
+ if (containsType(valueType, type -> type instanceof VectorType)) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot contain VECTOR
fields.");
+ }
}
}
@@ -748,8 +753,6 @@ public class SchemaValidation {
}
validateMapSharedShreddingFileFormat(
CoreOptions.CHANGELOG_FILE_FORMAT.key(),
options.changelogFileFormat());
- validateMapSharedShreddingFileFormat(
- CoreOptions.VECTOR_FILE_FORMAT.key(),
options.vectorFileFormatString());
}
private static void validateMapSharedShreddingCompressions(CoreOptions
options) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
index 8354945364..47ebf0fbaa 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
@@ -76,6 +76,8 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
@@ -617,6 +619,62 @@ public class AppendOnlyWriterTest {
3));
}
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET,
CoreOptions.FILE_FORMAT_ORC})
+ public void testSharedShreddingMapWithBlob(String fileFormat) throws
Exception {
+ RowType writeType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1,
+ "tags",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT())),
+ DataTypes.FIELD(2, "payload", new BlobType()));
+ Options rawOptions = sharedShreddingOptions("tags", 3);
+ SharedShreddingAppendContext context =
+ createSharedShreddingAppendContext(fileFormat, rawOptions);
+ BlobFileContext blobContext =
+ BlobFileContext.create(writeType, new CoreOptions(rawOptions));
+ assertThat(blobContext).isNotNull();
+
+ AppendOnlyWriter writer =
+ createSharedShreddingAppendWriter(
+ writeType, context, SCHEMA_ID, -1L, new
FileIndexOptions(), blobContext);
+
+ writer.write(
+ sharedShreddingLogicalRow(1, map("a", 10L), new BlobData(new
byte[] {1, 2, 3})));
+ CommitIncrement increment = writer.prepareCommit(true);
+ writer.close();
+
+ assertThat(increment.newFilesIncrement().newFiles()).hasSize(2);
+ DataFileMeta mainFile =
+ increment.newFilesIncrement().newFiles().stream()
+ .filter(file -> fileFormat.equals(file.fileFormat()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ assertThat(readSharedShreddingFieldMeta(context, mainFile, "tags"))
+ .isEqualTo(
+ sharedShreddingMeta(
+ nameToId("a", 0),
+ fieldToColumns(0, columns(0)),
+ overflowFields(),
+ 1,
+ 1));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET,
CoreOptions.FILE_FORMAT_ORC})
+ public void testSharedShreddingMapAllowsForceBufferSpill(String
fileFormat) throws Exception {
+ RowType writeType = sharedShreddingTagsWriteType();
+ Options rawOptions = sharedShreddingOptions("tags", 3);
+ SharedShreddingAppendContext context =
+ createSharedShreddingAppendContext(fileFormat, rawOptions);
+ AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType,
context);
+
+
Assertions.assertThatCode(writer::toBufferedWriter).doesNotThrowAnyException();
+ writer.close();
+ }
+
@Test
public void testNoBuffer() throws Exception {
AppendOnlyWriter writer = createEmptyWriter(Long.MAX_VALUE);
@@ -947,6 +1005,17 @@ public class AppendOnlyWriterTest {
long schemaId,
long maxSequenceNumber,
FileIndexOptions fileIndexOptions) {
+ return createSharedShreddingAppendWriter(
+ writeType, context, schemaId, maxSequenceNumber,
fileIndexOptions, null);
+ }
+
+ private AppendOnlyWriter createSharedShreddingAppendWriter(
+ RowType writeType,
+ SharedShreddingAppendContext context,
+ long schemaId,
+ long maxSequenceNumber,
+ FileIndexOptions fileIndexOptions,
+ @Nullable BlobFileContext blobContext) {
return new AppendOnlyWriter(
context.fileIO,
null,
@@ -975,7 +1044,7 @@ public class AppendOnlyWriterTest {
false,
context.options.dataEvolutionEnabled(),
null,
- null);
+ blobContext);
}
private DataFileMeta writeSharedShreddingFile(AppendOnlyWriter writer,
InternalRow... rows)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
index 556209b2d3..2d0e322910 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/BucketedAppendFileStoreWriteTest.java
@@ -24,28 +24,38 @@ import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.ExternalBuffer;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.RowBuffer;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
@@ -117,6 +127,52 @@ public class BucketedAppendFileStoreWriteTest {
Assertions.assertThat(records).isEqualTo(11);
}
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {"parquet", "orc"})
+ public void testSharedShreddingForceBufferSpill(String fileFormat) throws
Exception {
+ FileStoreTable table =
createSharedShreddingForceBufferSpillTable(fileFormat);
+ BucketedAppendFileStoreWrite write =
+ (BucketedAppendFileStoreWrite) table.store().newWrite("ss");
+ write.withIOManager(IOManager.create(tempDir.toString()));
+
+ write.write(partition(0), 0, GenericRow.of(0, 0, map("a", 10L)));
+ Assertions.assertThat(writeBuffers(write)).containsExactly((RowBuffer)
null);
+
+ // Creating the second writer reaches the configured limit and
triggers force spill.
+ write.write(partition(1), 1, GenericRow.of(1, 1, map("b", 20L)));
+ Assertions.assertThat(writeBuffers(write))
+ .hasSize(2)
+ .allSatisfy(
+ buffer ->
Assertions.assertThat(buffer).isInstanceOf(ExternalBuffer.class));
+
+ write.write(partition(0), 0, GenericRow.of(0, 2, map("c", 30L)));
+ write.write(partition(1), 1, GenericRow.of(1, 3, map("d", 40L)));
+
+ List<CommitMessage> messages = write.prepareCommit(true,
Long.MAX_VALUE);
+ try (StreamTableCommit commit =
table.newStreamWriteBuilder().newCommit()) {
+ commit.commit(0, messages);
+ }
+
+ List<List<Object>> actual = new ArrayList<>();
+ ReadBuilder readBuilder = table.newReadBuilder();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ actual.add(
+ Arrays.asList(
+ row.getInt(0),
+ row.getInt(1),
+ toJavaMap(row.getMap(2)))));
+ }
+ Assertions.assertThat(actual)
+ .containsExactlyInAnyOrder(
+ Arrays.asList(0, 0, Collections.singletonMap("a",
10L)),
+ Arrays.asList(1, 1, Collections.singletonMap("b",
20L)),
+ Arrays.asList(0, 2, Collections.singletonMap("c",
30L)),
+ Arrays.asList(1, 3, Collections.singletonMap("d",
40L)));
+ }
+
@Test
public void testWritesInBatchWithNoExtraFiles() throws Exception {
FileStoreTable table = createFileStoreTable();
@@ -176,6 +232,55 @@ public class BucketedAppendFileStoreWriteTest {
return (FileStoreTable) catalog.getTable(identifier);
}
+ private FileStoreTable createSharedShreddingForceBufferSpillTable(String
fileFormat)
+ throws Exception {
+ Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new
Path(tempDir.toString()));
+ Schema schema =
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))
+ .partitionKeys("f0")
+ .option(BUCKET.key(), "100")
+ .option("bucket-key", "f1")
+ .option(WRITE_ONLY.key(), "true")
+ .option("file.format", fileFormat)
+ .option(WRITE_MAX_WRITERS_TO_SPILL.key(), "1")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "2")
+ .build();
+ Identifier identifier = Identifier.create("default", "spill_" +
fileFormat);
+ catalog.createDatabase("default", true);
+ catalog.createTable(identifier, schema, false);
+ return (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ private GenericMap map(String key, long value) {
+ return new
GenericMap(Collections.singletonMap(BinaryString.fromString(key), value));
+ }
+
+ private List<RowBuffer> writeBuffers(BucketedAppendFileStoreWrite write) {
+ List<RowBuffer> buffers = new ArrayList<>();
+ for (Map<Integer, AbstractFileStoreWrite.WriterContainer<InternalRow>>
bucketWriters :
+ write.writers().values()) {
+ for (AbstractFileStoreWrite.WriterContainer<InternalRow>
writerContainer :
+ bucketWriters.values()) {
+ buffers.add(((AppendOnlyWriter)
writerContainer.writer).getWriteBuffer());
+ }
+ }
+ return buffers;
+ }
+
+ private Map<String, Long> toJavaMap(InternalMap map) {
+ Map<String, Long> result = new LinkedHashMap<>();
+ for (int i = 0; i < map.size(); i++) {
+ result.put(map.keyArray().getString(i).toString(),
map.valueArray().getLong(i));
+ }
+ return result;
+ }
+
@Test
public void testIgnorePreviousFilesChecksPartitionBucketNumber() throws
Exception {
FileStoreTable table = createFileStoreTable().copy(bucketOptions(2,
false, false));
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 12608d1720..ec07c17c4f 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
@@ -19,6 +19,7 @@
package org.apache.paimon.schema;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -526,15 +527,11 @@ class SchemaValidationTest {
assertMapSharedShreddingValidationFailed(
mapValueFields(DataTypes.BLOB()),
mapSharedShreddingOptions(),
- "MAP shared-shredding currently cannot be used with BLOB
fields.");
+ "MAP shared-shredding currently cannot contain BLOB fields.");
assertMapSharedShreddingValidationFailed(
nestedMapValueFields(DataTypes.BLOB()),
mapSharedShreddingOptions(),
- "MAP shared-shredding currently cannot be used with BLOB
fields.");
- assertMapSharedShreddingValidationFailed(
- topLevelPayloadFields(DataTypes.BLOB()),
- mapSharedShreddingOptions(),
- "MAP shared-shredding currently cannot be used with BLOB
fields.");
+ "MAP shared-shredding currently cannot contain BLOB fields.");
DataType multisetType = DataTypes.MULTISET(DataTypes.INT());
assertMapSharedShreddingValidationFailed(
@@ -554,11 +551,7 @@ class SchemaValidationTest {
assertMapSharedShreddingValidationFailed(
mapValueFields(vectorType),
mapSharedShreddingOptions(),
- "MAP shared-shredding currently cannot be used with VECTOR
fields.");
- assertMapSharedShreddingValidationFailed(
- topLevelPayloadFields(vectorType),
- mapSharedShreddingOptions(),
- "MAP shared-shredding currently cannot be used with VECTOR
fields.");
+ "MAP shared-shredding currently cannot contain VECTOR
fields.");
}
@Test
@@ -595,12 +588,11 @@ class SchemaValidationTest {
vectorFormatOptions.put(DATA_EVOLUTION_ENABLED.key(), "true");
vectorFormatOptions.put(CoreOptions.ROW_TRACKING_ENABLED.key(),
"true");
vectorFormatOptions.put(CoreOptions.VECTOR_FILE_FORMAT.key(), "json");
- assertThatThrownBy(
+ assertThatCode(
() ->
validateTableSchema(
mapSharedShreddingSchema(vectorFormatOptions, emptyList())))
- .hasMessageContaining(
- "MAP shared-shredding only supports parquet/orc file
formats, but vector.file.format is json.");
+ .doesNotThrowAnyException();
}
@Test
@@ -651,23 +643,40 @@ class SchemaValidationTest {
public void testMapSharedShreddingTableModeValidation() {
Map<String, String> primaryKeyOptions = mapSharedShreddingOptions();
primaryKeyOptions.put(BUCKET.key(), "-1");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ primaryKeyOptions,
singletonList("id"))));
+
+ primaryKeyOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ primaryKeyOptions,
singletonList("id"))));
+
+ Map<String, String> postponeBucketOptions =
mapSharedShreddingOptions();
+ postponeBucketOptions.put(BUCKET.key(),
String.valueOf(BucketMode.POSTPONE_BUCKET));
+ postponeBucketOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
assertThatThrownBy(
() ->
validateTableSchema(
mapSharedShreddingSchema(
- primaryKeyOptions,
singletonList("id"))))
+ postponeBucketOptions,
singletonList("id"))))
.hasMessageContaining(
- "MAP shared-shredding currently only supports
append-only tables.");
+ "MAP shared-shredding currently does not support
postpone bucket mode.");
Map<String, String> fixedBucketOptions = mapSharedShreddingOptions();
fixedBucketOptions.put(BUCKET.key(), "1");
fixedBucketOptions.put(CoreOptions.BUCKET_KEY.key(), "id");
- assertThatThrownBy(
+ assertThatNoException()
+ .isThrownBy(
() ->
validateTableSchema(
-
mapSharedShreddingSchema(fixedBucketOptions, emptyList())))
- .hasMessageContaining(
- "MAP shared-shredding currently requires bucket = -1
or write-only = true because rewrite/compaction is not supported.");
+
mapSharedShreddingSchema(fixedBucketOptions, emptyList())));
Map<String, String> writeOnlyOptions = mapSharedShreddingOptions();
writeOnlyOptions.put(BUCKET.key(), "1");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
index 66e166c7da..de5b74572b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
@@ -21,6 +21,8 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BinaryVector;
+import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
@@ -35,8 +37,11 @@ import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.SupportsFieldMetadata;
+import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -46,14 +51,19 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.math.BigDecimal;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -61,7 +71,9 @@ import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Stream;
+import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static org.assertj.core.api.Assertions.assertThat;
/** Table-level tests for MAP shared-shredding. */
@@ -134,6 +146,37 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
.containsEntry(3, javaMapOf("x", 41L));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testAppendOnlyCompaction(String format) throws Exception {
+ Table table = createCompactingAppendOnlyTable(format);
+
+ write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L)));
+ write(table, GenericRow.of(2, mapOf("c", 21L)));
+ write(table, GenericRow.of(3, mapOf("d", 31L, "e", 32L, "f", 33L, "g",
34L)));
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> files = currentDataFiles(fileStoreTable);
+ assertThat(files).hasSize(1);
+
assertThat(files.get(0).dataFile.fileSource()).hasValue(FileSource.COMPACT);
+ MapSharedShreddingFieldMeta compactedMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(0),
"metrics");
+ assertThat(compactedMeta.numColumns()).isEqualTo(2);
+ assertThat(compactedMeta.maxRowWidth()).isEqualTo(4);
+ assertThat(compactedMeta.nameToId()).containsOnlyKeys("a", "b", "c",
"d", "e", "f", "g");
+
+ Map<Integer, Map<String, Long>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(row.getInt(0), toJavaMap(row.getMap(1)));
+ }
+ assertThat(actual)
+ .containsOnlyKeys(1, 2, 3)
+ .containsEntry(1, javaMapOf("a", 11L, "b", 12L))
+ .containsEntry(2, javaMapOf("c", 21L))
+ .containsEntry(3, javaMapOf("d", 31L, "e", 32L, "f", 33L, "g",
34L));
+ }
+
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
public void testAppendOnlyTableReadWriteWithComplexValue(String format)
throws Exception {
@@ -317,6 +360,311 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
.containsEntry(3, javaMapOf("f", 31L));
}
+ @ParameterizedTest
+ @CsvSource({"orc,false", "orc,true", "parquet,false", "parquet,true"})
+ public void testPrimaryKeyWriteOnlyReadWrite(String format, boolean
thinMode) throws Exception {
+ Table table = createPrimaryKeyTable(format, thinMode, 2);
+
+ write(
+ table,
+ GenericRow.of(1, mapOf("a", 11L, "b", 12L, "c", 13L)),
+ GenericRow.of(2, null),
+ GenericRow.of(3, mapOf()));
+ write(
+ table,
+ GenericRow.of(1, mapOf("x", null, "y", 42L, "z", null)),
+ GenericRow.of(4, mapOf("d", 44L)));
+
+ Map<Integer, Map<String, Long>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(row.getInt(0), row.isNullAt(1) ? null :
toJavaMap(row.getMap(1)));
+ }
+
+ assertThat(actual)
+ .containsOnlyKeys(1, 2, 3, 4)
+ .containsEntry(1, javaMapOf("x", null, "y", 42L, "z", null))
+ .containsEntry(2, null)
+ .containsEntry(3, javaMapOf())
+ .containsEntry(4, javaMapOf("d", 44L));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"orc,parquet", "parquet,orc"})
+ public void testPrimaryKeyInputChangelog(String dataFileFormat, String
changelogFileFormat)
+ throws Exception {
+ Table table = createPrimaryKeyInputChangelogTable(dataFileFormat,
changelogFileFormat);
+
+ List<CommitMessage> messages =
+ writeAndCommit(
+ table,
+ GenericRow.ofKind(RowKind.INSERT, 1, mapOf("a", 11L,
"b", 12L)),
+ GenericRow.ofKind(RowKind.INSERT, 2, mapOf("c", 21L)),
+ GenericRow.ofKind(RowKind.UPDATE_BEFORE, 1, mapOf("a",
11L, "b", 12L)),
+ GenericRow.ofKind(
+ RowKind.UPDATE_AFTER,
+ 1,
+ mapOf("x", 31L, "y", 32L, "overflow", 33L)),
+ GenericRow.ofKind(RowKind.DELETE, 2, mapOf("c", 21L)));
+
+ List<DataFileWithSplit> changelogFiles = new ArrayList<>();
+ for (CommitMessage message : messages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ for (DataFileMeta file :
commitMessage.newFilesIncrement().changelogFiles()) {
+ changelogFiles.add(
+ new DataFileWithSplit(
+ commitMessage.partition(),
commitMessage.bucket(), file));
+ }
+ }
+ assertThat(changelogFiles).hasSize(1);
+ DataFileWithSplit changelogFile = changelogFiles.get(0);
+
assertThat(changelogFile.dataFile.fileFormat()).isEqualTo(changelogFileFormat);
+ assertThat(changelogFile.dataFile.fileName())
+ .startsWith(((FileStoreTable)
table).coreOptions().changelogFilePrefix());
+
+ MapSharedShreddingFieldMeta changelogMeta =
+ readSharedShreddingFieldMeta((FileStoreTable) table,
changelogFile, "metrics");
+ assertThat(changelogMeta.numColumns()).isEqualTo(2);
+ assertThat(changelogMeta.maxRowWidth()).isEqualTo(3);
+ assertThat(changelogMeta.nameToId()).containsOnlyKeys("a", "b", "c",
"x", "y", "overflow");
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<Split> changelogSplits =
+
fileStoreTable.newSnapshotReader().withMode(ScanMode.CHANGELOG).read().splits();
+ Map<String, Map<String, Long>> actualChangelog = new LinkedHashMap<>();
+ try (RecordReader<InternalRow> reader =
+ fileStoreTable.newRead().createReader(changelogSplits)) {
+ reader.forEachRemaining(
+ row ->
+ actualChangelog.put(
+ row.getRowKind().shortString() + ":" +
row.getInt(0),
+ toJavaMap(row.getMap(1))));
+ }
+ assertThat(actualChangelog)
+ .containsEntry("+I:1", javaMapOf("a", 11L, "b", 12L))
+ .containsEntry("+I:2", javaMapOf("c", 21L))
+ .containsEntry("-U:1", javaMapOf("a", 11L, "b", 12L))
+ .containsEntry("+U:1", javaMapOf("x", 31L, "y", 32L,
"overflow", 33L))
+ .containsEntry("-D:2", javaMapOf("c", 21L));
+
+ assertThat(readMapsById(table.newReadBuilder()))
+ .containsOnlyKeys(1)
+ .containsEntry(1, javaMapOf("x", 31L, "y", 32L, "overflow",
33L));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"orc,false", "orc,true", "parquet,false", "parquet,true"})
+ public void testPrimaryKeyMergeMapAggregation(String format, boolean
thinMode)
+ throws Exception {
+ Table table = createPrimaryKeyAggregationTable(format, thinMode);
+
+ write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L)),
GenericRow.of(2, null));
+ write(
+ table,
+ GenericRow.of(1, mapOf("b", 22L, "c", 23L)),
+ GenericRow.of(2, mapOf("x", 31L)));
+
+ assertThat(readMapsById(table.newReadBuilder()))
+ .containsOnlyKeys(1, 2)
+ .containsEntry(1, javaMapOf("a", 11L, "b", 22L, "c", 23L))
+ .containsEntry(2, javaMapOf("x", 31L));
+
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> files = currentDataFiles(fileStoreTable);
+ assertThat(files).hasSize(1);
+
assertThat(files.get(0).dataFile.fileSource()).hasValue(FileSource.COMPACT);
+ assertThat(readMapsById(table.newReadBuilder()))
+ .containsOnlyKeys(1, 2)
+ .containsEntry(1, javaMapOf("a", 11L, "b", 22L, "c", 23L))
+ .containsEntry(2, javaMapOf("x", 31L));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"orc,false", "orc,true", "parquet,false", "parquet,true"})
+ public void testPrimaryKeyExternalSpillReadWrite(String format, boolean
thinMode)
+ throws Exception {
+ Table table = createPrimaryKeyTable(format, thinMode, 2);
+ Map<String, String> spillOptions = new LinkedHashMap<>();
+ spillOptions.put(CoreOptions.PAGE_SIZE.key(), "4 kb");
+ spillOptions.put(CoreOptions.WRITE_BUFFER_SIZE.key(), "12 kb");
+ spillOptions.put(CoreOptions.WRITE_BUFFER_SPILLABLE.key(), "true");
+ table = table.copy(spillOptions);
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.withIOManager(ioManager);
+ for (int i = 0; i < 1000; i++) {
+ write.write(
+ GenericRow.of(
+ i % 100,
+ mapOf("a", (long) i, "b", (long) i + 1,
"overflow", (long) -i)));
+ }
+ assertThat(numberOfSpillFiles()).isGreaterThan(0);
+ commit.commit(write.prepareCommit());
+ }
+
+ Map<Integer, Map<String, Long>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(row.getInt(0), toJavaMap(row.getMap(1)));
+ }
+
+ assertThat(actual).hasSize(100);
+ for (int id = 0; id < 100; id++) {
+ long lastValue = 900L + id;
+ assertThat(actual)
+ .containsEntry(
+ id,
+ javaMapOf("a", lastValue, "b", lastValue + 1,
"overflow", -lastValue));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"orc,false", "orc,true", "parquet,false", "parquet,true"})
+ public void testPrimaryKeyCompaction(String format, boolean thinMode)
throws Exception {
+ Table table = createPrimaryKeyTable(format, thinMode, 2, false);
+
+ write(
+ table,
+ GenericRow.of(1, mapOf("old", 11L)),
+ GenericRow.of(2, mapOf("a", 21L, "b", 22L, "overflow", 23L)));
+ write(
+ table,
+ GenericRow.of(1, mapOf("x", 31L, "y", 32L, "overflow", 33L)),
+ GenericRow.of(3, null));
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> files = currentDataFiles(fileStoreTable);
+ assertThat(files).hasSize(1);
+
assertThat(files.get(0).dataFile.fileSource()).hasValue(FileSource.COMPACT);
+ MapSharedShreddingFieldMeta compactedMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(0),
"metrics");
+ assertThat(compactedMeta.numColumns()).isEqualTo(2);
+ assertThat(compactedMeta.maxRowWidth()).isEqualTo(3);
+ assertThat(compactedMeta.nameToId()).containsOnlyKeys("a", "b",
"overflow", "x", "y");
+
+ Map<Integer, Map<String, Long>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(row.getInt(0), row.isNullAt(1) ? null :
toJavaMap(row.getMap(1)));
+ }
+ assertThat(actual)
+ .containsOnlyKeys(1, 2, 3)
+ .containsEntry(1, javaMapOf("x", 31L, "y", 32L, "overflow",
33L))
+ .containsEntry(2, javaMapOf("a", 21L, "b", 22L, "overflow",
23L))
+ .containsEntry(3, null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testPrimaryKeyDeletionVectorCompaction(String format) throws
Exception {
+ Table table = createPrimaryKeyTable(format, false, 2, false, true);
+ String padding = String.join("", Collections.nCopies(2048, "X"));
+
+ write(
+ table,
+ ioManager,
+ GenericRow.of(1, mapOf("a", 10L, "b", 20L),
BinaryString.fromString(padding)),
+ GenericRow.of(2, mapOf("c", 30L),
BinaryString.fromString(padding)),
+ GenericRow.of(3, mapOf("d", 40L),
BinaryString.fromString(padding)),
+ GenericRow.of(4, null, BinaryString.fromString(padding)),
+ GenericRow.of(6, mapOf("j", 60L, "k", 70L),
BinaryString.fromString(padding)),
+ GenericRow.of(
+ 7, mapOf("l", 80L, "m", 90L, "n", 100L),
BinaryString.fromString(padding)),
+ GenericRow.of(8, mapOf("o", 110L),
BinaryString.fromString(padding)));
+ List<CommitMessage> upgradeMessages = compactAndCommit(table,
BinaryRow.EMPTY_ROW, 0, true);
+ assertThat(upgradeMessages)
+ .allSatisfy(
+ message ->
+ assertThat(
+ ((CommitMessageImpl) message)
+ .compactIncrement()
+ .newIndexFiles())
+ .isEmpty());
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> upgradedFiles =
currentDataFiles(fileStoreTable);
+ assertThat(upgradedFiles).hasSize(1);
+ assertThat(upgradedFiles.get(0).dataFile.fileSize())
+
.isGreaterThan(fileStoreTable.coreOptions().compactionFileSize(true));
+
+ List<CommitMessage> compactMessages = new ArrayList<>();
+ compactMessages.addAll(
+ writeAndCommit(
+ table,
+ GenericRow.of(
+ 1, mapOf("a", 100L, "e", 500L),
BinaryString.fromString("u1")),
+ GenericRow.of(5, mapOf("h", 80L),
BinaryString.fromString("u5"))));
+ compactMessages.addAll(
+ writeAndCommit(
+ table,
+ GenericRow.of(
+ 2,
+ mapOf("c", 300L, "f", 600L, "g", 700L),
+ BinaryString.fromString("u2")),
+ GenericRow.of(
+ 5,
+ mapOf("h", 800L, "i", 900L),
+ BinaryString.fromString("u5-new"))));
+ compactMessages.addAll(compactAndCommit(table, BinaryRow.EMPTY_ROW, 0,
false));
+ List<IndexFileMeta> deletionVectorFiles = new ArrayList<>();
+ for (CommitMessage message : compactMessages) {
+ deletionVectorFiles.addAll(
+ ((CommitMessageImpl)
message).compactIncrement().newIndexFiles());
+ }
+ assertThat(deletionVectorFiles)
+ .isNotEmpty()
+ .allSatisfy(file ->
assertThat(file.indexType()).isEqualTo(DELETION_VECTORS_INDEX));
+ assertThat(deletionVectorFiles)
+ .anySatisfy(file -> assertThat(file.dvRanges()).isNotEmpty());
+
+ Map<Integer, Map<String, Long>> actual =
readMapsById(table.newReadBuilder());
+ assertThat(actual)
+ .containsOnlyKeys(1, 2, 3, 4, 5, 6, 7, 8)
+ .containsEntry(1, javaMapOf("a", 100L, "e", 500L))
+ .containsEntry(2, javaMapOf("c", 300L, "f", 600L, "g", 700L))
+ .containsEntry(3, javaMapOf("d", 40L))
+ .containsEntry(4, null)
+ .containsEntry(5, javaMapOf("h", 800L, "i", 900L))
+ .containsEntry(6, javaMapOf("j", 60L, "k", 70L))
+ .containsEntry(7, javaMapOf("l", 80L, "m", 90L, "n", 100L))
+ .containsEntry(8, javaMapOf("o", 110L));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"orc,false", "orc,true", "parquet,false", "parquet,true"})
+ public void testPrimaryKeyInfersColumnCountPerFile(String format, boolean
thinMode)
+ throws Exception {
+ Table table = createPrimaryKeyTable(format, thinMode, 8);
+
+ write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L)));
+ write(table, GenericRow.of(2, mapOf("c", 22L)));
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> files = currentDataFiles(fileStoreTable);
+ files.sort(Comparator.comparingLong(file ->
file.dataFile.minSequenceNumber()));
+ assertThat(files).hasSize(2);
+
+ MapSharedShreddingFieldMeta firstFileMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(0),
"metrics");
+ assertThat(firstFileMeta.numColumns()).isEqualTo(2);
+ assertThat(firstFileMeta.maxRowWidth()).isEqualTo(2);
+
+ MapSharedShreddingFieldMeta secondFileMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(1),
"metrics");
+ assertThat(secondFileMeta.numColumns()).isEqualTo(1);
+ assertThat(secondFileMeta.maxRowWidth()).isEqualTo(1);
+
+ Map<Integer, Map<String, Long>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(row.getInt(0), toJavaMap(row.getMap(1)));
+ }
+ assertThat(actual)
+ .containsEntry(1, javaMapOf("a", 11L, "b", 12L))
+ .containsEntry(2, javaMapOf("c", 22L));
+ }
+
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
public void testSwitchMapLayoutAndInferColumns(String format) throws
Exception {
@@ -462,6 +810,87 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
102, Arrays.asList(javaMapOf("new-b", 222L),
javaMapOf("label-b", 102L)));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testSharedShreddingWithBlobAndVector(String format) throws
Exception {
+ Table table = createBlobAndVectorTable(format);
+
+ write(
+ table,
+ GenericRow.of(
+ 1,
+ mapOf("a", 11L, "b", 12L, "overflow", 13L),
+ new BlobData(new byte[] {1, 2, 3}),
+ BinaryVector.fromPrimitiveArray(new float[] {1.0f,
2.0f, 3.0f})),
+ GenericRow.of(2, null, null, null),
+ GenericRow.of(
+ 3,
+ mapOf(),
+ new BlobData(new byte[] {4, 5}),
+ BinaryVector.fromPrimitiveArray(new float[] {4.0f,
5.0f, 6.0f})));
+
+ Map<Integer, Map<String, Long>> initialMaps = new LinkedHashMap<>();
+ for (InternalRow row : read(table, new int[] {0, 1})) {
+ initialMaps.put(row.getInt(0), row.isNullAt(1) ? null :
toJavaMap(row.getMap(1)));
+ }
+ assertThat(initialMaps)
+ .containsEntry(1, javaMapOf("a", 11L, "b", 12L, "overflow",
13L))
+ .containsEntry(2, null)
+ .containsEntry(3, javaMapOf());
+
+ RowType rowType = table.rowType();
+ writeWithWriteType(
+ table,
+ rowType.project(Collections.singletonList("metrics")),
+ 0L,
+ GenericRow.of(mapOf("updated", 101L)),
+ GenericRow.of(mapOf("updated", 102L)),
+ GenericRow.of(mapOf("updated", 103L)));
+
+ Map<Integer, Map<String, Long>> maps = new LinkedHashMap<>();
+ Map<Integer, byte[]> blobs = new LinkedHashMap<>();
+ Map<Integer, float[]> vectors = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ int id = row.getInt(0);
+ maps.put(id, toJavaMap(row.getMap(1)));
+ blobs.put(id, row.isNullAt(2) ? null : row.getBlob(2).toData());
+ vectors.put(id, row.isNullAt(3) ? null :
row.getVector(3).toFloatArray());
+ }
+
+ assertThat(maps)
+ .containsEntry(1, javaMapOf("updated", 101L))
+ .containsEntry(2, javaMapOf("updated", 102L))
+ .containsEntry(3, javaMapOf("updated", 103L));
+ assertThat(blobs.get(1)).containsExactly(1, 2, 3);
+ assertThat(blobs.get(2)).isNull();
+ assertThat(blobs.get(3)).containsExactly(4, 5);
+ assertThat(vectors.get(1)).containsExactly(1.0f, 2.0f, 3.0f);
+ assertThat(vectors.get(2)).isNull();
+ assertThat(vectors.get(3)).containsExactly(4.0f, 5.0f, 6.0f);
+
+ Map<Integer, Map<String, Long>> projectedMaps = new LinkedHashMap<>();
+ for (InternalRow row : read(table, new int[] {0, 1})) {
+ projectedMaps.put(row.getInt(0), toJavaMap(row.getMap(1)));
+ }
+ assertThat(projectedMaps).isEqualTo(maps);
+
+ for (InternalRow row : read(table, new int[] {0, 2, 3})) {
+ int id = row.getInt(0);
+ byte[] expectedBlob = blobs.get(id);
+ if (expectedBlob == null) {
+ assertThat(row.isNullAt(1)).isTrue();
+ } else {
+
assertThat(row.getBlob(1).toData()).containsExactly(expectedBlob);
+ }
+ float[] expectedVector = vectors.get(id);
+ if (expectedVector == null) {
+ assertThat(row.isNullAt(2)).isTrue();
+ } else {
+
assertThat(row.getVector(2).toFloatArray()).containsExactly(expectedVector);
+ }
+ }
+ }
+
private Table createTable(String format, String... sharedShreddingFields)
throws Exception {
return createTable(format, 2, sharedShreddingFields);
}
@@ -515,6 +944,179 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
return catalog.getTable(identifier(format));
}
+ private Table createBlobAndVectorTable(String format) throws Exception {
+ catalog.createTable(
+ identifier(format),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))
+ .column("payload", DataTypes.BLOB())
+ .column("embedding", DataTypes.VECTOR(3,
DataTypes.FLOAT()))
+ .option("bucket", "-1")
+ .option("file.format", format)
+ .option(CoreOptions.FILE_COMPRESSION.key(), "none")
+ .option(CoreOptions.WRITE_ONLY.key(), "true")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.VECTOR_FILE_FORMAT.key(), "json")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "2")
+ .build(),
+ true);
+ return catalog.getTable(identifier(format));
+ }
+
+ private Table createCompactingAppendOnlyTable(String format) throws
Exception {
+ catalog.createTable(
+ identifier(format),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("file.format", format)
+ .option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(),
"99")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "64")
+ .build(),
+ true);
+ return catalog.getTable(identifier(format));
+ }
+
+ private Table createPrimaryKeyTable(String format, boolean thinMode, int
maxColumns)
+ throws Exception {
+ return createPrimaryKeyTable(format, thinMode, maxColumns, true);
+ }
+
+ private Table createPrimaryKeyTable(
+ String format, boolean thinMode, int maxColumns, boolean
writeOnly) throws Exception {
+ return createPrimaryKeyTable(format, thinMode, maxColumns, writeOnly,
false);
+ }
+
+ private Table createPrimaryKeyTable(
+ String format,
+ boolean thinMode,
+ int maxColumns,
+ boolean writeOnly,
+ boolean deletionVectors)
+ throws Exception {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()));
+ if (deletionVectors) {
+ builder.column("padding", DataTypes.STRING())
+ .option(CoreOptions.TARGET_FILE_SIZE.key(), "1 kb")
+ .option(CoreOptions.FILE_COMPRESSION.key(), "none");
+ }
+ builder.primaryKey("id")
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("file.format", format)
+ .option(CoreOptions.WRITE_ONLY.key(),
String.valueOf(writeOnly))
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
String.valueOf(deletionVectors))
+ .option(CoreOptions.DATA_FILE_THIN_MODE.key(),
String.valueOf(thinMode))
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+ .option(
+ "fields.metrics.map.shared-shredding.max-columns",
+ String.valueOf(maxColumns));
+ catalog.createTable(identifier(format), builder.build(), true);
+ return catalog.getTable(identifier(format));
+ }
+
+ private Table createPrimaryKeyInputChangelogTable(
+ String dataFileFormat, String changelogFileFormat) throws
Exception {
+ catalog.createTable(
+ identifier(dataFileFormat),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))
+ .primaryKey("id")
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("file.format", dataFileFormat)
+ .option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
+ .option(CoreOptions.CHANGELOG_FILE_FORMAT.key(),
changelogFileFormat)
+ .option(CoreOptions.FILE_COMPRESSION.key(), "zstd")
+ .option(CoreOptions.CHANGELOG_FILE_COMPRESSION.key(),
"zstd")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "2")
+ .build(),
+ true);
+ return catalog.getTable(identifier(dataFileFormat));
+ }
+
+ private Table createPrimaryKeyAggregationTable(String format, boolean
thinMode)
+ throws Exception {
+ catalog.createTable(
+ identifier(format),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))
+ .primaryKey("id")
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("file.format", format)
+ .option(CoreOptions.DATA_FILE_THIN_MODE.key(),
String.valueOf(thinMode))
+ .option(CoreOptions.MERGE_ENGINE.key(), "aggregation")
+ .option("fields.metrics.aggregate-function",
"merge_map")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "2")
+ .build(),
+ true);
+ return catalog.getTable(identifier(format));
+ }
+
+ private Map<Integer, Map<String, Long>> readMapsById(ReadBuilder
readBuilder) throws Exception {
+ Map<Integer, Map<String, Long>> result = new LinkedHashMap<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ result.put(
+ row.getInt(0),
+ row.isNullAt(1) ? null :
toJavaMap(row.getMap(1))));
+ }
+ return result;
+ }
+
+ private List<CommitMessage> compactAndCommit(
+ Table table, BinaryRow partition, int bucket, boolean
fullCompaction) throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.withIOManager(ioManager);
+ write.compact(partition, bucket, fullCompaction);
+ List<CommitMessage> messages = write.prepareCommit();
+ commit.commit(messages);
+ return messages;
+ }
+ }
+
+ private List<CommitMessage> writeAndCommit(Table table, InternalRow...
rows) throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.withIOManager(ioManager);
+ for (InternalRow row : rows) {
+ write.write(row);
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ commit.commit(messages);
+ return messages;
+ }
+ }
+
private Table createComplexValueTable(String format) throws Exception {
catalog.createTable(
identifier(format),
@@ -822,6 +1424,14 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
return result;
}
+ private long numberOfSpillFiles() throws Exception {
+ try (Stream<java.nio.file.Path> paths = Files.walk(tempPath)) {
+ return paths.filter(Files::isRegularFile)
+ .filter(path ->
path.getFileName().toString().endsWith(".channel"))
+ .count();
+ }
+ }
+
private Map<String, Long> javaMapOf(Object... entries) {
Map<String, Long> map = new LinkedHashMap<>();
for (int i = 0; i < entries.length; i += 2) {
