Skip to content

[core][vector] add log for lumina index writer #7512

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Mar 26, 2026
Merged

[core][vector] add log for lumina index writer #7512

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 41 additions & 3 deletions ...on-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,23 +196,48 @@ public List<ResultEntry> finish() {

private ResultEntry buildIndex() throws IOException {
configureExecutorThreadCount();
LOG.info(
"Lumina index build started: {} vectors, dim={}, type={}, metric={}",
count,
dim,
options.indexType(),
options.metric());
long buildStart = System.currentTimeMillis();

try (LuminaIndex index =
LuminaIndex.createForBuild(
options.indexType(), dim, options.metric(), options.toLuminaOptions())) {

// Pretrain and insert via streaming file-backed Dataset API
try (FileBackedDataset ds = new FileBackedDataset(tempVectorFile, dim, count)) {
long phaseStart = System.currentTimeMillis();
LOG.info("Lumina pretrain phase started");
try (FileBackedDataset ds =
new FileBackedDataset(tempVectorFile, dim, count, "pretrain")) {
index.pretrainFrom(ds);
}
try (FileBackedDataset ds = new FileBackedDataset(tempVectorFile, dim, count)) {
LOG.info(
"Lumina pretrain phase done in {} ms", System.currentTimeMillis() - phaseStart);

phaseStart = System.currentTimeMillis();
LOG.info("Lumina insert phase started");
try (FileBackedDataset ds =
new FileBackedDataset(tempVectorFile, dim, count, "insert")) {
index.insertFrom(ds);
}
LOG.info("Lumina insert phase done in {} ms", System.currentTimeMillis() - phaseStart);

phaseStart = System.currentTimeMillis();
LOG.info("Lumina dump phase started");
String fileName = fileWriter.newFileName(FILE_NAME_PREFIX);
try (PositionOutputStream out = fileWriter.newOutputStream(fileName)) {
index.dump(new OutputStreamFileOutput(out));
out.flush();
}
LOG.info("Lumina dump phase done in {} ms", System.currentTimeMillis() - phaseStart);

LOG.info(
"Lumina index build completed in {} ms",
System.currentTimeMillis() - buildStart);

LuminaIndexMeta meta = new LuminaIndexMeta(options.toLuminaOptions());
return new ResultEntry(fileName, count, meta.serialize());
Expand Down Expand Up @@ -308,8 +333,10 @@ static class FileBackedDataset implements LuminaDataset, Closeable {
private final int totalCount;
private int cursor;
private final ByteBuffer readBuf;
private final String phase;
private int lastLoggedPercent;

FileBackedDataset(File file, int dim, int totalCount) throws IOException {
FileBackedDataset(File file, int dim, int totalCount, String phase) throws IOException {
this.raf = new RandomAccessFile(file, "r");
this.channel = raf.getChannel();
this.dim = dim;
Expand All @@ -318,6 +345,8 @@ static class FileBackedDataset implements LuminaDataset, Closeable {
this.readBuf = ByteBuffer.allocateDirect(IO_BUFFER_SIZE);
this.readBuf.order(ByteOrder.nativeOrder());
this.readBuf.limit(0); // empty initially
this.phase = phase;
this.lastLoggedPercent = -1;
}

@Override
Expand Down Expand Up @@ -370,6 +399,15 @@ public long getNextBatch(float[] vectorBuf, long[] idBuf) {
idBuf[i] = cursor + i;
}
cursor += batchSize;

int percent = (int) ((long) cursor * 100 / totalCount);
if (percent / 10 > lastLoggedPercent / 10) {
LOG.info(
"Lumina {} progress: {}/{} vectors ({}%)",
phase, cursor, totalCount, percent);
lastLoggedPercent = percent;
}

return batchSize;
}

Expand Down