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 f35f2d8467 [core] Support vector-store file compaction for data
evolution tables (#10749)
f35f2d8467 is described below
commit f35f2d84671a8c1429182d2c0f17ad8fd4f3efc2
Author: Wenchao Wu <[email protected]>
AuthorDate: Mon Oct 26 14:31:31 2026 +0800
[core] Support vector-store file compaction for data evolution tables
(#10749)
---
.../data-evolution-maintenance.mdx | 12 +-
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 11 +
.../DataEvolutionCompactCoordinator.java | 111 +++++++--
.../dataevolution/DataEvolutionCompactTask.java | 66 ++++-
.../DataEvolutionCompactTaskSerializer.java | 4 +
.../DataEvolutionNormalCompactTask.java | 70 +-----
.../DataEvolutionVectorCompactTask.java | 277 +++++++++++++++++++++
.../apache/paimon/append/VectorStoreTableTest.java | 235 +++++++++++++++++
.../DataEvolutionCompactCoordinatorTest.java | 189 +++++++++++++-
.../source/DataEvolutionTableCompactSource.java | 2 +-
.../paimon/spark/procedure/CompactProcedure.java | 22 +-
.../spark/procedure/CompactProcedureTestBase.scala | 67 +++++
13 files changed, 971 insertions(+), 101 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 7666899d42..f0fe004077 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -68,9 +68,15 @@ CALL sys.compact(`table` => 'default.target_table');
```
BLOB compaction is separately controlled by `blob-compaction.enabled`
-(default `false`). Dedicated vector-file compaction is not currently
-supported. See [Dedicated Compaction](../maintenance/dedicated-compaction)
-for scheduling compaction jobs.
+(default `false`). Dedicated vector-file compaction is separately controlled by
+`vector-compaction.enabled` (default `false`). It merges the vector files
+contained in one normal range, or in normal files merged by the same task, into
+files sized by `vector.target-file-size`, keeping the latest value of every
+vector column for each row. A group is skipped when some vector column is
+missing for part of the group's row range, or when its files neither overlap
+nor include more than one file smaller than `vector.target-file-size`. See
+[Dedicated Compaction](../maintenance/dedicated-compaction) for scheduling
+compaction jobs.
Ordinary compaction **does not physically remove rows hidden by deletion
vectors**. Removing their positions would change the alignment with untouched
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index d519e80197..05585ef5a4 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1963,6 +1963,12 @@ The size strategy ranges by the data size allocated to
each sorting task, which
<td>String</td>
<td>The Variant shredding schema for writing.</td>
</tr>
+ <tr>
+ <td><h5>vector-compaction.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to compact vector-store files when compacting a data
evolution table.</td>
+ </tr>
<tr>
<td><h5>vector-field</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 737152d89a..97b91e4f29 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -3168,6 +3168,13 @@ public class CoreOptions implements Serializable {
+ " Default is the same as
TARGET_FILE_SIZE.")
.build());
+ public static final ConfigOption<Boolean> VECTOR_COMPACTION_ENABLED =
+ key("vector-compaction.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to compact vector-store files when
compacting a data evolution table.");
+
public static final ConfigOption<Boolean> VECTOR_SEARCH_DISTRIBUTE_ENABLED
=
key("vector-search.distribute.enabled")
.booleanType()
@@ -4998,6 +5005,10 @@ public class CoreOptions implements Serializable {
.orElse(targetFileSize(false));
}
+ public boolean vectorCompactionEnabled() {
+ return options.get(VECTOR_COMPACTION_ENABLED);
+ }
+
public boolean vectorSearchDistributeEnabled() {
return options.get(VECTOR_SEARCH_DISTRIBUTE_ENABLED);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
index 5d80b8d971..c177481875 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
@@ -57,6 +57,7 @@ import java.util.stream.Collectors;
import static java.util.Comparator.comparingLong;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.types.BlobType.isBlobFileField;
+import static org.apache.paimon.types.VectorType.fieldsInVectorFile;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -66,7 +67,6 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
public class DataEvolutionCompactCoordinator {
private static final int BLOB_COMPACT_MIN_FILE_NUM = 2;
-
private final CompactScanner scanner;
private final CompactPlanner planner;
@@ -122,6 +122,11 @@ public class DataEvolutionCompactCoordinator {
.map(DataField::id)
.collect(Collectors.toSet())
: null;
+ Set<Integer> currentVectorFieldIds =
+ compactVector
+ ? DataEvolutionVectorCompactTask.fieldIds(
+ fieldsInVectorFile(table.rowType(),
options.withVectorFormat()))
+ : null;
DataEvolutionCompactRangePlanner.CandidateOptions candidateOptions =
new DataEvolutionCompactRangePlanner.CandidateOptions(
compactBlob,
@@ -152,7 +157,9 @@ public class DataEvolutionCompactCoordinator {
openFileCost,
compactMinFileNum,
schemaFetcher,
- currentBlobFieldIds);
+ currentBlobFieldIds,
+ options.vectorTargetFileSize(),
+ currentVectorFieldIds);
}
static long largeFileThreshold(long targetFileSize, double ratio) {
@@ -179,6 +186,16 @@ public class DataEvolutionCompactCoordinator {
return this;
}
+ /**
+ * Prevents this run from repeatedly compacting vector-store groups
containing only its own
+ * vector outputs. Size-based selection alone cannot guarantee this,
because formats roll files
+ * by estimated size.
+ */
+ public DataEvolutionCompactCoordinator
withCompletedVectorFiles(Set<String> fileNames) {
+ planner.withCompletedVectorFiles(fileNames);
+ return this;
+ }
+
public List<DataEvolutionCompactTask> plan() {
// scan files in snapshot
List<ManifestEntry> entries = scanner.scan();
@@ -266,7 +283,10 @@ public class DataEvolutionCompactCoordinator {
private final long compactMinFileNum;
private final LongFunction<RowType> schemaFetcher;
@Nullable private final Set<Integer> currentBlobFieldIds;
+ private final long vectorTargetFileSize;
+ @Nullable private final Set<Integer> currentVectorFieldIds;
private Set<String> completedNormalFiles = Collections.emptySet();
+ private Set<String> completedVectorFiles = Collections.emptySet();
@VisibleForTesting
CompactPlanner(
@@ -300,6 +320,32 @@ public class DataEvolutionCompactCoordinator {
long compactMinFileNum,
LongFunction<RowType> schemaFetcher,
@Nullable Set<Integer> currentBlobFieldIds) {
+ this(
+ compactBlob,
+ compactVector,
+ largeFileThreshold,
+ targetFileSize,
+ blobTargetFileSize,
+ openFileCost,
+ compactMinFileNum,
+ schemaFetcher,
+ currentBlobFieldIds,
+ targetFileSize,
+ null);
+ }
+
+ CompactPlanner(
+ boolean compactBlob,
+ boolean compactVector,
+ long largeFileThreshold,
+ long targetFileSize,
+ long blobTargetFileSize,
+ long openFileCost,
+ long compactMinFileNum,
+ LongFunction<RowType> schemaFetcher,
+ @Nullable Set<Integer> currentBlobFieldIds,
+ long vectorTargetFileSize,
+ @Nullable Set<Integer> currentVectorFieldIds) {
this.compactBlob = compactBlob;
this.compactVector = compactVector;
this.largeFileThreshold = largeFileThreshold;
@@ -311,6 +357,13 @@ public class DataEvolutionCompactCoordinator {
this.schemaFetcher =
schemaId -> schemaCache.computeIfAbsent(schemaId,
schemaFetcher::apply);
this.currentBlobFieldIds = currentBlobFieldIds;
+ this.vectorTargetFileSize = vectorTargetFileSize;
+ this.currentVectorFieldIds = currentVectorFieldIds;
+ }
+
+ CompactPlanner withCompletedVectorFiles(Set<String> fileNames) {
+ this.completedVectorFiles = checkNotNull(fileNames);
+ return this;
}
List<DataEvolutionCompactTask> compactPlan(List<ManifestEntry> input) {
@@ -490,24 +543,25 @@ public class DataEvolutionCompactCoordinator {
}
if (compactVector) {
- if (triggerNormalFile) {
- List<DataFileMeta> vectorStoreFiles = new ArrayList<>();
- for (DataFileMeta dataFile : dataFiles) {
- vectorStoreFiles.addAll(
- dataFileToVectorStoreFiles.getOrDefault(
- dataFile, Collections.emptyList()));
- }
- if (vectorStoreFiles.size() >= compactMinFileNum) {
- tasks.add(new
DataEvolutionNormalCompactTask(partition, vectorStoreFiles));
- }
+ List<DataFileMeta> binVectorStoreFiles = new ArrayList<>();
+ for (DataFileMeta dataFile : dataFiles) {
+ binVectorStoreFiles.addAll(
+ dataFileToVectorStoreFiles.getOrDefault(
+ dataFile, Collections.emptyList()));
+ }
+ // Output may span several normal files only when they are
merged in this batch.
+ if (triggerNormalFile
+ && dataFiles.size() >= compactMinFileNum
+ && vectorStoreFilesCompactable(binVectorStoreFiles)) {
+ tasks.add(new DataEvolutionVectorCompactTask(partition,
binVectorStoreFiles));
} else {
for (DataFileMeta dataFile : dataFiles) {
List<DataFileMeta> vectorStoreFiles =
dataFileToVectorStoreFiles.getOrDefault(
dataFile, Collections.emptyList());
- if (vectorStoreFiles.size() >= compactMinFileNum) {
+ if (vectorStoreFilesCompactable(vectorStoreFiles)) {
tasks.add(
- new DataEvolutionNormalCompactTask(
+ new DataEvolutionVectorCompactTask(
partition, vectorStoreFiles));
}
}
@@ -530,6 +584,35 @@ public class DataEvolutionCompactCoordinator {
return tasks;
}
+ private boolean vectorStoreFilesCompactable(List<DataFileMeta>
vectorStoreFiles) {
+ return vectorStoreFiles.size() > 1
+ && vectorStoreFiles.size() >= compactMinFileNum
+ && !vectorStoreFiles.stream()
+ .allMatch(file ->
completedVectorFiles.contains(file.fileName()))
+ && vectorStoreFilesNeedRewrite(vectorStoreFiles)
+ && DataEvolutionVectorCompactTask.coversRangeForEveryField(
+ vectorStoreFiles, schemaFetcher,
currentVectorFieldIds);
+ }
+
+ /**
+ * A rewrite is useful only if some rows are stored in several files
or several files are
+ * below the vector target size. Rolled compaction output normally has
neither, so a later
+ * run does not rewrite it again.
+ */
+ private boolean vectorStoreFilesNeedRewrite(List<DataFileMeta>
vectorStoreFiles) {
+ List<Range> ranges =
+ vectorStoreFiles.stream()
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList());
+ if (Range.sortAndMergeOverlap(ranges).size() <
vectorStoreFiles.size()) {
+ return true;
+ }
+ return vectorStoreFiles.stream()
+ .filter(file -> file.fileSize() <
vectorTargetFileSize)
+ .count()
+ > 1;
+ }
+
private boolean canSplit(Range normalRange, List<Range>
protectedRanges) {
return normalRange.from < normalRange.to
&& Range.sortAndMergeOverlap(protectedRanges).stream()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
index 6f24a41bcc..25930db78a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
@@ -23,20 +23,29 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.DataEvolutionUtils;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.function.Function;
import static java.util.Comparator.comparingLong;
import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
+import static
org.apache.paimon.utils.DataEvolutionUtils.fieldMaxSequenceNumber;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Base class for data evolution table compaction tasks. */
@@ -132,6 +141,60 @@ public abstract class DataEvolutionCompactTask {
"Cannot get max sequence id from
compact before files."));
}
+ @Nullable
+ protected long[] compactedWriteColsSequences(FileStoreTable table,
DataFileMeta outputFile) {
+ SchemaManager schemaManager = table.schemaManager();
+ Map<Long, TableSchema> schemaCache = new HashMap<>();
+ Function<Long, TableSchema> schemaLoader =
+ schemaId -> schemaCache.computeIfAbsent(schemaId,
schemaManager::schema);
+ Map<Pair<Long, List<String>>, List<DataField>> fileFieldsCache = new
HashMap<>();
+
+ Map<Integer, Long> fieldMaxSequences = new HashMap<>();
+ for (DataFileMeta input : compactBefore) {
+ List<DataField> inputFields =
+ fileFieldsCache.computeIfAbsent(
+ Pair.of(input.schemaId(), input.writeCols()),
+ key -> fileFields(schemaLoader, input));
+ long[] inputWriteColsSequences = input.writeColsSequences();
+ for (int inputPosition = 0; inputPosition < inputFields.size();
inputPosition++) {
+ fieldMaxSequences.merge(
+ inputFields.get(inputPosition).id(),
+ fieldMaxSequenceNumber(
+ input, inputWriteColsSequences, inputPosition,
inputFields.size()),
+ Math::max);
+ }
+ }
+
+ long fallbackSequence = outputFile.maxSequenceNumber();
+ List<DataField> outputFields =
+ fileFieldsCache.computeIfAbsent(
+ Pair.of(outputFile.schemaId(), outputFile.writeCols()),
+ key -> fileFields(schemaLoader, outputFile));
+ boolean allEqualToFileMax =
+ outputFields.stream()
+ .allMatch(
+ field ->
+
fieldMaxSequences.getOrDefault(field.id(), fallbackSequence)
+ == fallbackSequence);
+ if (allEqualToFileMax) {
+ return null;
+ }
+
+ long[] result = new long[outputFields.size()];
+ for (int outputPosition = 0; outputPosition < outputFields.size();
outputPosition++) {
+ result[outputPosition] =
+ fieldMaxSequences.getOrDefault(
+ outputFields.get(outputPosition).id(),
fallbackSequence);
+ }
+ return result;
+ }
+
+ private static List<DataField> fileFields(
+ Function<Long, TableSchema> schemaLoader, DataFileMeta file) {
+ TableSchema fileSchema = schemaLoader.apply(file.schemaId());
+ return DataEvolutionUtils.fileFields(fileSchema, file);
+ }
+
@Override
public int hashCode() {
return Objects.hash(partition, compactBefore, compactAfter);
@@ -163,7 +226,8 @@ public abstract class DataEvolutionCompactTask {
public enum TaskType {
NORMAL(0),
BLOB(1),
- MATERIALIZE_DELETION(2);
+ MATERIALIZE_DELETION(2),
+ VECTOR(3);
private final int code;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
index 5b4e463dcb..d7ede1fb8b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
@@ -43,6 +43,8 @@ import static
org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
public class DataEvolutionCompactTaskSerializer
implements VersionedSerializer<DataEvolutionCompactTask> {
+ // A task type without its own payload keeps the version, so restored jobs
can still read
+ // in-flight tasks. Only older binaries reading the new type fail, at
TaskType.fromCode.
private static final int CURRENT_VERSION = 4;
private final DataFileMetaSerializer dataFileSerializer;
@@ -135,6 +137,8 @@ public class DataEvolutionCompactTaskSerializer
return new DataEvolutionNormalCompactTask(partition, files,
ranges);
case BLOB:
return new DataEvolutionBlobCompactTask(partition, files);
+ case VECTOR:
+ return new DataEvolutionVectorCompactTask(partition, files);
case MATERIALIZE_DELETION:
return new DataEvolutionMaterializeDeletionCompactTask(
partition,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
index 7065a37bc1..3b78099cf2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
@@ -28,15 +28,11 @@ import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.operation.AppendFileStoreWrite;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.reader.RecordReaderIterator;
-import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
-import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RecordWriter;
import org.apache.paimon.utils.SetUtils;
@@ -44,8 +40,6 @@ import org.apache.paimon.utils.SetUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -53,14 +47,12 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.paimon.types.BlobType.fieldNamesInBlobFile;
import static org.apache.paimon.types.VectorType.fieldNamesInVectorFile;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
-import static
org.apache.paimon.utils.DataEvolutionUtils.fieldMaxSequenceNumber;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Compacts normal structured files of a data evolution table. */
@@ -106,10 +98,10 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
public CommitMessage doCompact(FileStoreTable table, String commitUser)
throws Exception {
CoreOptions options = table.coreOptions();
- if (isVectorStoreFile(compactBefore.get(0).fileName())) {
- // TODO: support vector-store file compaction
- throw new UnsupportedOperationException("Vector-store task is not
supported");
- }
+ checkArgument(
+ !isVectorStoreFile(compactBefore.get(0).fileName()),
+ "Vector-store files are compacted by %s.",
+ DataEvolutionVectorCompactTask.class.getSimpleName());
Set<String> fieldsInDedicatedFile =
SetUtils.union(
@@ -228,58 +220,4 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
start = end + 1;
}
}
-
- @Nullable
- private long[] compactedWriteColsSequences(FileStoreTable table,
DataFileMeta outputFile) {
- SchemaManager schemaManager = table.schemaManager();
- Map<Long, TableSchema> schemaCache = new HashMap<>();
- Function<Long, TableSchema> schemaLoader =
- schemaId -> schemaCache.computeIfAbsent(schemaId,
schemaManager::schema);
- Map<Pair<Long, List<String>>, List<DataField>> fileFieldsCache = new
HashMap<>();
-
- Map<Integer, Long> fieldMaxSequences = new HashMap<>();
- for (DataFileMeta input : compactBefore) {
- List<DataField> inputFields =
- fileFieldsCache.computeIfAbsent(
- Pair.of(input.schemaId(), input.writeCols()),
- key -> fileFields(schemaLoader, input));
- long[] inputWriteColsSequences = input.writeColsSequences();
- for (int inputPosition = 0; inputPosition < inputFields.size();
inputPosition++) {
- fieldMaxSequences.merge(
- inputFields.get(inputPosition).id(),
- fieldMaxSequenceNumber(
- input, inputWriteColsSequences, inputPosition,
inputFields.size()),
- Math::max);
- }
- }
-
- long fallbackSequence = outputFile.maxSequenceNumber();
- List<DataField> outputFields =
- fileFieldsCache.computeIfAbsent(
- Pair.of(outputFile.schemaId(), outputFile.writeCols()),
- key -> fileFields(schemaLoader, outputFile));
- boolean allEqualToFileMax =
- outputFields.stream()
- .allMatch(
- field ->
-
fieldMaxSequences.getOrDefault(field.id(), fallbackSequence)
- == fallbackSequence);
- if (allEqualToFileMax) {
- return null;
- }
-
- long[] result = new long[outputFields.size()];
- for (int outputPosition = 0; outputPosition < outputFields.size();
outputPosition++) {
- result[outputPosition] =
- fieldMaxSequences.getOrDefault(
- outputFields.get(outputPosition).id(),
fallbackSequence);
- }
- return result;
- }
-
- private static List<DataField> fileFields(
- Function<Long, TableSchema> schemaLoader, DataFileMeta file) {
- TableSchema fileSchema = schemaLoader.apply(file.schemaId());
- return
org.apache.paimon.utils.DataEvolutionUtils.fileFields(fileSchema, file);
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionVectorCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionVectorCompactTask.java
new file mode 100644
index 0000000000..e61b287e8e
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionVectorCompactTask.java
@@ -0,0 +1,277 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.AppendOnlyFileStore;
+import org.apache.paimon.CoreOptions;
+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.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.io.RollingFileWriterImpl;
+import org.apache.paimon.io.RowDataFileWriter;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.LongCounter;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.StatsCollectorFactories;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.LongFunction;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.types.VectorType.fieldsInVectorFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Compacts the dedicated vector-store files of one normal row-ID range of a
data evolution table.
+ *
+ * <p>The task must receive every vector-store file of that range. The
rewritten files take the
+ * highest sequence number of the inputs, so a vector-store file left out of
the task could start or
+ * stop winning rows against them.
+ */
+public class DataEvolutionVectorCompactTask extends DataEvolutionCompactTask {
+
+ public DataEvolutionVectorCompactTask(BinaryRow partition,
List<DataFileMeta> files) {
+ super(partition, files);
+ checkArgument(
+ files.stream().allMatch(file ->
isVectorStoreFile(file.fileName())),
+ "Vector compaction task %s should only contain vector-store
files.",
+ files);
+ checkContiguousRowRange(files);
+ }
+
+ @Override
+ public TaskType type() {
+ return TaskType.VECTOR;
+ }
+
+ /**
+ * Whether every current vector field of {@code files} is present in some
file for every row of
+ * their combined range. Otherwise the rewrite would store NULL for rows
that never had the
+ * field. Fields dropped from the table are ignored because the rewrite
does not keep them; a
+ * null {@code currentFieldIds} checks every field.
+ */
+ static boolean coversRangeForEveryField(
+ List<DataFileMeta> files,
+ LongFunction<RowType> schemaFetcher,
+ @Nullable Set<Integer> currentFieldIds) {
+ List<Range> merged =
+ Range.sortAndMergeOverlap(
+ files.stream()
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()),
+ true);
+ if (merged.size() != 1) {
+ return false;
+ }
+
+ Map<Integer, List<Range>> fieldRanges = new HashMap<>();
+ Map<Long, RowType> schemaCache = new HashMap<>();
+ for (DataFileMeta file : files) {
+ for (int fieldId : fieldIds(file, schemaCache, schemaFetcher)) {
+ if (currentFieldIds == null ||
currentFieldIds.contains(fieldId)) {
+ fieldRanges
+ .computeIfAbsent(fieldId, ignored -> new
ArrayList<>())
+ .add(file.nonNullRowIdRange());
+ }
+ }
+ }
+ if (fieldRanges.isEmpty()) {
+ return false;
+ }
+ for (List<Range> ranges : fieldRanges.values()) {
+ if (!Range.sortAndMergeOverlap(ranges, true).equals(merged)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public CommitMessage doCompact(FileStoreTable table, String commitUser)
throws Exception {
+ CoreOptions options = table.coreOptions();
+ List<DataFileMeta> sortedCompactBefore =
sortedByFirstRowId(compactBefore);
+ checkArgument(
+ sortedCompactBefore.size() > 1,
+ "Vector compaction task %s should contain at least two files
to compact.",
+ this);
+ FileFormat vectorFileFormat =
+ checkNotNull(
+ FileFormat.vectorFileFormat(options),
+ "Vector compaction requires option '%s'.",
+ CoreOptions.VECTOR_FILE_FORMAT.key());
+ LongFunction<RowType> schemaFetcher =
+ schemaId ->
table.schemaManager().schema(schemaId).logicalRowType();
+ List<DataField> currentFields = fieldsInVectorFile(table.rowType(),
true);
+ checkArgument(
+ coversRangeForEveryField(
+ sortedCompactBefore, schemaFetcher,
fieldIds(currentFields)),
+ "Every current vector field of compaction task %s should cover
the whole row range.",
+ this);
+ Range compactBeforeRange =
checkContiguousRowRange(sortedCompactBefore);
+ RowType writeType =
+ new RowType(inputFields(currentFields, sortedCompactBefore,
schemaFetcher));
+
+ AppendOnlyFileStore store = (AppendOnlyFileStore) table.store();
+ DataFilePathFactory pathFactory =
+ store.pathFactory().createDataFilePathFactory(partition, 0);
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(0)
+ .withDataFiles(sortedCompactBefore)
+ .withBucketPath(pathFactory.parent().toString())
+ .rawConvertible(false)
+ .build();
+ RecordReader<InternalRow> reader =
+
store.newDataEvolutionRead().withReadType(writeType).createReader(dataSplit);
+ RollingFileWriterImpl<InternalRow, DataFileMeta> writer =
+ createVectorFileWriter(table, options, vectorFileFormat,
writeType, pathFactory);
+
+ long rowCount = 0;
+ try (RecordReaderIterator<InternalRow> iterator = new
RecordReaderIterator<>(reader)) {
+ while (iterator.hasNext()) {
+ writer.write(iterator.next());
+ rowCount++;
+ }
+ writer.close();
+ } catch (Exception e) {
+ writer.abort();
+ throw e;
+ }
+ checkArgument(
+ rowCount == compactBeforeRange.count(),
+ "Vector compaction read %s rows, but the compacted range %s
has %s rows.",
+ rowCount,
+ compactBeforeRange,
+ compactBeforeRange.count());
+
+ long minSequenceNumber = minSequenceId(sortedCompactBefore);
+ long maxSequenceNumber = maxSequenceId(sortedCompactBefore);
+ long nextRowId = compactBeforeRange.from;
+ List<DataFileMeta> writeResult = writer.result();
+ long[] writeColsSequences =
+ options.ignoreIndexColumnUpdate() && !writeResult.isEmpty()
+ ? compactedWriteColsSequences(
+ table,
+ writeResult
+ .get(0)
+
.assignSequenceNumber(minSequenceNumber, maxSequenceNumber))
+ : null;
+ for (DataFileMeta file : writeResult) {
+ DataFileMeta compacted =
+ file.assignFirstRowId(nextRowId)
+ .assignSequenceNumber(minSequenceNumber,
maxSequenceNumber);
+ if (writeColsSequences != null) {
+ compacted =
compacted.withWriteColsSequences(writeColsSequences);
+ }
+ compactAfter.add(compacted);
+ nextRowId += compacted.rowCount();
+ }
+ checkSameRowRange("Vector compact files", sortedCompactBefore,
compactAfter);
+
+ return commitMessage(sortedCompactBefore, compactAfter);
+ }
+
+ private RollingFileWriterImpl<InternalRow, DataFileMeta>
createVectorFileWriter(
+ FileStoreTable table,
+ CoreOptions options,
+ FileFormat vectorFileFormat,
+ RowType writeType,
+ DataFilePathFactory pathFactory) {
+ List<String> fieldNames = writeType.getFieldNames();
+ StatsCollectorFactories statsCollectorFactories = new
StatsCollectorFactories(options);
+ return new RollingFileWriterImpl<>(
+ () ->
+ new RowDataFileWriter(
+ table.fileIO(),
+ RollingFileWriter.createFileWriterContext(
+ vectorFileFormat,
+ writeType,
+
statsCollectorFactories.statsCollectors(fieldNames),
+ options.fileCompression()),
+
pathFactory.newVectorPath(vectorFileFormat.getFormatIdentifier()),
+ writeType,
+ table.schema().id(),
+ () -> new LongCounter(0),
+ new FileIndexOptions(),
+ FileSource.COMPACT,
+ false,
+ options.statsDenseStore(),
+ pathFactory.isExternalPath(),
+ fieldNames,
+ null,
+ null),
+ options.vectorTargetFileSize(),
+ Long.MAX_VALUE);
+ }
+
+ /** Current vector-store fields written by any input, in table order. */
+ private static List<DataField> inputFields(
+ List<DataField> currentFields,
+ List<DataFileMeta> files,
+ LongFunction<RowType> schemaFetcher) {
+ Set<Integer> inputFieldIds = new HashSet<>();
+ Map<Long, RowType> schemaCache = new HashMap<>();
+ for (DataFileMeta file : files) {
+ inputFieldIds.addAll(fieldIds(file, schemaCache, schemaFetcher));
+ }
+ return currentFields.stream()
+ .filter(field -> inputFieldIds.contains(field.id()))
+ .collect(Collectors.toList());
+ }
+
+ static Set<Integer> fieldIds(List<DataField> fields) {
+ return fields.stream().map(DataField::id).collect(Collectors.toSet());
+ }
+
+ private static List<Integer> fieldIds(
+ DataFileMeta file,
+ Map<Long, RowType> schemaCache,
+ LongFunction<RowType> schemaFetcher) {
+ checkArgument(
+ file.writeCols() != null && !file.writeCols().isEmpty(),
+ "Vector-store file %s should record its write columns.",
+ file);
+ RowType rowType = schemaCache.computeIfAbsent(file.schemaId(),
schemaFetcher::apply);
+ return file.writeCols().stream()
+ .map(name -> rowType.getField(name).id())
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
index ea6d242f40..0de783b084 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
@@ -65,6 +65,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -477,6 +478,240 @@ public class VectorStoreTableTest extends
DataEvolutionTestBase {
assertThat(readIdAndVector(builder)).containsExactly("1=200.0",
"4=null", "7=300.0");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "parquet"})
+ public void testCompactVectorFilesKeepsLatestColumnValues(String format)
throws Exception {
+ catalog.createTable(
+ identifier(),
+ vectorSchema(format)
+ .column("embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))
+ .build(),
+ false);
+ write(
+ getTableDefault(),
+ GenericRow.of(0, vector(1), vector(10)),
+ GenericRow.of(1, vector(1), vector(10)));
+ write(
+ getTableDefault(),
+ GenericRow.of(2, vector(1), vector(10)),
+ GenericRow.of(3, vector(1), vector(10)));
+ updateVectors(
+ 2,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(100)),
+ GenericRow.of(vector(100)));
+ updateVectors(
+ 2,
+ Arrays.asList("embedding", "embedding_v2"),
+ GenericRow.of(vector(2), vector(200)),
+ GenericRow.of(vector(2), vector(200)));
+ updateVectors(
+ 0,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(300)),
+ GenericRow.of(vector(300)));
+ List<String> expected = readAllVectors(getTableDefault());
+ assertThat(expected)
+ .containsExactly("0|1.0|300.0", "1|1.0|300.0", "2|2.0|200.0",
"3|2.0|200.0");
+ assertThat(vectorFiles(getTableDefault())).hasSize(5);
+
+ compactVectorFiles();
+
+ List<DataFileMeta> vectorFiles = vectorFiles(getTableDefault());
+ assertThat(vectorFiles).hasSize(1);
+ assertThat(vectorFiles.get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 3));
+
assertThat(vectorFiles.get(0).writeCols()).containsExactly("embedding",
"embedding_v2");
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+ assertThat(read(getTableDefault(), new int[] {2}))
+ .extracting(row -> row.getVector(0).toFloatArray()[0])
+ .containsExactly(300F, 300F, 200F, 200F);
+
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.renameColumn("embedding_v2",
"query_embedding")),
+ false);
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+ }
+
+ @Test
+ public void testCompactVectorFilesSkipsPartiallyPopulatedColumn() throws
Exception {
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ write(getTableDefault(), GenericRow.of(0, vector(1)), GenericRow.of(1,
vector(1)));
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.addColumn(
+ "embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))),
+ false);
+ write(
+ getTableDefault(),
+ GenericRow.of(2, vector(1), vector(10)),
+ GenericRow.of(3, vector(1), vector(10)));
+ updateVectors(
+ 2,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(200)),
+ GenericRow.of(vector(200)));
+ List<String> expected = readAllVectors(getTableDefault());
+ assertThat(expected)
+ .containsExactly("0|1.0|null", "1|1.0|null", "2|1.0|200.0",
"3|1.0|200.0");
+
+ compactVectorFiles();
+
+ // embedding_v2 is missing for rows [0, 1], so the merged normal range
keeps the vector
+ // file of [0, 1] and only the two vector files of [2, 3] are
compacted.
+ List<DataFileMeta> vectorFiles = vectorFiles(getTableDefault());
+ assertThat(vectorFiles)
+ .extracting(DataFileMeta::nonNullRowIdRange)
+ .containsExactlyInAnyOrder(new Range(0, 1), new Range(2, 3));
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+ }
+
+ @Test
+ public void testCompactVectorFilesIgnoresDroppedColumn() throws Exception {
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ write(getTableDefault(), GenericRow.of(0, vector(1)), GenericRow.of(1,
vector(1)));
+ write(getTableDefault(), GenericRow.of(2, vector(2)), GenericRow.of(3,
vector(2)));
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.addColumn(
+ "embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))),
+ false);
+ updateVectors(
+ 2,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(200)),
+ GenericRow.of(vector(200)));
+ catalog.alterTable(
+ identifier(),
+
Collections.singletonList(SchemaChange.dropColumn("embedding_v2")),
+ false);
+ List<String> expected = readAllVectors(getTableDefault());
+ assertThat(expected).containsExactly("0|1.0", "1|1.0", "2|2.0",
"3|2.0");
+
+ compactVectorFiles();
+
+ List<DataFileMeta> vectorFiles = vectorFiles(getTableDefault());
+ assertThat(vectorFiles).hasSize(1);
+ assertThat(vectorFiles.get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 3));
+
assertThat(vectorFiles.get(0).writeCols()).containsExactly("embedding");
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+ }
+
+ @Test
+ public void
testCompactVectorFilesKeepsColumnSequencesWhenIgnoringIndexUpdates()
+ throws Exception {
+ catalog.createTable(
+ identifier(),
+ vectorSchema("json")
+ .column("embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))
+
.option(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(), "IGNORE")
+ .build(),
+ false);
+ write(getTableDefault(), GenericRow.of(0, vector(1), vector(10)));
+ write(getTableDefault(), GenericRow.of(1, vector(1), vector(10)));
+ updateVectors(0, Collections.singletonList("embedding_v2"),
GenericRow.of(vector(100)));
+ long embeddingSequence =
+ vectorFiles(getTableDefault()).stream()
+ .filter(file -> file.writeCols().contains("embedding"))
+ .mapToLong(DataFileMeta::maxSequenceNumber)
+ .max()
+ .getAsLong();
+ long updateSequence =
+ vectorFiles(getTableDefault()).stream()
+ .mapToLong(DataFileMeta::maxSequenceNumber)
+ .max()
+ .getAsLong();
+ assertThat(embeddingSequence).isLessThan(updateSequence);
+ List<String> expected = readAllVectors(getTableDefault());
+
+ compactVectorFiles();
+
+ List<DataFileMeta> vectorFiles = vectorFiles(getTableDefault());
+ assertThat(vectorFiles).hasSize(1);
+
assertThat(vectorFiles.get(0).writeCols()).containsExactly("embedding",
"embedding_v2");
+
assertThat(vectorFiles.get(0).maxSequenceNumber()).isEqualTo(updateSequence);
+ assertThat(vectorFiles.get(0).writeColsSequences())
+ .containsExactly(embeddingSequence, updateSequence);
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+ }
+
+ @Test
+ public void testCompactVectorFilesRollsByVectorTargetSize() throws
Exception {
+ catalog.createTable(identifier(), vectorSchema("parquet").build(),
false);
+ int rowsPerWrite = 1250;
+ for (int write = 0; write < 2; write++) {
+ InternalRow[] rows = new InternalRow[rowsPerWrite];
+ for (int i = 0; i < rowsPerWrite; i++) {
+ int id = write * rowsPerWrite + i;
+ rows[i] = GenericRow.of(id, vector(id));
+ }
+ write(getTableDefault(), rows);
+ }
+ // Non-overlapping files are only rewritten when small; overlap them
with an update.
+ updateVectors(0, Collections.singletonList("embedding"),
vectorRows(rowsPerWrite, 7));
+ List<String> expected = readAllVectors(getTableDefault());
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+
SchemaChange.setOption(CoreOptions.VECTOR_TARGET_FILE_SIZE.key(), "1 b")),
+ false);
+
+ compactVectorFiles();
+
+ // The rolling writer checks the target size every 1000 rows.
+ List<DataFileMeta> vectorFiles = vectorFiles(getTableDefault());
+
vectorFiles.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
+ assertThat(vectorFiles)
+ .extracting(DataFileMeta::nonNullRowIdRange)
+ .containsExactly(new Range(0, 999), new Range(1000, 1999), new
Range(2000, 2499));
+ assertThat(readAllVectors(getTableDefault())).isEqualTo(expected);
+
+ FileStoreTable table = getTableDefault();
+ assertThat(
+ new DataEvolutionCompactCoordinator(
+ table, false, true,
table.latestSnapshot().get())
+ .plan())
+ .isEmpty();
+ }
+
+ private List<String> readAllVectors(FileStoreTable table) throws Exception
{
+ List<String> result = new ArrayList<>();
+ for (InternalRow row : read(table)) {
+ StringBuilder builder = new StringBuilder().append(row.getInt(0));
+ for (int i = 1; i < row.getFieldCount(); i++) {
+ builder.append('|')
+ .append(row.isNullAt(i) ? null :
row.getVector(i).toFloatArray()[0]);
+ }
+ result.add(builder.toString());
+ }
+ return result;
+ }
+
+ private static List<DataFileMeta> vectorFiles(FileStoreTable table) {
+ return table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file -> file.fileName().contains(".vector."))
+ .collect(Collectors.toList());
+ }
+
+ private void compactVectorFiles() throws Exception {
+ FileStoreTable table = getTableDefault();
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataEvolutionCompactTask task :
+ new DataEvolutionCompactCoordinator(table, false, true,
snapshot).plan()) {
+ messages.add(task.doCompact(table, commitUser));
+ }
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
private static List<String> readIdAndVector(ReadBuilder builder) throws
IOException {
List<String> actual = new ArrayList<>();
try (RecordReader<InternalRow> reader =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
index f222705a1e..f6e9090b4d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
@@ -527,10 +527,127 @@ public class DataEvolutionCompactCoordinatorTest {
assertThat(tasks).hasSize(2);
assertThat(tasks.get(0).compactBefore())
.containsExactly(entries.get(0).file(), entries.get(2).file());
+
assertThat(tasks.get(1)).isInstanceOf(DataEvolutionVectorCompactTask.class);
assertThat(tasks.get(1).compactBefore())
.containsExactly(entries.get(1).file(), entries.get(3).file());
}
+ @Test
+ public void testCompactPlannerCompactsVectorStoreFilesWithinOneDataFile() {
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(makeEntry("file1.parquet", 0L, 100L, 100));
+ entries.add(makeVectorStoreEntry("v1.vector.json", 0L, 50L, 10));
+ entries.add(makeVectorStoreEntry("v2.vector.json", 50L, 50L, 10));
+
+ List<DataEvolutionCompactTask> tasks =
+ vectorStorePlanner(1024, 1024, 2).compactPlan(entries);
+
+ assertThat(tasks).hasSize(1);
+
assertThat(tasks.get(0).type()).isEqualTo(DataEvolutionCompactTask.TaskType.VECTOR);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactly(entries.get(1).file(), entries.get(2).file());
+
+ tasks = vectorStorePlanner(1024, 1024, 3).compactPlan(entries);
+ assertThat(tasks).isEmpty();
+ }
+
+ @Test
+ public void testCompactPlannerSkipsCompactedVectorStoreFiles() {
+ // Rolled compaction output: non-overlapping files, only the last
below the target size.
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(makeEntry("file1.parquet", 0L, 300L, 100));
+ entries.add(makeVectorStoreEntry("v1.vector.json", 0L, 100L, 1024));
+ entries.add(makeVectorStoreEntry("v2.vector.json", 100L, 100L, 1024));
+ entries.add(makeVectorStoreEntry("v3.vector.json", 200L, 100L, 10));
+ assertThat(vectorStorePlanner(1024, 1024,
2).compactPlan(entries)).isEmpty();
+
+ // Two small files are worth merging, and the task still takes every
file of the range.
+ List<ManifestEntry> twoSmall = new ArrayList<>(entries);
+ twoSmall.set(2, makeVectorStoreEntry("v2.vector.json", 100L, 100L,
10));
+ List<DataEvolutionCompactTask> tasks =
+ vectorStorePlanner(1024, 1024, 2).compactPlan(twoSmall);
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactly(
+ twoSmall.get(1).file(), twoSmall.get(2).file(),
twoSmall.get(3).file());
+
+ // Overlapping files are always merged, regardless of their sizes.
+ List<ManifestEntry> overlapping = new ArrayList<>(entries);
+ overlapping.add(makeVectorStoreEntry("update.vector.json", 0L, 100L,
1024));
+ tasks = vectorStorePlanner(1024, 1024, 2).compactPlan(overlapping);
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore()).hasSize(4);
+ }
+
+ @Test
+ public void testCompactPlannerSkipsCompletedVectorStoreFiles() {
+ // Estimated rolling sizes may leave several outputs below the target
size.
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(makeEntry("file1.parquet", 0L, 200L, 100));
+ entries.add(makeVectorStoreEntry("out1.vector.json", 0L, 100L, 10));
+ entries.add(makeVectorStoreEntry("out2.vector.json", 100L, 100L, 10));
+ Set<String> completed =
+ new HashSet<>(Arrays.asList("out1.vector.json",
"out2.vector.json"));
+ assertThat(vectorStorePlanner(1024, 1024,
2).compactPlan(entries)).hasSize(1);
+ assertThat(
+ vectorStorePlanner(1024, 1024, 2)
+ .withCompletedVectorFiles(completed)
+ .compactPlan(entries))
+ .isEmpty();
+
+ // A new file in the range requires rewriting the completed outputs
with it.
+ entries.add(makeVectorStoreEntry("update.vector.json", 0L, 100L, 10));
+ List<DataEvolutionCompactTask> tasks =
+ vectorStorePlanner(1024, 1024, 2)
+ .withCompletedVectorFiles(completed)
+ .compactPlan(entries);
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore()).hasSize(3);
+ }
+
+ @Test
+ public void testCompactPlannerSkipsVectorFieldMissingForPartOfRange() {
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(makeEntry("file1.parquet", 0L, 100L, 100));
+ entries.add(makeVectorStoreEntry("v1.vector.json", 0L, 100L, 1, 10,
Arrays.asList("vec")));
+ entries.add(makeVectorStoreEntry("v2.vector.json", 50L, 50L, 2, 10,
Arrays.asList("vec2")));
+
+ assertThat(vectorStorePlanner(1024, 1024,
2).compactPlan(entries)).isEmpty();
+
+ entries.add(makeVectorStoreEntry("v3.vector.json", 0L, 50L, 3, 10,
Arrays.asList("vec2")));
+ List<DataEvolutionCompactTask> tasks =
+ vectorStorePlanner(1024, 1024, 2).compactPlan(entries);
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactlyInAnyOrder(
+ entries.get(1).file(), entries.get(2).file(),
entries.get(3).file());
+ }
+
+ @Test
+ public void testCompactPlannerIgnoresDroppedVectorFieldCoverage() {
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(makeEntry("file1.parquet", 0L, 100L, 100));
+ entries.add(makeVectorStoreEntry("v1.vector.json", 0L, 100L, 1, 10,
Arrays.asList("vec")));
+ entries.add(
+ makeVectorStoreEntry(
+ "v2.vector.json", 50L, 50L, 2, 10,
Arrays.asList("vec", "dropped")));
+
+ List<DataEvolutionCompactTask> tasks =
+ vectorStorePlanner(1024, 1024, 2).compactPlan(entries);
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactly(entries.get(1).file(), entries.get(2).file());
+
+ List<ManifestEntry> droppedOnly =
+ Arrays.asList(
+ makeEntry("file1.parquet", 0L, 100L, 100),
+ makeVectorStoreEntry(
+ "v1.vector.json", 0L, 50L, 1, 10,
Arrays.asList("dropped")),
+ makeVectorStoreEntry(
+ "v2.vector.json", 50L, 50L, 2, 10,
Arrays.asList("dropped")));
+ assertThat(vectorStorePlanner(1024, 1024,
2).compactPlan(droppedOnly)).isEmpty();
+ }
+
@Test
public void testCompactPlannerSkipsSingleFilePerBlobField() {
List<ManifestEntry> entries = new ArrayList<>();
@@ -954,19 +1071,23 @@ public class DataEvolutionCompactCoordinatorTest {
private ManifestEntry makeVectorStoreEntry(
String fileName, long firstRowId, long rowCount, long fileSize) {
+ return makeVectorStoreEntry(
+ fileName, firstRowId, rowCount, 0, fileSize,
Collections.singletonList("vec"));
+ }
+
+ private ManifestEntry makeVectorStoreEntry(
+ String fileName,
+ long firstRowId,
+ long rowCount,
+ long maxSeq,
+ long fileSize,
+ List<String> writeCols) {
return ManifestEntry.create(
FileKind.ADD,
BinaryRow.EMPTY_ROW,
0,
0,
- createDataFileMeta(
- fileName,
- firstRowId,
- rowCount,
- 0,
- fileSize,
- 0,
- Collections.singletonList("vec")));
+ createDataFileMeta(fileName, firstRowId, rowCount, maxSeq,
fileSize, 0, writeCols));
}
private DataFileMeta createDataFileMeta(
@@ -1042,8 +1163,26 @@ public class DataEvolutionCompactCoordinatorTest {
private DataEvolutionCompactCoordinator.CompactPlanner vectorStorePlanner(
long targetFileSize, long openFileCost, long compactMinFileNum) {
+ RowType rowType =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "vec", DataTypes.VECTOR(2,
DataTypes.FLOAT())),
+ new DataField(2, "vec2", DataTypes.VECTOR(2,
DataTypes.FLOAT())),
+ new DataField(
+ 3, "dropped", DataTypes.VECTOR(2,
DataTypes.FLOAT()))));
return new DataEvolutionCompactCoordinator.CompactPlanner(
- false, true, targetFileSize, openFileCost, compactMinFileNum);
+ false,
+ true,
+ Long.MAX_VALUE,
+ targetFileSize,
+ targetFileSize,
+ openFileCost,
+ compactMinFileNum,
+ schemaId -> rowType,
+ null,
+ targetFileSize,
+ new HashSet<>(Arrays.asList(1, 2)));
}
private RowType rowType(DataField... fields) {
@@ -1101,6 +1240,38 @@ public class DataEvolutionCompactCoordinatorTest {
assertThat(deserialized.type()).isEqualTo(DataEvolutionCompactTask.TaskType.BLOB);
}
+ @Test
+ public void testSerializerVectorTask() throws IOException {
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+
+ List<DataFileMeta> files =
+ Arrays.asList(
+ createDataFileMeta("file1.vector.json", 0L, 100L, 0,
1024),
+ createDataFileMeta("file2.vector.json", 100L, 100L, 0,
1024));
+
+ DataEvolutionCompactTask task =
+ new DataEvolutionVectorCompactTask(BinaryRow.EMPTY_ROW, files);
+
+ byte[] bytes = serializer.serialize(task);
+ DataEvolutionCompactTask deserialized =
+ serializer.deserialize(serializer.getVersion(), bytes);
+
+ assertThat(deserialized).isEqualTo(task);
+
assertThat(deserialized.type()).isEqualTo(DataEvolutionCompactTask.TaskType.VECTOR);
+ }
+
+ @Test
+ public void testVectorCompactTaskRejectsNonVectorFiles() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ createDataFileMeta("file1.vector.json", 0L, 10L, 0,
1024),
+ createDataFileMeta("file2.parquet", 10L, 10L, 0,
1024));
+
+ assertThatThrownBy(() -> new
DataEvolutionVectorCompactTask(BinaryRow.EMPTY_ROW, files))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("only contain vector-store files");
+ }
+
@Test
public void testSerializerWithPartition() throws IOException {
DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionTableCompactSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionTableCompactSource.java
index b2546428fe..78618366b5 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionTableCompactSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionTableCompactSource.java
@@ -89,7 +89,7 @@ public class DataEvolutionTableCompactSource
table,
partitions,
table.coreOptions().blobCompactionEnabled(),
- false,
+ table.coreOptions().vectorCompactionEnabled(),
snapshot);
taskPlanner = coordinator::plan;
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 218dc6d0a9..021c04d7af 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -669,6 +669,7 @@ public class CompactProcedure extends BaseProcedure {
}
AtomicReference<DataEvolutionCompactCoordinator> coordinatorRef = new
AtomicReference<>();
Set<String> completedNormalFiles = new HashSet<>();
+ Set<String> completedVectorFiles = new HashSet<>();
Function<Snapshot, List<DataEvolutionCompactTask>> taskPlanner =
planningSnapshot -> {
DataEvolutionCompactCoordinator coordinator =
coordinatorRef.get();
@@ -680,34 +681,41 @@ public class CompactProcedure extends BaseProcedure {
table,
partitionPredicate,
table.coreOptions().blobCompactionEnabled(),
- false,
+
table.coreOptions().vectorCompactionEnabled(),
planningSnapshot)
: new DataEvolutionCompactCoordinator(
table,
partitionPredicate,
table.coreOptions().blobCompactionEnabled(),
- false,
+
table.coreOptions().vectorCompactionEnabled(),
planningSnapshot,
candidateFilesPerBatch);
-
coordinator.withCompletedNormalFiles(completedNormalFiles);
+ coordinator
+ .withCompletedNormalFiles(completedNormalFiles)
+
.withCompletedVectorFiles(completedVectorFiles);
coordinatorRef.set(coordinator);
}
return filterIdlePartitions(
coordinator.plan(), table, partitionPredicate,
partitionIdleTime);
};
+ boolean trackNormalFiles =
table.coreOptions().dataEvolutionCompactionSplitLargeFiles();
+ boolean trackVectorFiles =
table.coreOptions().vectorCompactionEnabled();
Consumer<List<CommitMessage>> commitObserver = null;
- if (table.coreOptions().dataEvolutionCompactionSplitLargeFiles()) {
+ if (trackNormalFiles || trackVectorFiles) {
commitObserver =
messages -> {
// Output sizes can still trigger splitting or
small-file merging.
- // Skip bins containing only this invocation's
completed normal outputs.
+ // Skip bins and vector groups containing only this
invocation's outputs.
for (CommitMessage message : messages) {
for (DataFileMeta file :
((CommitMessageImpl) message)
.compactIncrement()
.compactAfter()) {
- if (!isBlobFile(file.fileName())
- &&
!isVectorStoreFile(file.fileName())) {
+ if (isVectorStoreFile(file.fileName())) {
+ if (trackVectorFiles) {
+
completedVectorFiles.add(file.fileName());
+ }
+ } else if (trackNormalFiles &&
!isBlobFile(file.fileName())) {
completedNormalFiles.add(file.fileName());
}
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 8a64696ab2..9d0bfa751e 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -1930,6 +1930,73 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
}
}
+ test("Paimon Procedure: data evolution vector compaction does not recompact
rolled output") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, embedding ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'vector.file.format' = 'parquet',
+ | 'vector-field' = 'embedding',
+ | 'field.embedding.vector-dim' = '2',
+ | 'vector-compaction.enabled' = 'true',
+ | 'compaction.min.file-num' = '2')
+ |""".stripMargin)
+ for (start <- Seq(0, 1250)) {
+ sql(s"""
+ |INSERT INTO T SELECT /*+ REPARTITION(1) */
+ |CAST(id AS INT), array(CAST(id AS FLOAT), 0.0f) FROM
range($start, ${start + 1250})
+ |""".stripMargin)
+ }
+ // Non-overlapping files are only rewritten when small; overlap them
with an update.
+ sql("""
+ |MERGE INTO T USING (SELECT CAST(id AS INT) AS id FROM range(0,
1250)) S
+ |ON T.id = S.id
+ |WHEN MATCHED THEN UPDATE SET T.embedding = array(CAST(S.id AS
FLOAT), 0.0f)
+ |""".stripMargin)
+ sql("ALTER TABLE T SET TBLPROPERTIES ('vector.target-file-size' = '1
b')")
+
+ def vectorRanges(): Seq[Range] = loadTable("T")
+ .newSnapshotReader()
+ .read()
+ .dataSplits()
+ .asScala
+ .flatMap(_.dataFiles().asScala)
+ .filter(_.fileName().contains(".vector."))
+ .map(_.nonNullRowIdRange())
+ .sortBy(_.from)
+ .toSeq
+ assert(
+ vectorRanges() == Seq(new Range(0, 1249), new Range(0, 1249), new
Range(1250, 2499)),
+ vectorRanges())
+
+ val table = loadTable("T")
+ val beforeSnapshot = lastSnapshotId(table)
+ val attempts = new AtomicInteger()
+ CompactProcedure.executeDataEvolutionCompaction(
+ table,
+ null,
+ null,
+ null,
+ new JavaSparkContext(spark.sparkContext),
+ spark,
+ null,
+ _ => assert(attempts.incrementAndGet() <= 1, "Recompacted this
invocation's output")
+ )
+
+ assert(lastSnapshotId(table) == beforeSnapshot + 1)
+ // The rolling writer checks the target size every 1000 rows.
+ assert(
+ vectorRanges() == Seq(new Range(0, 999), new Range(1000, 1999), new
Range(2000, 2499)),
+ vectorRanges())
+ checkAnswer(
+ sql("SELECT id, embedding[0] FROM T ORDER BY id"),
+ (0 until 2500).map(id => Row(id, id.toFloat)))
+ }
+ }
+
test("Paimon Procedure: materialize deletion vectors across planner
batches") {
withTable("T") {
sql("""
