chrevanthreddy commented on code in PR #20107: URL: https://github.com/apache/hudi/pull/20107#discussion_r4384291517
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/vector/VectorIndexer.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.hudi.metadata.index.vector; + +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.index.vector.VectorIndexOptions; +import org.apache.hudi.common.model.HoodieIndexDefinition; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieMetadataException; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.metadata.index.BaseIndexer; +import org.apache.hudi.metadata.index.EngineIndexerSupport; +import org.apache.hudi.metadata.index.model.IndexCleanContext; +import org.apache.hudi.metadata.index.model.IndexInitializationContext; +import org.apache.hudi.metadata.index.model.IndexInitializationPlan; +import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords; +import org.apache.hudi.metadata.index.model.IndexUpdateContext; +import org.apache.hudi.metadata.model.FileSliceAndPartition; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.hudi.metadata.MetadataPartitionType.VECTOR_INDEX; + +/** + * Implementation of {@link MetadataPartitionType#VECTOR_INDEX} index. + * <p> + * Owns partition sizing (IVF cluster to file-group mapping) and initialization-plan assembly. + * The engine-specific record generation (reading the source VECTOR column, IVF clustering and + * RaBitQ quantization) is dispatched through {@link EngineIndexerSupport}. + * <p> + * Index update, clean and restore lifecycle is handled in a follow-up change; for now those hooks + * are no-ops so the initial bootstrap path is self-contained and reviewable. + */ +@Slf4j +public class VectorIndexer extends BaseIndexer { + + /** + * Target size per metadata file group used by the sizing guard. Vector postings are routed to + * file groups by cluster id, so this is only a sanity bound against gross over-provisioning. + */ + private static final long TARGET_FILE_GROUP_BYTES = 512L * 1024L * 1024L; + + private final EngineIndexerSupport engineIndexerSupport; + + public VectorIndexer( + HoodieEngineContext engineContext, + HoodieWriteConfig dataTableWriteConfig, + HoodieTableMetaClient dataTableMetaClient, + EngineIndexerSupport engineIndexerSupport) { + super(engineContext, dataTableWriteConfig, dataTableMetaClient); + this.engineIndexerSupport = engineIndexerSupport; + } + + @Override + public List<IndexInitializationPlan> buildInitialization(IndexInitializationContext context) throws IOException { + Set<String> completedPartitions = dataTableMetaClient.getTableConfig().getMetadataPartitions(); + Set<String> vectorIndexPartitionsToInit = dataTableMetaClient.getIndexMetadata() + .map(metadata -> metadata.getIndexDefinitions().values().stream() + .map(HoodieIndexDefinition::getIndexName) + .filter(name -> name.startsWith(VECTOR_INDEX.getPartitionPath())) + .filter(name -> !completedPartitions.contains(name)) + .collect(Collectors.toSet())) + .orElse(Collections.emptySet()); + if (vectorIndexPartitionsToInit.size() != 1) { + if (vectorIndexPartitionsToInit.size() > 1) { + log.warn("Skipping vector index initialization as only one vector index bootstrap at a time " + + "is supported for now. Provided: {}", vectorIndexPartitionsToInit); + } + return Collections.emptyList(); + } + + String indexName = vectorIndexPartitionsToInit.iterator().next(); + HoodieIndexDefinition indexDefinition = HoodieTableMetadataUtil.getHoodieIndexDefinition(indexName, dataTableMetaClient); + ValidationUtils.checkState(indexDefinition != null, "Vector index definition is not present for index " + indexName); + + int fileGroupCount = resolveFileGroupCount(indexDefinition); + + HoodieSchema tableSchema = context.tableSchema().get() + .orElseThrow(() -> new HoodieMetadataException("Table schema is not available for vector index initialization")); + List<FileSliceAndPartition> fileSlices = context.latestFileSlices().get(); + + // Initial bootstrap always allocates generation 1 (no pre-existing MDT partition to advance). + HoodieData<HoodieRecord> records = engineIndexerSupport.generateVectorIndexRecords( + indexDefinition, dataTableMetaClient, fileSlices, tableSchema, 1); + + return Collections.singletonList(IndexInitializationPlan.of(fileGroupCount, indexName, records)); + } + + /** + * Sizes the vector metadata partition by IVF clusters. Posting rows are routed to file groups by + * cluster id ({@code mapVectorPostingKeyToFileGroupIndex}), so + * {@code fileGroupCount = ceil(numClusters / clustersPerFileGroup)} keeps one cluster per file group + * by default. An explicit {@code hoodie.metadata.vector.index.file.group.count} overrides the derived + * value. + */ + private int resolveFileGroupCount(HoodieIndexDefinition indexDefinition) { + VectorIndexOptions.ResolvedOptions resolvedOptions = VectorIndexOptions.resolve(indexDefinition.getIndexOptions()); + int numClusters = resolvedOptions.numClusters; + int clustersPerFileGroup = 1; + int configuredFileGroupCount = dataTableWriteConfig.getMetadataConfig().getVectorIndexFileGroupCount(); + int fileGroupCount = configuredFileGroupCount > 0 + ? configuredFileGroupCount + : Math.max(1, (int) Math.ceil((double) numClusters / clustersPerFileGroup)); + + long targetBlockBytes = 64L * 1024L; + long projectedIndexBytes = Math.max(1L, (long) numClusters * targetBlockBytes); + if ((long) fileGroupCount * TARGET_FILE_GROUP_BYTES > projectedIndexBytes * 10L) { + log.warn("Vector index {} is initializing with {} file groups for projected index size {} bytes. " + + "This exceeds the 10x sizing guard against 512 MiB target file groups; consider setting {} " + + "to a smaller value before bootstrap.", + indexDefinition.getIndexName(), fileGroupCount, projectedIndexBytes, + org.apache.hudi.common.config.HoodieMetadataConfig.VECTOR_INDEX_FILE_GROUP_COUNT.key()); + } + log.info("Initializing vector index {} with {} file groups " + + "[numClusters={}, clustersPerFileGroup={}, configuredFileGroupCount={}]", + indexDefinition.getIndexName(), fileGroupCount, numClusters, clustersPerFileGroup, configuredFileGroupCount); + return fileGroupCount; + } + + @Override + public List<IndexPartitionAndRecords> buildUpdate(IndexUpdateContext context) { + // Vector index incremental maintenance (append/delete postings, generation advance) is handled + // in a follow-up change. Initialization is a full re-bootstrap until then. + return Collections.emptyList(); + } + + @Override + public List<IndexPartitionAndRecords> buildClean(IndexCleanContext context) { + // See buildUpdate: lifecycle maintenance is handled in a follow-up change. + return Collections.emptyList(); + } Review Comment: Yes. This PR is intentionally bootstrap + read path only. Synchronous maintenance lives in the stacked PR **https://github.com/chrevanthreddy/hudi/pull/5** (based on this one): `buildUpdate` encodes inserts, vector updates, deletes and cluster moves per source commit, and skips posting work on compaction (as RLI/SI do). It's validated by an insert / upsert / replay / delete / clustering lifecycle that checks exact search == brute force at 10M (COW + MOR) and at 1B (COW). `buildClean` stays a no-op by design: cleaning removes no live rows, so postings don't change. Compatibility follows the existing MDT dispatch contract: pre-vector binaries reject the unknown vector partition in `MetadataPartitionType.fromPartitionPath`; tables with a vector index are not writable by those binaries. Same-version writers maintain it unconditionally, including clients initialized before CREATE INDEX (covered by the production-client regression). Per-commit markers are audit records; strict validation is read-only, opt-in, defaults off, and has no correctness role. Its policy defaults WARN in both modes. There is no per-commit frontier/barrier or catch-up protocol. Diagnostic gaps are repaired by dropping and recreating the index, not an unimplemented replay operation. This describes the complete maintenance stack; base alone still has a buildUpdate stub. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieVectorSearchPlanBuilder.scala: ########## @@ -17,365 +17,473 @@ package org.apache.spark.sql.hudi.analysis +import org.apache.hudi.SparkAdapterSupport +import org.apache.hudi.client.common.HoodieSparkEngineContext +import org.apache.hudi.common.config.HoodieMetadataConfig +import org.apache.hudi.common.config.TypedProperties +import org.apache.hudi.common.index.vector.{ScoredVectorPostingMatch, VectorDistanceMetric, VectorIndexArbiter, VectorIndexMdtSearchUtils, VectorIndexMetadataCache, VectorIndexOptions, VectorPostingMatch, VectorStalePolicy} +import org.apache.hudi.common.model.{FileSlice, HoodieIndexDefinition, HoodieRecord} import org.apache.hudi.common.schema.HoodieSchema - -import org.apache.spark.sql.{AnalysisException, DataFrame, SparkSession} +import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver} +import org.apache.hudi.common.table.timeline.{HoodieTimeline, InstantComparison} +import org.apache.hudi.common.table.timeline.InstantComparison.compareTimestamps +import org.apache.hudi.common.table.view.HoodieTableFileSystemView +import org.apache.hudi.data.HoodieJavaRDD +import org.apache.hudi.hadoop.fs.HadoopFSUtils +import org.apache.hudi.metadata.{HoodieTableMetadata, HoodieTableMetadataUtil} + +import org.apache.spark.api.java.JavaSparkContext +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SparkSession} +import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.HoodieVectorSearchTableValuedFunction import org.apache.spark.sql.catalyst.plans.logical.HoodieVectorSearchTableValuedFunction.{DistanceMetric, SearchAlgorithm} import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan import org.apache.spark.sql.expressions.Window import org.apache.spark.sql.functions.{broadcast, col, monotonically_increasing_id, row_number} import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException -import org.apache.spark.sql.types.{ArrayType, ByteType, DataType, DoubleType, FloatType} +import org.apache.spark.sql.types.{ArrayType, ByteType, DataType, DoubleType, FloatType, Metadata, StructField, StructType} +import org.apache.spark.util.LongAccumulator +import org.slf4j.LoggerFactory +import scala.collection.JavaConverters._ +import scala.collection.mutable.ArrayBuffer import scala.util.{Failure, Success, Try} /** - * Extension point for vector search algorithms. Each implementation provides - * the Spark logical plan for single-query and batch-query KNN search. - * - * To add a new algorithm (e.g. RowMatrix, HNSW): - * 1. Create an object extending this trait - * 2. Add a value to [[SearchAlgorithm]] - * 3. Register the mapping in [[HoodieVectorSearchPlanBuilder.resolveAlgorithm]] + * IVF + RaBitQ vector search backed by native metadata-table posting prefix lookups. * - * Implementations can use the shared validation helpers on - * [[HoodieVectorSearchPlanBuilder]] (validateEmbeddingColumn, validateBatchDimensions, etc.) - * and the raw distance functions on [[VectorDistanceUtils]]. - * - * The output schema contract: - * - Single-query: all corpus columns (minus the embedding column) + `_hudi_distance: Double` - * - Batch-query: all corpus columns (minus the embedding column) + clashing query columns - * (prefixed with `_hudi_query_`) + `_hudi_distance: Double` + `_hudi_query_index: Long` - * - Results are ordered by `_hudi_distance` ascending (lower = more similar) - * - `_hudi_query_index` is an opaque grouping identifier (not a sequential index). Values may be - * large non-contiguous numbers because they are generated by `monotonically_increasing_id()`. + * The algorithm keeps Spark SQL out of the posting scan: it probes centroids, converts + * the selected clusters to MDT posting prefixes, reads posting rows with HFile-friendly + * prefix lookups, reduces with RaBitQ approximate scores before record-location lookup, + * then exact-reranks the fetched Hudi rows. */ -trait VectorSearchAlgorithm { +object IvfRaBitQMdtSearchAlgorithm extends VectorSearchAlgorithm with SparkAdapterSupport { + private val APPROXIMATE_CANDIDATE_FLOOR = 32 - /** Human-readable name for error messages and logging. */ - def name: String + private[analysis] def approximateCandidateHeapSize(k: Int): Int = + math.max(k, APPROXIMATE_CANDIDATE_FLOOR) - /** - * Build a plan that finds the k nearest corpus rows to a single query vector. - * - * @param spark active SparkSession - * @param corpusDf resolved corpus DataFrame (may be Hudi, Parquet, or temp view) - * @param embeddingCol name of the array-typed embedding column in corpusDf - * @param queryVector the query vector, normalized to Array[Double] - * @param k number of nearest neighbors to return - * @param metric distance metric (COSINE, L2, DOT_PRODUCT) - * @param filter optional SQL predicate applied to the corpus before distance computation, - * String literals inside the predicate must use double quotes (e.g. {@code label = "z-axis"}) - * because the filter is passed as a single-quoted TVF argument and - * Spark's lexer does not support {@code ''} as an escape sequence - * inside string literals. - * @param maxDistance optional distance threshold; results exceeding this value are excluded - * @return an analyzed LogicalPlan whose output matches the single-query schema contract - */ - def buildSingleQueryPlan( - spark: SparkSession, - corpusDf: DataFrame, - embeddingCol: String, - queryVector: Array[Double], - k: Int, - metric: DistanceMetric.Value, - filter: Option[String] = None, - maxDistance: Option[Double] = None): LogicalPlan - - /** - * Build a plan that finds the k nearest corpus rows for each row in the query table. - * - * @param spark active SparkSession - * @param corpusDf resolved corpus DataFrame - * @param corpusEmbeddingCol name of the embedding column in corpusDf - * @param queryDf resolved query DataFrame - * @param queryEmbeddingCol name of the embedding column in queryDf - * @param k number of nearest neighbors per query - * @param metric distance metric (COSINE, L2, DOT_PRODUCT) - * @param filter optional SQL predicate applied to the corpus before distance - * computation, and shrinking cross-join cardinality. Applied to the - * corpus only — to restrict the query side, apply a projection or - * filter on the query table before invoking the TVF. See - * [[buildSingleQueryPlan]] for quoting requirements. - * @param maxDistance optional distance threshold; results exceeding this value are - * excluded before per-query top-K selection, reducing shuffle volume. - * @return an analyzed LogicalPlan whose output matches the batch-query schema contract - * @note Batch mode broadcasts the query table to all executors via a cross-join. - * This is designed for small-to-medium query sets (tens to low hundreds of rows). - * For large query tables, memory pressure on executors may occur. - */ - def buildBatchQueryPlan( - spark: SparkSession, - corpusDf: DataFrame, - corpusEmbeddingCol: String, - queryDf: DataFrame, - queryEmbeddingCol: String, - k: Int, - metric: DistanceMetric.Value, - filter: Option[String] = None, - maxDistance: Option[Double] = None): LogicalPlan -} + import HoodieVectorSearchPlanBuilder._ -/** - * Resolves [[SearchAlgorithm]] values to [[VectorSearchAlgorithm]] implementations - * and provides shared validation helpers used across algorithms. - */ -object HoodieVectorSearchPlanBuilder { - - val DISTANCE_COL = "_hudi_distance" - private[analysis] val QUERY_ID_COL = "_hudi_query_index" - private[analysis] val QUERY_EMB_ALIAS = "_hudi_query_emb" - private[analysis] val RANK_COL = "_hudi_rank" - private[analysis] val QUERY_COL_PREFIX = "_hudi_query_" - - /** Resolve a [[SearchAlgorithm]] enum value to its implementation. */ - def resolveAlgorithm(algorithm: SearchAlgorithm.Value): VectorSearchAlgorithm = algorithm match { - case SearchAlgorithm.BRUTE_FORCE => BruteForceSearchAlgorithm - case other => throw new HoodieAnalysisException( - s"Unsupported search algorithm: $other") + private[analysis] def isLogResidentCandidate( + candidate: ScoredVectorPostingMatch, + currentBaseInstant: String, + sliceHasLogs: Boolean): Boolean = { + // A negative position is the authoritative log-resident signal: base-file records always carry a + // non-negative row position, so a candidate with position < 0 in a slice that still has log files + // lives in a log block and cannot be materialized from the base Parquet. The candidate's instant + // is NOT a reliable discriminator here -- a log delta can share the slice's base instant time. + val location = candidate.getLocation + sliceHasLogs && location != null && location.getPosition < 0 } - private[analysis] def validateEmbeddingColumn(df: DataFrame, colName: String): Unit = { - val fieldOpt = df.schema.fields.find(_.name.equalsIgnoreCase(colName)) - val field = fieldOpt.getOrElse( - throw new HoodieAnalysisException( - s"Embedding column '$colName' not found in table schema. " + - s"Available columns: ${df.schema.fieldNames.mkString(", ")}")) - field.dataType match { - case ArrayType(FloatType, _) | ArrayType(DoubleType, _) | ArrayType(ByteType, _) => // valid - case other => - throw new HoodieAnalysisException( - s"Embedding column '$colName' has type $other, " + - "expected array<float>, array<double>, or array<byte>") + private[analysis] def approximateArbitratedCandidate( + candidate: ScoredVectorPostingMatch): Option[ScoredVectorPostingMatch] = + candidate.getArbiterDecision match { + case VectorIndexArbiter.Decision.SERVE => Some(candidate) + // A delta carries the updated vector code. Refreshing only its locator is safe and preserves + // the new score; a stale packed-block row carries old code and must never be resurrected. + case VectorIndexArbiter.Decision.STALE if candidate.isDelta => Some(candidate) + case _ => None } + + private[analysis] case class FreshnessLag(firstUnmarkedInstant: String, lagCount: Int) + + private[analysis] def completedSourceWriteTimeline(timeline: HoodieTimeline): HoodieTimeline = + timeline.getWriteTimeline.filterCompletedInstants() + + private[analysis] def freshnessLag( + completedWriteInstants: Seq[String], + coveredInstant: Option[String]): Option[FreshnessLag] = { + val uncovered = completedWriteInstants.filter(instant => coveredInstant.forall(covered => Review Comment: **Fixed (bootstrap)** in `41bdd71c5b`; maintenance follow-up: https://github.com/chrevanthreddy/hudi/pull/5. Bootstrap uses the contiguous completed-write cutoff (`getContiguousCompletedWriteTimeline`), consistent with async index scheduling, rather than treating an inflight write as covered. The inflight-writer regression verifies the baseline excludes writer A, A emits its marker when it commits, and exact search includes both A and B. Update (2026-10-27): the maintenance follow-up no longer advances a per-write manifest frontier or maintains coverage barriers. Normal writes maintain postings and an audit marker together; strict read validation is opt-in/default off, with WARN the default policy in both modes. This does not remove the existing async-index bootstrap/catch-up lifecycle. Latest Spark 4.0 vector-only validation: the inflight-bootstrap case first raised `TableNotFoundException` while constructing the commit's table client, then passed on retry. This concurrent config-read issue is retained as a separate investigation, not claimed fixed by the vector patch. The same-version pre-index client and four lifecycle/rollback cases passed without reruns. The stacked PR records the validation details. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
