This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 6bc31fa8830db76ff91b1af220740e2ba3140142 Author: Eunbin Son <[email protected]> AuthorDate: Sat Sep 19 23:03:09 2026 +0900 [eslib] Do not require build-time es-index options when reading persisted indexes (#9969) --- .../eslib/index/ESIndexGlobalIndexReader.java | 28 +++-- .../paimon/eslib/index/ESIndexGlobalIndexer.java | 44 +++++--- .../apache/paimon/eslib/index/ESIndexOptions.java | 21 +++- .../eslib/index/ESIndexVectorMetricTest.java | 116 +++++++++++++++++++++ 4 files changed, 186 insertions(+), 23 deletions(-) diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java index 8e5bcd8ab7..4d0093abb1 100644 --- a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java +++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java @@ -52,6 +52,8 @@ import org.elasticsearch.eslib.api.model.SearchResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -121,15 +123,22 @@ public class ESIndexGlobalIndexReader implements GlobalIndexReader { GlobalIndexFileReader fileReader, List<GlobalIndexIOMeta> files, List<DataField> fields, - ESIndexOptions indexOptions) { + @Nullable ESIndexOptions indexOptions) { this(fileReader, files, fields, indexOptions, null); } + /** + * Creates a reader for one es-index shard. + * + * @param indexOptions the es-index options parsed from the current table options; only used as + * a fallback for legacy metadata that does not persist the field configuration, so it may + * be {@code null} when the current options cannot describe a build + */ public ESIndexGlobalIndexReader( GlobalIndexFileReader fileReader, List<GlobalIndexIOMeta> files, List<DataField> fields, - ESIndexOptions indexOptions, + @Nullable ESIndexOptions indexOptions, ExecutorService queryExecutor) { Objects.requireNonNull(files, "files"); checkArgument(files.size() == 1, "Expected exactly one ES index file per shard"); @@ -138,7 +147,6 @@ public class ESIndexGlobalIndexReader implements GlobalIndexReader { this.fields = Collections.unmodifiableList( new ArrayList<>(Objects.requireNonNull(fields, "fields"))); - ESIndexOptions fallbackIndexOptions = Objects.requireNonNull(indexOptions, "indexOptions"); try { GlobalIndexIOMeta archiveMeta = Objects.requireNonNull(this.files.get(0), "index file metadata"); @@ -146,10 +154,16 @@ public class ESIndexGlobalIndexReader implements GlobalIndexReader { this.fileOffsets = parsed.fileOffsets(); validateArchiveOffsets(archiveMeta, fileOffsets); this.hasPersistedFieldConfigs = parsed.hasFieldConfigs(); - this.indexOptions = - parsed.hasFieldConfigs() - ? ESIndexOptions.fromFieldConfigs(parsed.fieldConfigs()) - : fallbackIndexOptions; + if (parsed.hasFieldConfigs()) { + this.indexOptions = ESIndexOptions.fromFieldConfigs(parsed.fieldConfigs()); + } else { + checkArgument( + indexOptions != null, + "Legacy es-index metadata does not persist the field configuration, so " + + "the current es-index options must describe how the index was " + + "built."); + this.indexOptions = indexOptions; + } List<String> physicalFields = parsed.hasFieldConfigs() ? parsed.indexedFieldNames() diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java index 3ae8620624..287505204c 100644 --- a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java +++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java @@ -27,7 +27,7 @@ import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.options.Options; import org.apache.paimon.types.DataField; -import org.elasticsearch.eslib.api.model.FieldIndexConfig; +import javax.annotation.Nullable; import java.io.IOException; import java.util.ArrayList; @@ -42,19 +42,43 @@ import java.util.concurrent.ExecutorService; public class ESIndexGlobalIndexer implements VectorGlobalIndexer { private final List<DataField> fields; - private final ESIndexOptions indexOptions; + private final Options options; + @Nullable private final ESIndexOptions indexOptions; private final String configuredVectorMetric; private volatile String readerVectorMetric; public ESIndexGlobalIndexer(List<DataField> fields, Options options) { this.fields = Collections.unmodifiableList(new ArrayList<>(fields)); - this.indexOptions = new ESIndexOptions(this.fields, options); - this.configuredVectorMetric = primaryVectorMetric(this.fields, this.indexOptions); + this.options = options; + this.indexOptions = tryParseIndexOptions(this.fields, options); + this.configuredVectorMetric = + this.fields.isEmpty() + ? null + : ESIndexOptions.configuredVectorMetric(this.fields.get(0), options); + } + + /** + * Readers use the field configuration persisted in the index metadata, so the current options + * only have to describe a build when a writer is created. For example, the dimension of an + * ARRAY<FLOAT> column is usually passed to the build procedure only and is absent from + * the table options that the read path passes here. + */ + @Nullable + private static ESIndexOptions tryParseIndexOptions(List<DataField> fields, Options options) { + try { + return new ESIndexOptions(fields, options); + } catch (IllegalArgumentException e) { + return null; + } } @Override public GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter) throws IOException { - return new ESIndexGlobalIndexWriter(fileWriter, fields, indexOptions); + // Re-parse so that options which cannot describe a build fail here with the original + // message instead of being silently ignored. + ESIndexOptions writerOptions = + indexOptions != null ? indexOptions : new ESIndexOptions(fields, options); + return new ESIndexGlobalIndexWriter(fileWriter, fields, writerOptions); } @Override @@ -99,14 +123,4 @@ public class ESIndexGlobalIndexer implements VectorGlobalIndexer { + "."); } } - - private static String primaryVectorMetric(List<DataField> fields, ESIndexOptions indexOptions) { - if (fields.isEmpty()) { - return null; - } - FieldIndexConfig config = indexOptions.getConfig(fields.get(0).name()); - return config != null && config.indexType() == FieldIndexConfig.IndexType.VECTOR - ? ESIndexOptions.toPaimonVectorMetric(config.metric()) - : null; - } } diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java index 60b8a704f7..7678954bb4 100644 --- a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java +++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java @@ -159,6 +159,21 @@ public class ESIndexOptions { return fieldConfigs.get(fieldName); } + /** + * Resolves the vector metric configured for {@code field} in Paimon's exact-search vocabulary, + * or {@code null} when the field is not indexed as a vector. Unlike the constructor, this does + * not require the options to fully describe a build (for example an ARRAY<FLOAT> + * dimension), so it is usable when the options only need to describe reads. + */ + static String configuredVectorMetric(DataField field, Options options) { + String explicitType = resolve(options, field.name(), "type", null); + boolean vector = + explicitType == null + ? isVectorType(field.type()) + : "vector".equalsIgnoreCase(explicitType); + return vector ? toPaimonVectorMetric(resolveVectorMetric(field.name(), options)) : null; + } + /** Maps an ESLib/Lucene vector metric name to Paimon's exact-search metric vocabulary. */ static String toPaimonVectorMetric(String metric) { String normalized = metric == null ? "l2" : metric.toLowerCase(Locale.ROOT); @@ -345,7 +360,7 @@ public class ESIndexOptions { + MAX_VECTOR_DIMENSION + "."); } - String metric = validateMetric(fieldName, resolve(options, fieldName, "metric", "cosine")); + String metric = resolveVectorMetric(fieldName, options); Map<String, String> params = new LinkedHashMap<>(); // Table-level algorithm parameters are defaults only for fields using that algorithm. A @@ -529,6 +544,10 @@ public class ESIndexOptions { return value; } + private static String resolveVectorMetric(String fieldName, Options options) { + return validateMetric(fieldName, resolve(options, fieldName, "metric", "cosine")); + } + private static String validateMetric(String fieldName, String metric) { String normalized = metric.toLowerCase(Locale.ROOT); switch (normalized) { diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java index 505de4d53e..255fb65ed0 100644 --- a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java +++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java @@ -18,11 +18,17 @@ package org.apache.paimon.eslib.index; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.ScoredGlobalIndexResult; import org.apache.paimon.globalindex.VectorGlobalIndexer; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.VectorSearch; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; @@ -31,11 +37,18 @@ import org.elasticsearch.eslib.api.model.VectorAlgorithm; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -45,6 +58,11 @@ class ESIndexVectorMetricTest { private static final List<DataField> VECTOR_FIELDS = List.of(new DataField(0, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT()))); + /** An ARRAY<FLOAT> column carries no dimension, so only the options can describe it. */ + private static final List<DataField> ARRAY_FIELDS = + Collections.singletonList( + new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT()))); + @Test void exposesCanonicalPaimonMetric() { GlobalIndexer indexer = indexer("dp"); @@ -126,6 +144,83 @@ class ESIndexVectorMetricTest { .hasMessageContaining("must be positive"); } + @Test + void readerDoesNotRequireCurrentOptionsToDescribeBuild(@TempDir Path tempDir) throws Exception { + // Build time: the dimension only reaches the indexer through the procedure options. + Map<String, String> buildOptions = new LinkedHashMap<>(); + buildOptions.put("global-index.es-index.fields.embedding.dimension", "2"); + buildOptions.put("global-index.es-index.fields.embedding.metric", "cosine"); + ESIndexGlobalIndexer buildIndexer = + new ESIndexGlobalIndexer(ARRAY_FIELDS, Options.fromMap(buildOptions)); + Path archiveDir = tempDir.resolve("build"); + Files.createDirectories(archiveDir); + ESIndexGlobalIndexWriter writer = + (ESIndexGlobalIndexWriter) + buildIndexer.createWriter(new LocalDirWriter(archiveDir)); + writer.write(new float[] {1f, 0f}, 0); + writer.write(new float[] {0f, 1f}, 1); + writer.write(new float[] {-1f, 0f}, 2); + ResultEntry entry = writer.finish().get(0); + Path archive = archiveDir.resolve(entry.fileName()); + GlobalIndexIOMeta ioMeta = + new GlobalIndexIOMeta( + new org.apache.paimon.fs.Path(archive.toString()), + Files.size(archive), + entry.meta()); + + // Read time: core only passes the table options, which carry no dimension. + ESIndexGlobalIndexer readIndexer = new ESIndexGlobalIndexer(ARRAY_FIELDS, new Options()); + GlobalIndexReader reader = + readIndexer.createReader( + meta -> LocalFileIO.create().newInputStream(meta.filePath()), + Collections.singletonList(ioMeta), + 3, + null); + try { + Optional<ScoredGlobalIndexResult> result = + reader.visitVectorSearch(new VectorSearch(new float[] {1f, 0f}, 1, "embedding")) + .join(); + assertThat(result).isPresent(); + assertThat(result.get().results().contains(0L)).isTrue(); + assertThat(readIndexer.metric()).isEqualTo("cosine"); + } finally { + reader.close(); + } + } + + @Test + void writerStillRejectsOptionsThatCannotDescribeBuild(@TempDir Path tempDir) { + ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(ARRAY_FIELDS, new Options()); + assertThatThrownBy(() -> indexer.createWriter(new LocalDirWriter(tempDir))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("requires a positive dimension"); + } + + @Test + void legacyMetadataStillRequiresCurrentOptions() throws Exception { + // Offset-only metadata predates persisted field configs: [fileCount][name][offset][len]. + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(bytes); + out.writeInt(1); + byte[] name = "segments_1".getBytes(StandardCharsets.UTF_8); + out.writeInt(name.length); + out.write(name); + out.writeLong(0L); + out.writeLong(1L); + out.flush(); + GlobalIndexIOMeta legacy = + new GlobalIndexIOMeta( + new org.apache.paimon.fs.Path("legacy.index"), 1L, bytes.toByteArray()); + + ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(ARRAY_FIELDS, new Options()); + assertThatThrownBy( + () -> + indexer.createReader( + meta -> null, Collections.singletonList(legacy), 1, null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Legacy es-index metadata"); + } + private static ESIndexGlobalIndexer indexer(String metric) { Map<String, String> options = new LinkedHashMap<>(); options.put("global-index.es-index.fields.embedding.metric", metric); @@ -161,4 +256,25 @@ class ESIndexVectorMetricTest { archiveSize, metadata); } + + /** {@link GlobalIndexFileWriter} backed by a local directory. */ + private static final class LocalDirWriter implements GlobalIndexFileWriter { + private final Path dir; + private final LocalFileIO fio = LocalFileIO.create(); + + LocalDirWriter(Path dir) { + this.dir = dir; + } + + @Override + public String newFileName(String prefix) { + return prefix + "-" + UUID.randomUUID() + ".index"; + } + + @Override + public PositionOutputStream newOutputStream(String fileName) throws IOException { + return fio.newOutputStream( + new org.apache.paimon.fs.Path(dir.resolve(fileName).toString()), true); + } + } }
