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 c8a87e9cbd [core] Build source-backed sorted index payloads (#8600)
c8a87e9cbd is described below
commit c8a87e9cbdd948fd555d3ade6f7ae21b58b77c4d
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 13 20:19:01 2026 +0800
[core] Build source-backed sorted index payloads (#8600)
Provide the reusable core primitives needed to build BTree and Bitmap
global-index payloads from primary-key table data files.
---
.../sorted/SortedGlobalIndexBuilder.java | 24 +-
.../sorted/SortedSingleColumnIndexWriter.java | 87 ++++++
.../index/pksorted/PkSortedDataFileReader.java | 138 ++++++++++
.../index/pksorted/PkSortedIndexBuilder.java | 179 +++++++++++++
.../paimon/index/pksorted/PkSortedIndexFile.java | 202 ++++++++++++++
.../paimon/index/pksorted/PkSortedIndexGroup.java | 79 ++++++
.../sorted/SortedGlobalIndexBuilderTest.java | 36 +++
.../index/pksorted/PkSortedDataFileReaderTest.java | 161 +++++++++++
.../index/pksorted/PkSortedIndexBuilderTest.java | 294 +++++++++++++++++++++
.../index/pksorted/PkSortedIndexFileTest.java | 186 +++++++++++++
10 files changed, 1368 insertions(+), 18 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
index 6ee00b3bb8..6173c070f9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
@@ -244,31 +244,19 @@ public class SortedGlobalIndexBuilder implements
Serializable {
public List<CommitMessage> buildForSinglePartition(
Range rowRange, BinaryRow partition, Iterator<InternalRow> data)
throws IOException {
- long counter = 0;
- GlobalIndexSingleColumnWriter currentWriter = null;
- List<CommitMessage> commitMessages = new ArrayList<>();
FieldGetter indexFieldGetter =
InternalRow.createFieldGetter(indexField.type(), 0);
+ SortedSingleColumnIndexWriter writer =
+ new SortedSingleColumnIndexWriter(recordsPerRange,
this::createWriter);
while (data.hasNext()) {
InternalRow row = data.next();
-
- if (currentWriter != null && counter >= recordsPerRange) {
- commitMessages.add(flushIndex(rowRange,
currentWriter.finish(), partition));
- currentWriter = null;
- counter = 0;
- }
-
- counter++;
- if (currentWriter == null) {
- currentWriter = createWriter();
- }
-
long localRowId = row.getLong(1) - rowRange.from;
- currentWriter.write(indexFieldGetter.getFieldOrNull(row),
localRowId);
+ writer.write(indexFieldGetter.getFieldOrNull(row), localRowId);
}
- if (counter > 0) {
- commitMessages.add(flushIndex(rowRange, currentWriter.finish(),
partition));
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (List<ResultEntry> resultEntries : writer.finish()) {
+ commitMessages.add(flushIndex(rowRange, resultEntries, partition));
}
return commitMessages;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
new file mode 100644
index 0000000000..6bc37a0f7d
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
@@ -0,0 +1,87 @@
+/*
+ * 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.globalindex.sorted;
+
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Rotates single-column index writers after a configured number of logical
records. */
+public final class SortedSingleColumnIndexWriter {
+
+ private final long recordsPerRange;
+ private final Factory factory;
+ private final List<List<ResultEntry>> resultGroups = new ArrayList<>();
+
+ @Nullable private GlobalIndexSingleColumnWriter currentWriter;
+ private long currentRecordCount;
+ private boolean finished;
+
+ public SortedSingleColumnIndexWriter(long recordsPerRange, Factory
factory) {
+ checkArgument(recordsPerRange > 0, "Records per range must be
positive.");
+ this.recordsPerRange = recordsPerRange;
+ this.factory = factory;
+ }
+
+ public void write(@Nullable Object value, long localRowId) throws
IOException {
+ checkState(!finished, "Cannot write after the sorted index writer is
finished.");
+ if (currentWriter != null && currentRecordCount >= recordsPerRange) {
+ finishCurrentWriter();
+ }
+ if (currentWriter == null) {
+ currentWriter = factory.create();
+ }
+ currentWriter.write(value, localRowId);
+ currentRecordCount++;
+ }
+
+ public List<List<ResultEntry>> finish() {
+ checkState(!finished, "Sorted index writer is already finished.");
+ if (currentWriter != null) {
+ finishCurrentWriter();
+ }
+ finished = true;
+ List<List<ResultEntry>> copy = new ArrayList<>(resultGroups.size());
+ for (List<ResultEntry> group : resultGroups) {
+ copy.add(Collections.unmodifiableList(new ArrayList<>(group)));
+ }
+ return Collections.unmodifiableList(copy);
+ }
+
+ private void finishCurrentWriter() {
+ resultGroups.add(currentWriter.finish());
+ currentWriter = null;
+ currentRecordCount = 0;
+ }
+
+ /** Creates one underlying algorithm writer. */
+ @FunctionalInterface
+ public interface Factory {
+ GlobalIndexSingleColumnWriter create() throws IOException;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedDataFileReader.java
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedDataFileReader.java
new file mode 100644
index 0000000000..2819961ecc
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedDataFileReader.java
@@ -0,0 +1,138 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Reads one scalar value from every physical record in a data file. */
+public final class PkSortedDataFileReader implements Closeable,
PkSortedIndexBuilder.Reader {
+
+ private final DataFileMeta dataFile;
+ private final RecordReaderIterator<KeyValue> iterator;
+ private final InternalRow.FieldGetter fieldGetter;
+ private long rowsRead;
+
+ public PkSortedDataFileReader(
+ KeyValueFileReaderFactory readerFactory, DataFileMeta dataFile,
DataField indexField)
+ throws IOException {
+ this.dataFile = dataFile;
+ this.iterator = new
RecordReaderIterator<>(readerFactory.createRecordReader(dataFile));
+ this.fieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
+ }
+
+ public long rowCount() {
+ return dataFile.rowCount();
+ }
+
+ @Nullable
+ public Entry readNext() {
+ if (rowsRead == rowCount()) {
+ checkArgument(
+ !iterator.hasNext(),
+ "Data file %s contains more rows than declared.",
+ dataFile.fileName());
+ return null;
+ }
+ checkArgument(
+ iterator.hasNext(),
+ "Data file %s ended before its declared row count.",
+ dataFile.fileName());
+ KeyValue keyValue = iterator.next();
+ long rowPosition = rowsRead++;
+ if (rowsRead == rowCount()) {
+ checkArgument(
+ !iterator.hasNext(),
+ "Data file %s contains more rows than declared.",
+ dataFile.fileName());
+ }
+ return new Entry(fieldGetter.getFieldOrNull(keyValue.value()),
rowPosition);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ iterator.close();
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException(
+ "Failed to close sorted index reader for " +
dataFile.fileName(), e);
+ }
+ }
+
+ /** One projected value and its zero-based physical position. */
+ public static final class Entry {
+
+ @Nullable private final Object value;
+ private final long rowPosition;
+
+ public Entry(@Nullable Object value, long rowPosition) {
+ this.value = value;
+ this.rowPosition = rowPosition;
+ }
+
+ @Nullable
+ public Object value() {
+ return value;
+ }
+
+ public long rowPosition() {
+ return rowPosition;
+ }
+ }
+
+ /** Bucket-scoped factory with scalar-only projection and no
deletion-vector filtering. */
+ public static final class Factory {
+
+ private final KeyValueFileReaderFactory readerFactory;
+ private final DataField indexField;
+
+ public Factory(
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder,
+ BinaryRow partition,
+ int bucket,
+ DataField indexField) {
+ this.readerFactory =
+ readerFactoryBuilder
+ .copyWithoutProjection()
+ .withReadKeyType(RowType.of())
+ .withReadValueType(RowType.of(indexField))
+ .buildWithoutDeletionVector(partition, bucket);
+ this.indexField = indexField;
+ }
+
+ public PkSortedDataFileReader create(DataFileMeta dataFile) throws
IOException {
+ return new PkSortedDataFileReader(readerFactory, dataFile,
indexField);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
new file mode 100644
index 0000000000..7a7604abe9
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
@@ -0,0 +1,179 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.sort.BinaryExternalSortBuffer;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.MutableObjectIteratorAdapter;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Sorts one physical data file and builds its source-backed scalar index
payloads. */
+public class PkSortedIndexBuilder {
+
+ private static final int LOCAL_ROW_ID_FIELD_ID = Integer.MAX_VALUE;
+
+ private final ReaderFactory readerFactory;
+ private final PkSortedIndexFile indexFile;
+ private final DataField indexField;
+ private final String indexType;
+ private final Options options;
+ @Nullable private final IOManager ioManager;
+
+ public PkSortedIndexBuilder(
+ PkSortedDataFileReader.Factory readerFactory,
+ PkSortedIndexFile indexFile,
+ DataField indexField,
+ String indexType,
+ Options options,
+ @Nullable IOManager ioManager) {
+ this(readerFactory::create, indexFile, indexField, indexType, options,
ioManager);
+ }
+
+ PkSortedIndexBuilder(
+ ReaderFactory readerFactory,
+ PkSortedIndexFile indexFile,
+ DataField indexField,
+ String indexType,
+ Options options,
+ @Nullable IOManager ioManager) {
+ this.readerFactory = readerFactory;
+ this.indexFile = indexFile;
+ this.indexField = indexField;
+ this.indexType = indexType;
+ this.options = options;
+ this.ioManager = ioManager;
+ }
+
+ public List<org.apache.paimon.index.IndexFileMeta> build(DataFileMeta
dataFile)
+ throws IOException {
+ IOManager actualIOManager = ioManager;
+ boolean ownsIOManager = false;
+ if (actualIOManager == null) {
+ actualIOManager = createTemporaryIOManager();
+ ownsIOManager = true;
+ }
+
+ BinaryExternalSortBuffer sortBuffer = null;
+ try {
+ CoreOptions coreOptions = new CoreOptions(options);
+ RowType sortRowType =
+ RowType.of(
+ indexField,
+ new DataField(
+ LOCAL_ROW_ID_FIELD_ID,
+ "_LOCAL_ROW_ID",
+ DataTypes.BIGINT().notNull()));
+ sortBuffer =
+ BinaryExternalSortBuffer.create(
+ actualIOManager,
+ sortRowType,
+ new int[] {0},
+ coreOptions.writeBufferSize(),
+ coreOptions.pageSize(),
+ coreOptions.localSortMaxNumFileHandles(),
+ coreOptions.spillCompressOptions(),
+ coreOptions.writeBufferSpillDiskSize());
+
+ try (Reader reader = readerFactory.create(dataFile)) {
+ checkArgument(
+ reader.rowCount() == dataFile.rowCount(),
+ "Sorted reader row count %s does not match data file
%s row count %s.",
+ reader.rowCount(),
+ dataFile.fileName(),
+ dataFile.rowCount());
+ PkSortedDataFileReader.Entry entry;
+ while ((entry = reader.readNext()) != null) {
+ sortBuffer.write(GenericRow.of(entry.value(),
entry.rowPosition()));
+ }
+ }
+
+ Iterator<InternalRow> sortedRows =
+ new MutableObjectIteratorAdapter<>(
+ sortBuffer.sortedIterator(),
+ new BinaryRow(sortRowType.getFieldCount()));
+ InternalRow.FieldGetter valueGetter =
+ InternalRow.createFieldGetter(indexField.type(), 0);
+ Iterator<PkSortedIndexFile.Entry> sortedEntries =
+ new Iterator<PkSortedIndexFile.Entry>() {
+ @Override
+ public boolean hasNext() {
+ return sortedRows.hasNext();
+ }
+
+ @Override
+ public PkSortedIndexFile.Entry next() {
+ InternalRow row = sortedRows.next();
+ return new PkSortedIndexFile.Entry(
+ valueGetter.getFieldOrNull(row),
row.getLong(1));
+ }
+ };
+ return indexFile.build(
+ new PrimaryKeyIndexSourceFile(dataFile.fileName(),
dataFile.rowCount()),
+ indexField,
+ indexType,
+ options,
+ sortedEntries);
+ } finally {
+ if (sortBuffer != null) {
+ sortBuffer.clear();
+ }
+ if (ownsIOManager) {
+ IOUtils.closeQuietly(actualIOManager);
+ }
+ }
+ }
+
+ protected IOManager createTemporaryIOManager() {
+ return IOManager.create(System.getProperty("java.io.tmpdir"));
+ }
+
+ /** Physical source reader used by the sorter. */
+ interface Reader extends Closeable {
+
+ long rowCount();
+
+ @Nullable
+ PkSortedDataFileReader.Entry readNext();
+ }
+
+ @FunctionalInterface
+ interface ReaderFactory {
+
+ Reader create(DataFileMeta dataFile) throws IOException;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
new file mode 100644
index 0000000000..f6ab090e48
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
@@ -0,0 +1,202 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedSingleColumnIndexWriter;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFile;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builds source-backed BTree or Bitmap payloads for one physical data file.
*/
+public class PkSortedIndexFile extends IndexFile {
+
+ public PkSortedIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
+ super(fileIO, pathFactory);
+ }
+
+ public List<IndexFileMeta> build(
+ PrimaryKeyIndexSourceFile sourceFile,
+ DataField indexField,
+ String indexType,
+ Options indexOptions,
+ Iterator<Entry> sortedEntries)
+ throws IOException {
+ checkArgument(
+ sourceFile.rowCount() > 0,
+ "A sorted index group must reference at least one source
row.");
+
+ TrackingFileWriter fileWriter = new TrackingFileWriter();
+ boolean success = false;
+ try {
+ long recordsPerRange =
+
indexOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
+ SortedSingleColumnIndexWriter writer =
+ new SortedSingleColumnIndexWriter(
+ recordsPerRange,
+ () -> createWriter(indexType, indexField,
indexOptions, fileWriter));
+
+ long writtenRows = 0;
+ while (sortedEntries.hasNext()) {
+ Entry entry = sortedEntries.next();
+ checkArgument(
+ entry.localRowId >= 0 && entry.localRowId <
sourceFile.rowCount(),
+ "Local row id %s is outside source file %s row range
[0, %s).",
+ entry.localRowId,
+ sourceFile.fileName(),
+ sourceFile.rowCount());
+ writer.write(entry.value, entry.localRowId);
+ writtenRows++;
+ }
+ checkArgument(
+ writtenRows == sourceFile.rowCount(),
+ "Sorted index input row count %s does not match source
file %s row count %s.",
+ writtenRows,
+ sourceFile.fileName(),
+ sourceFile.rowCount());
+
+ List<List<ResultEntry>> resultGroups = writer.finish();
+ List<IndexFileMeta> payloads = new ArrayList<>();
+ long payloadRows = 0;
+ byte[] sourceMeta = new
PrimaryKeyIndexSourceMeta(sourceFile).serialize();
+ for (List<ResultEntry> resultGroup : resultGroups) {
+ for (ResultEntry result : resultGroup) {
+ payloadRows = Math.addExact(payloadRows,
result.rowCount());
+ Path payloadPath = fileWriter.path(result.fileName());
+ payloads.add(
+ new IndexFileMeta(
+ indexType,
+ result.fileName(),
+ fileIO.getFileSize(payloadPath),
+ result.rowCount(),
+ new GlobalIndexMeta(
+ 0,
+ sourceFile.rowCount() - 1,
+ indexField.id(),
+ null,
+ result.meta(),
+ sourceMeta),
+ pathFactory.isExternalPath() ?
payloadPath.toString() : null));
+ }
+ }
+ checkArgument(
+ payloadRows == sourceFile.rowCount(),
+ "Sorted payload row count %s does not match source file %s
row count %s.",
+ payloadRows,
+ sourceFile.fileName(),
+ sourceFile.rowCount());
+ success = true;
+ return Collections.unmodifiableList(payloads);
+ } finally {
+ if (!success) {
+ fileWriter.deleteCreatedFiles();
+ }
+ }
+ }
+
+ protected GlobalIndexSingleColumnWriter createWriter(
+ String indexType,
+ DataField indexField,
+ Options indexOptions,
+ GlobalIndexFileWriter fileWriter)
+ throws IOException {
+ GlobalIndexer indexer = GlobalIndexer.create(indexType, indexField,
indexOptions);
+ GlobalIndexWriter writer = indexer.createWriter(fileWriter);
+ checkArgument(
+ writer instanceof GlobalIndexSingleColumnWriter,
+ "Index algorithm %s does not create a single-column writer.",
+ indexType);
+ return (GlobalIndexSingleColumnWriter) writer;
+ }
+
+ /** One sorted scalar value and its zero-based position in the source data
file. */
+ public static final class Entry {
+
+ @Nullable private final Object value;
+ private final long localRowId;
+
+ public Entry(@Nullable Object value, long localRowId) {
+ this.value = value;
+ this.localRowId = localRowId;
+ }
+
+ @Nullable
+ public Object value() {
+ return value;
+ }
+
+ public long localRowId() {
+ return localRowId;
+ }
+ }
+
+ private final class TrackingFileWriter implements GlobalIndexFileWriter {
+
+ private final Map<String, Path> createdFiles = new LinkedHashMap<>();
+
+ @Override
+ public String newFileName(String prefix) {
+ Path path = pathFactory.newPath();
+ createdFiles.put(path.getName(), path);
+ return path.getName();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) throws
IOException {
+ return fileIO.newOutputStream(path(fileName), false);
+ }
+
+ private Path path(String fileName) {
+ Path path = createdFiles.get(fileName);
+ checkArgument(path != null, "Sorted payload file %s was not
allocated.", fileName);
+ return path;
+ }
+
+ private void deleteCreatedFiles() {
+ for (Path path : createdFiles.values()) {
+ fileIO.deleteQuietly(path);
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
new file mode 100644
index 0000000000..46f161eadb
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
@@ -0,0 +1,79 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+/** All rotated payloads that index one physical source data file. */
+public final class PkSortedIndexGroup {
+
+ private final PrimaryKeyIndexSourceFile sourceFile;
+ private final List<IndexFileMeta> payloads;
+
+ PkSortedIndexGroup(PrimaryKeyIndexSourceFile sourceFile,
List<IndexFileMeta> payloads) {
+ this.sourceFile = sourceFile;
+ this.payloads = Collections.unmodifiableList(new
ArrayList<>(payloads));
+ }
+
+ static Optional<PkSortedIndexGroup> create(
+ int fieldId,
+ String indexType,
+ PrimaryKeyIndexSourceFile sourceFile,
+ List<IndexFileMeta> payloads) {
+ long payloadRowCount = 0;
+ Set<String> payloadNames = new HashSet<>();
+ for (IndexFileMeta payload : payloads) {
+ GlobalIndexMeta meta = payload.globalIndexMeta();
+ PrimaryKeyIndexSourceFile payloadSource =
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile();
+ if (!payloadNames.add(payload.fileName())
+ || !sourceFile.equals(payloadSource)
+ || !indexType.equals(payload.indexType())
+ || meta == null
+ || meta.indexFieldId() != fieldId
+ || meta.rowRangeStart() != 0
+ || meta.rowRangeEnd() != sourceFile.rowCount() - 1) {
+ return Optional.empty();
+ }
+ payloadRowCount += payload.rowCount();
+ }
+ if (payloadRowCount != sourceFile.rowCount()) {
+ return Optional.empty();
+ }
+ return Optional.of(new PkSortedIndexGroup(sourceFile, payloads));
+ }
+
+ public PrimaryKeyIndexSourceFile sourceFile() {
+ return sourceFile;
+ }
+
+ public List<IndexFileMeta> payloads() {
+ return payloads;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
index 7e0d18c2c9..cb3e55d1a2 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
@@ -24,7 +24,9 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
@@ -50,13 +52,21 @@ import org.apache.paimon.utils.Pair;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayDeque;
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;
import java.util.Optional;
+import java.util.Queue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Test class for {@link SortedGlobalIndexBuilder}. */
public class SortedGlobalIndexBuilderTest extends TableTestBase {
@@ -65,6 +75,32 @@ public class SortedGlobalIndexBuilderTest extends
TableTestBase {
private static final KeySerializer KEY_SERIALIZER =
KeySerializer.create(DataTypes.INT());
private static final Comparator<Object> COMPARATOR =
KEY_SERIALIZER.createComparator();
+ @Test
+ public void testSingleColumnWriterRotationPreservesResultGroups() throws
Exception {
+ GlobalIndexSingleColumnWriter first =
mock(GlobalIndexSingleColumnWriter.class);
+ GlobalIndexSingleColumnWriter second =
mock(GlobalIndexSingleColumnWriter.class);
+ when(first.finish())
+ .thenReturn(Collections.singletonList(new
ResultEntry("index-1", 2, null)));
+ when(second.finish())
+ .thenReturn(Collections.singletonList(new
ResultEntry("index-2", 1, null)));
+ Queue<GlobalIndexSingleColumnWriter> writers =
+ new ArrayDeque<>(Arrays.asList(first, second));
+ SortedSingleColumnIndexWriter rotatingWriter =
+ new SortedSingleColumnIndexWriter(2, writers::remove);
+
+ rotatingWriter.write(10, 0);
+ rotatingWriter.write(20, 1);
+ rotatingWriter.write(30, 2);
+ List<List<ResultEntry>> results = rotatingWriter.finish();
+
+ verify(first).write(10, 0);
+ verify(first).write(20, 1);
+ verify(second).write(30, 2);
+ assertThat(results).hasSize(2);
+
assertThat(results.get(0)).extracting(ResultEntry::fileName).containsExactly("index-1");
+
assertThat(results.get(1)).extracting(ResultEntry::fileName).containsExactly("index-2");
+ }
+
@Override
public Schema schemaDefault() {
Schema.Builder schemaBuilder = Schema.newBuilder();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedDataFileReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedDataFileReaderTest.java
new file mode 100644
index 0000000000..424689c2e8
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedDataFileReaderTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.format.FlushingFileFormat;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
+import org.apache.paimon.io.KeyValueFileWriterFactory;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.KeyValueFieldsExtractor;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.options.MemorySize.VALUE_128_MB;
+import static
org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests reading scalar values with their physical positions. */
+class PkSortedDataFileReaderTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testReadsNullWithItsPhysicalPosition() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ DataField keyField = new DataField(0, "id", DataTypes.INT());
+ DataField payloadField = new DataField(1, "payload",
DataTypes.STRING());
+ RowType keyType = RowType.of(keyField);
+ RowType valueType = RowType.of(payloadField);
+ List<DataField> fields = Arrays.asList(keyField, payloadField);
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ fields,
+ 1,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyMap(),
+ "");
+ SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
+ assertThat(schemaManager.commit(schema)).isTrue();
+ FileStorePathFactory pathFactory = createNonPartFactory(tablePath);
+ FlushingFileFormat format = new FlushingFileFormat("avro");
+ CoreOptions options = new CoreOptions(new Options());
+
+ KeyValueFileWriterFactory writerFactory =
+ KeyValueFileWriterFactory.builder(
+ fileIO,
+ schema.id(),
+ keyType,
+ valueType,
+ format,
+ ignored -> pathFactory,
+ VALUE_128_MB.getBytes())
+ .build(BinaryRow.EMPTY_ROW, 0, options);
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ writerFactory.createRollingMergeTreeFileWriter(1,
FileSource.COMPACT);
+ try {
+ writer.write(keyValue(1, "a"));
+ writer.write(keyValue(2, null));
+ writer.write(keyValue(3, "c"));
+ } finally {
+ writer.close();
+ }
+ DataFileMeta dataFile = writer.result().get(0);
+
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder =
+ KeyValueFileReaderFactory.builder(
+ fileIO,
+ schemaManager,
+ schema,
+ keyType,
+ valueType,
+ ignored -> format,
+ pathFactory,
+ fieldsExtractor(keyType, valueType),
+ options);
+ PkSortedDataFileReader reader =
+ new PkSortedDataFileReader.Factory(
+ readerFactoryBuilder, BinaryRow.EMPTY_ROW, 0,
payloadField)
+ .create(dataFile);
+ try {
+ assertEntry(reader.readNext(), 0, "a");
+ assertEntry(reader.readNext(), 1, null);
+ assertEntry(reader.readNext(), 2, "c");
+ assertThat(reader.readNext()).isNull();
+ } finally {
+ reader.close();
+ }
+ }
+
+ private static void assertEntry(
+ PkSortedDataFileReader.Entry entry, long position, String value) {
+ assertThat(entry.rowPosition()).isEqualTo(position);
+ if (value == null) {
+ assertThat(entry.value()).isNull();
+ } else {
+
assertThat(entry.value()).isEqualTo(BinaryString.fromString(value));
+ }
+ }
+
+ private static KeyValue keyValue(int key, String payload) {
+ return new KeyValue()
+ .replace(
+ GenericRow.of(key),
+ RowKind.INSERT,
+ GenericRow.of(payload == null ? null :
BinaryString.fromString(payload)));
+ }
+
+ private static KeyValueFieldsExtractor fieldsExtractor(RowType keyType,
RowType valueType) {
+ return new KeyValueFieldsExtractor() {
+ @Override
+ public List<DataField> keyFields(TableSchema schema) {
+ return keyType.getFields();
+ }
+
+ @Override
+ public List<DataField> valueFields(TableSchema schema) {
+ return valueType.getFields();
+ }
+ };
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
new file mode 100644
index 0000000000..1c14998503
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
@@ -0,0 +1,294 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.disk.BufferFileWriter;
+import org.apache.paimon.disk.FileIOChannel;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexFileReadWrite;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests physical source sorting and payload construction. */
+class PkSortedIndexBuilderTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testBuildsQueryableBTreeAndBitmapFromUnsortedPhysicalRows() throws
Exception {
+ List<PkSortedDataFileReader.Entry> entries =
+ Arrays.asList(entry(20, 0), entry(null, 1), entry(10, 2),
entry(20, 3));
+ DataFileMeta source = dataFile("data-file", entries.size());
+
+ for (String indexType : Arrays.asList("btree", "bitmap")) {
+ java.nio.file.Path directory =
FilesHelper.createDirectory(tempPath, indexType);
+ LocalFileIO fileIO = LocalFileIO.create();
+ IndexPathFactory pathFactory = pathFactory(directory);
+ Options options = options();
+ IOManager ioManager =
IOManager.create(directory.resolve("spill").toString());
+ List<IndexFileMeta> payloads;
+ try {
+ payloads =
+ new PkSortedIndexBuilder(
+ ignored -> new ArrayReader(entries),
+ new PkSortedIndexFile(fileIO,
pathFactory),
+ field(),
+ indexType,
+ options,
+ ioManager)
+ .build(source);
+ } finally {
+ ioManager.close();
+ }
+
+ assertThat(payloads).hasSize(2);
+ assertQuery(fileIO, pathFactory, indexType, options, payloads,
false, 20, 0L, 3L);
+ assertQuery(fileIO, pathFactory, indexType, options, payloads,
true, null, 1L);
+ }
+ }
+
+ @Test
+ void testForcedSpillSortsRowsAndClosesTaskOwnedIoManager() throws
Exception {
+ int rowCount = 20_000;
+ List<PkSortedDataFileReader.Entry> entries = new ArrayList<>(rowCount);
+ for (int position = 0; position < rowCount; position++) {
+ entries.add(entry(rowCount - position - 1, position));
+ }
+ List<Integer> sortedValues = new ArrayList<>(rowCount);
+ TrackingIOManager ioManager =
+ new
TrackingIOManager(tempPath.resolve("owned-spill").toString());
+ PkSortedIndexFile capturingFile =
+ new PkSortedIndexFile(LocalFileIO.create(),
pathFactory(tempPath)) {
+ @Override
+ public List<IndexFileMeta> build(
+ PrimaryKeyIndexSourceFile sourceFile,
+ DataField indexField,
+ String indexType,
+ Options indexOptions,
+ Iterator<Entry> sortedEntries) {
+ while (sortedEntries.hasNext()) {
+ sortedValues.add((Integer)
sortedEntries.next().value());
+ }
+ return Collections.emptyList();
+ }
+ };
+ Options options = options();
+ options.set(
+ CoreOptions.WRITE_BUFFER_SIZE,
+ org.apache.paimon.options.MemorySize.parse("128 kb"));
+ options.set(CoreOptions.PAGE_SIZE,
org.apache.paimon.options.MemorySize.parse("32 kb"));
+
+ new PkSortedIndexBuilder(
+ ignored -> new ArrayReader(entries),
+ capturingFile,
+ field(),
+ "btree",
+ options,
+ null) {
+ @Override
+ protected IOManager createTemporaryIOManager() {
+ return ioManager;
+ }
+ }.build(dataFile("large-data-file", rowCount));
+
+ assertThat(sortedValues)
+ .containsExactlyElementsOf(
+ LongStream.range(0, rowCount)
+ .mapToObj(value -> (int) value)
+ .collect(Collectors.toList()));
+ assertThat(ioManager.createdSpillWriters).isGreaterThan(0);
+ assertThat(ioManager.closed).isTrue();
+ }
+
+ private static void assertQuery(
+ LocalFileIO fileIO,
+ IndexPathFactory pathFactory,
+ String indexType,
+ Options options,
+ List<IndexFileMeta> payloads,
+ boolean isNull,
+ Object literal,
+ Long... expected)
+ throws Exception {
+ List<GlobalIndexIOMeta> ioMetas = new ArrayList<>();
+ for (IndexFileMeta payload : payloads) {
+ ioMetas.add(
+ new GlobalIndexIOMeta(
+ pathFactory.toPath(payload.fileName()),
+ payload.fileSize(),
+ payload.globalIndexMeta().indexMeta()));
+ }
+ ExecutorService executor = newDirectExecutorService();
+ try (GlobalIndexReader reader =
+ GlobalIndexer.create(indexType, field(), options)
+ .createReader(
+ new GlobalIndexFileReadWrite(fileIO,
pathFactory),
+ ioMetas,
+ executor)) {
+ FieldRef fieldRef = new FieldRef(7, "indexed", DataTypes.INT());
+ GlobalIndexResult result =
+ (isNull ? reader.visitIsNull(fieldRef) :
reader.visitEqual(fieldRef, literal))
+ .join()
+ .get();
+ assertThat(result.results()).containsExactlyInAnyOrder(expected);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private static PkSortedDataFileReader.Entry entry(Object value, long
position) {
+ return new PkSortedDataFileReader.Entry(value, position);
+ }
+
+ private static DataField field() {
+ return new DataField(7, "indexed", DataTypes.INT());
+ }
+
+ private static Options options() {
+ Options options = new Options();
+ options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 2L);
+ return options;
+ }
+
+ private static DataFileMeta dataFile(String fileName, long rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static IndexPathFactory pathFactory(java.nio.file.Path directory) {
+ Path path = new Path(directory.toUri());
+ return new IndexPathFactory() {
+ @Override
+ public Path toPath(String fileName) {
+ return new Path(path, fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ return new Path(path, UUID.randomUUID().toString());
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+ };
+ }
+
+ private static final class ArrayReader implements
PkSortedIndexBuilder.Reader {
+
+ private final List<PkSortedDataFileReader.Entry> entries;
+ private int position;
+
+ private ArrayReader(List<PkSortedDataFileReader.Entry> entries) {
+ this.entries = entries;
+ }
+
+ @Override
+ public long rowCount() {
+ return entries.size();
+ }
+
+ @Override
+ public PkSortedDataFileReader.Entry readNext() {
+ return position == entries.size() ? null : entries.get(position++);
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static final class TrackingIOManager extends IOManagerImpl {
+
+ private int createdSpillWriters;
+ private boolean closed;
+
+ private TrackingIOManager(String tempDir) {
+ super(tempDir);
+ }
+
+ @Override
+ public BufferFileWriter createBufferFileWriter(FileIOChannel.ID
channelID)
+ throws IOException {
+ createdSpillWriters++;
+ return super.createBufferFileWriter(channelID);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed = true;
+ super.close();
+ }
+ }
+
+ private static final class FilesHelper {
+
+ private static java.nio.file.Path createDirectory(java.nio.file.Path
parent, String child)
+ throws IOException {
+ return java.nio.file.Files.createDirectory(parent.resolve(child));
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
new file mode 100644
index 0000000000..adc8ba93d2
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
@@ -0,0 +1,186 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests source-backed sorted index payload creation. */
+class PkSortedIndexFileTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testBuildsRotatedBTreeAndBitmapPayloadGroups() throws Exception {
+ for (String indexType : Arrays.asList("btree", "bitmap")) {
+ java.nio.file.Path indexDirectory =
Files.createDirectory(tempPath.resolve(indexType));
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(LocalFileIO.create(),
pathFactory(indexDirectory));
+ PrimaryKeyIndexSourceFile source = new
PrimaryKeyIndexSourceFile("data-file", 3);
+
+ List<IndexFileMeta> payloads =
+ indexFile.build(
+ source,
+ field(),
+ indexType,
+ options(),
+ Arrays.asList(
+ new PkSortedIndexFile.Entry(null,
1),
+ new PkSortedIndexFile.Entry(10, 2),
+ new PkSortedIndexFile.Entry(20, 0))
+ .iterator());
+
+ assertThat(payloads).hasSize(2);
+
assertThat(payloads).extracting(IndexFileMeta::indexType).containsOnly(indexType);
+
assertThat(payloads).extracting(IndexFileMeta::rowCount).containsExactly(2L,
1L);
+ assertThat(payloads)
+ .allSatisfy(
+ payload -> {
+ GlobalIndexMeta meta =
payload.globalIndexMeta();
+ assertThat(meta.rowRangeStart()).isZero();
+ assertThat(meta.rowRangeEnd()).isEqualTo(2);
+ assertThat(meta.indexFieldId()).isEqualTo(7);
+ assertThat(meta.indexMeta()).isNotEmpty();
+ assertThat(
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
+ .sourceFile())
+ .isEqualTo(source);
+ assertThat(indexFile.exists(payload)).isTrue();
+ });
+ }
+ }
+
+ @Test
+ void testFailedSecondPayloadDeletesWholeGroup() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ AtomicInteger writerNumber = new AtomicInteger();
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(fileIO, pathFactory(tempPath)) {
+ @Override
+ protected GlobalIndexSingleColumnWriter createWriter(
+ String indexType,
+ DataField indexField,
+ Options indexOptions,
+ GlobalIndexFileWriter fileWriter) {
+ int currentWriter = writerNumber.incrementAndGet();
+ return new GlobalIndexSingleColumnWriter() {
+ private long rowCount;
+
+ @Override
+ public void write(Object key, long relativeRowId) {
+ if (currentWriter == 2) {
+ throw new RuntimeException("injected
second payload failure");
+ }
+ rowCount++;
+ }
+
+ @Override
+ public List<ResultEntry> finish() {
+ String fileName =
fileWriter.newFileName("test");
+ try (PositionOutputStream output =
+ fileWriter.newOutputStream(fileName)) {
+ output.write(new byte[] {1});
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return Collections.singletonList(
+ new ResultEntry(fileName, rowCount,
new byte[] {2}));
+ }
+ };
+ }
+ };
+ Options options = options();
+ options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 1L);
+
+ assertThatThrownBy(
+ () ->
+ indexFile.build(
+ new
PrimaryKeyIndexSourceFile("data-file", 2),
+ field(),
+ "btree",
+ options,
+ Arrays.asList(
+ new
PkSortedIndexFile.Entry(10, 0),
+ new
PkSortedIndexFile.Entry(20, 1))
+ .iterator()))
+ .hasMessageContaining("injected second payload failure");
+
+ try (Stream<java.nio.file.Path> files = Files.list(tempPath)) {
+ assertThat(files).isEmpty();
+ }
+ }
+
+ private static DataField field() {
+ return new DataField(7, "indexed", DataTypes.INT());
+ }
+
+ private static Options options() {
+ Options options = new Options();
+ options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 2L);
+ return options;
+ }
+
+ private static IndexPathFactory pathFactory(java.nio.file.Path directory) {
+ Path path = new Path(directory.toUri());
+ return new IndexPathFactory() {
+ @Override
+ public Path toPath(String fileName) {
+ return new Path(path, fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ return new Path(path, UUID.randomUUID().toString());
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+ };
+ }
+}
