rmahindra123 commented on code in PR #19613: URL: https://github.com/apache/hudi/pull/19613#discussion_r3806435371
########## rfc/rfc-110/rfc-110.md: ########## @@ -0,0 +1,1145 @@ +<!-- + 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. +--> + +# RFC-110: Hudi Full-Text Search Index + +## Proposers + +- @danny0405 + +## Reviewers + +- @vinothchandar + +## Approvers + +- TBD + +## Status + +Issue: TBD + +RFC number: [RFC-110 reservation PR](https://github.com/apache/hudi/pull/19614) + +Status: Under review + +## Abstract + +This RFC proposes a full-text index for Hudi tables. Spark exposes SQL predicates +for text filtering, while Hudi RS provides a Python API for filtered and ranked +search. Hudi's index metadata stores the definition, and the metadata table +(MDT) tracks index state and coverage. + +Index creation and refresh run through Hudi's existing indexing action, outside +the ingestion job. Normal queries remain complete while an index is building or +awaiting refresh: Hudi uses the index where it is current and reads the remaining +file slices normally. MDT tracks index state and coverage, while the larger +dictionaries and posting lists live in auxiliary files. + +## Background + +Hudi tables often contain logs, support conversations, product descriptions, +audit events, and other free-form text. Existing Hudi indexes help with record +lookup and structured predicates, but cannot answer content-based queries over +these fields. + +The proposed index makes text search part of the Hudi table instead of requiring +a separately managed copy of its data. Spark SQL can combine text and structured +predicates, and Hudi RS can serve ranked searches from Python. Both operate on a +Hudi snapshot and retain normal Hudi behavior for time travel, rollback, +cleaning, record materialization, and access filtering. + +### Motivation and use cases + +Native full-text search is useful when text queries must follow Hudi commits, +time travel, rollback, cleaning, and access filters: + +- **Logs and observability.** Find records containing an error signature or + phrase while applying normal predicates such as service, environment, and + event time. +- **Product catalogs, support tickets, and CRM text.** Match user terms across + title and description while filtering on category, tenant, status, or access + policy. +- **Security, audit, and eDiscovery.** Investigate text as of an incident or + legal-hold snapshot and reproduce the exact records visible at that instant. + +For example, a CDC pipeline may lag after a GDPR deletion commits to Hudi. A +separate search cluster can still return the deleted record until the pipeline +catches up. With a Hudi text index, the query runs against the table snapshot +that contains the deletion. The same model also allows an application to search +`AS OF` an older instant without maintaining a separate search copy for each +retained snapshot. + +Text and vector indexes could later support hybrid BM25 and semantic retrieval +over one Hudi snapshot. Hybrid ranking is outside this RFC and does not create a +dependency on RFC-109. + +### Relation to existing Hudi indexes + +The proposal builds on: + +- [RFC-45](../rfc-45/rfc-45.md), whose MDT indexing lifecycle is used for + bootstrap and maintenance; +- [RFC-77](../rfc-77/rfc-77.md), which established dynamically named secondary + index partitions and index definitions; +- Spark SQL scalar predicates for relational queries; and +- Hudi RS table readers and Python bindings for direct search. + +Term dictionaries, posting lists, and BM25 ranking are standard search +structures. This proposal focuses on the Hudi-specific parts: building them from +file slices, publishing them through MDT, and selecting the right files for a +table snapshot. + +### User experience at a glance + +1. A user creates a text index on one or more string columns with `CREATE + INDEX`. Hudi builds it through the existing indexing action, outside the + ingestion job. +2. Spark users add `hudi_match`, `hudi_match_phrase`, or `hudi_multi_match` to a + normal `WHERE` clause. Hudi RS users can filter a snapshot or request ranked + results from Python. +3. Normal queries return complete results even while an index is building or + awaiting refresh. Hudi uses the index for covered file slices and scans the + remaining slices through the normal reader. +4. Users run `REFRESH INDEX` to index changed file slices. Operators can schedule + the same refresh with their normal workflow orchestration. Ingestion does not + load the native builder. +5. Applications that prefer lower latency over complete results can use Hudi RS + `.fast_search()`. The result reports whether any slices were skipped. + +### Goals + +- Match token, boolean, prefix, fuzzy, phrase, and multi-column queries. +- Support copy-on-write (COW) and merge-on-read (MOR) tables. +- Build asynchronously and incrementally through Hudi's existing indexing + action. The text-index procedure handles bootstrap, refresh, consolidation, + rebuild, repair, and cleanup without introducing another table service. +- Return complete results from SQL and from the default direct API mode. +- Range-read index blocks from object storage without loading an entire index + into a Spark executor or Hudi RS process. Builders spill sorted runs and + readers bound dictionary, posting-block, and query-expansion memory. +- Provide a direct Hudi RS Python API alongside Spark SQL. + +### Non-goals + +- Elasticsearch API, aggregation, highlighting, or percolator compatibility. +- Highlighting and custom relevance models in the first text-index file format. +- Updating posting lists in place. +- Replacing SQL predicate indexes or the record index. + +## User interface + +### Create and refresh an index + +Spark SQL is the table-query interface. Hudi RS Python is the direct API for +applications that do not run Spark. Index creation follows Hudi's existing +secondary-index syntax: + +```sql +CREATE INDEX log_message_fts +ON application_logs +USING text_index (message) +OPTIONS ( + 'base_tokenizer' = 'simple', + 'lower_case' = 'true', + 'language' = 'und', + 'with_position' = 'true', + 'posting_block_size' = '128' +); +``` + +`CREATE INDEX` starts the first build through Hudi's existing indexing action. +The build does not run in the ingestion job. New writes may leave some file +slices temporarily uncovered, but normal queries still return complete results +by scanning those slices. + +Users can bring the index up to date explicitly: + +```sql +REFRESH INDEX log_message_fts ON application_logs; +``` + +The same refresh can be scheduled through the `HoodieIndexer` utility. No +resident text-index service is required. + +### Spark SQL predicates + +Spark exposes three Hudi scalar predicates for use in `WHERE` clauses: +`hudi_match`, `hudi_match_phrase`, and `hudi_multi_match`. + +| Predicate | Use | +| --- | --- | +| `hudi_match` | Match analyzed terms in one column. | +| `hudi_match_phrase` | Match terms in order, with optional phrase slop. | +| `hudi_multi_match` | Apply one text query across several columns. | + +```sql +SELECT event_ts, service, message +FROM application_logs +WHERE hudi_match(message, 'connection refused', 'operator=AND') + AND service = 'payments'; +``` + +Phrase and multi-column queries use companion predicates: + +```sql +SELECT event_ts, service, message +FROM application_logs +WHERE hudi_match_phrase(message, 'out of memory', 0) + AND environment = 'production'; + +SELECT _hoodie_record_key, subject, description +FROM support_tickets +WHERE hudi_multi_match('payment declined', subject, description) + AND status = 'open'; +``` + +The v1 signatures are: + +```text +hudi_match(column, query [, 'key=value,...']) -> boolean +hudi_match_phrase(column, query [, slop]) -> boolean +hudi_multi_match(query [, 'operator=AND|OR'], column, ...) -> boolean +``` + +`hudi_match` accepts `operator`, `fuzziness`, `prefix_length`, and +`max_expansions`; unknown options are rejected. These functions can be combined +with normal Spark predicates. Pushdown is limited to expressions whose Boolean +semantics can be preserved. + +The functions return Boolean values. They do not add `_score`, and `LIMIT` does +not imply relevance order. Ranked search is provided by the direct API. + +### Hudi RS Python table API + +Hudi RS extends `HudiTableBuilder` and `read_snapshot` with structured text +queries. `read_snapshot` returns every matching row, while `search_text` ranks +the best matches with BM25. + +```python +import pyarrow as pa + +from hudi import HudiTableBuilder +from hudi.search import FullTextOperator, MatchQuery, PhraseQuery + +table = ( + HudiTableBuilder + .from_base_uri("s3://warehouse/support_tickets") + .build() +) + +query = MatchQuery( + "payment declined", + column="description", + operator=FullTextOperator.AND, +) + +batches = table.read_snapshot( + columns=["_hoodie_record_key", "subject", "description"], + filters=[("status", "=", "open")], + full_text_query=query, +) +tickets = pa.Table.from_batches(batches) +``` + +Queries can be combined: + +```python +query = ( + MatchQuery("refund", column="description") + & PhraseQuery("credit card", column="description", slop=1) +) + +batches = table.read_snapshot(full_text_query=query) +``` + +For ranked search, the API returns BM25 scores: + +```python +results = ( + table.search_text( + MatchQuery("payment declined", column="description"), + ) + .where([("status", "=", "open")]) + .select(["_hoodie_record_key", "subject", "description"]) + .limit(20) + .to_arrow() +) + +# Ranked by BM25 descending; `_score` is included in `results`. +``` + +`read_snapshot(full_text_query=...)` returns all matches. +`search_text(...).limit(k)` returns ranked rows and an `_score` column. Both use +the complete query behavior described below. Latency-sensitive ranked searches +may call `.fast_search()` before execution: + +```python +results = ( + table.search_text(MatchQuery("payment declined", column="description")) + .fast_search() + .limit(20) + .to_arrow() +) +``` + +`fast_search()` skips file slices without an exact compatible index segment. +If it skips any slice, result metadata sets `is_complete=false`; otherwise it +sets `is_complete=true`. Spark SQL and `read_snapshot(full_text_query=...)` +always use fallback scans and return complete results in the first release. + +## Detailed design + +The design has four main pieces: + +1. Hudi's existing indexing action schedules and coordinates each build or + refresh. +2. `TextIndexer` performs the text-specific work for a `TEXT_INDEX` MDT + partition. +3. MDT stores small control records that identify usable index files and the + data file slices they cover. +4. Query planning uses those records to choose index lookup or the normal Hudi + reader for each file slice. + +### Terminology + +| Term | Meaning | +| --- | --- | +| Text index definition | The named `HoodieIndexDefinition`, indexed columns, analyzer options, and `HoodieIndexVersion`. | +| Text index MDT partition | Dynamic `text_index_<name>` metadata partition backed by normal MDT MOR file groups. It contains only text-index control records, not dictionaries or postings. | +| MDT file group | A file group in the metadata table. Its base and log files store the `HEAD`, `SEGMENT`, `COVERAGE`, and `TOMBSTONE` records defined below. | +| Hudi indexing action | The existing `INDEXING_ACTION` used to build MDT index partitions. It owns scheduling, execution, timeline state, and coordination for every index type. | +| `TextIndexer` | The `BaseIndexer` implementation for `TEXT_INDEX`. It builds or refreshes auxiliary index files and returns the MDT control records to publish. It is not a table service. | +| Text-index file | An auxiliary file containing term dictionaries, postings, positions, or record addresses. It lives outside the MDT partition's file groups and becomes visible through MDT records. | +| Data table file group | The normal Hudi file group identified by partition path and file ID. | +| Data table file slice | One base file and its ordered log files for a file group at a particular Hudi instant. | +| Index segment | A logical, immutable set of text-index files built from one or more data table file slices. It has no Hudi base file or log files and is not a data-table or MDT file group. | +| Segment descriptor | The small `SEGMENT` record stored in an MDT file group. It identifies an index segment and contains its file paths, checksums, source instant, statistics, and file-slice mapping; it does not contain postings. | +| File-slice identity | The exact table UUID, partition, file ID, base file, log files, schema, and merger state represented by an index segment. | +| Coverage record | An MDT record mapping one data table file group to candidate index segments and their exact file-slice identities. | +| Fallback scan set | Data table file slices without compatible index coverage for the pinned snapshot, including changed slices awaiting refresh and slices affected by bootstrap, rebuild, or recovery. | +| Indexed record | One Hudi table record, located by its Hudi record key and data table file slice. | + +MDT MOR file groups store small, upsertable control records. The immutable term +dictionaries and postings live in auxiliary files referenced by those records. +Updating a `COVERAGE` record changes the segments a reader may select; it does +not write postings into an MDT file group or modify a published segment. + +An index segment may cover several data table file groups to avoid creating a +small object for every write. It maps each `file_slice_ordinal` to a partition +path, file ID, and exact file-slice identity. Each covered file group has a +`COVERAGE` record that points to the segment and ordinal. The record may retain +candidates for several data instants; the planner selects only an exact match +for the requested slice. + +### Index definition + +`CREATE INDEX` stores the following `HoodieIndexDefinition` for the earlier SQL +example: + +```json +{ + "indexName": "text_index_log_message_fts", + "indexType": "text_index", + "version": "V1", + "sourceFields": ["message"], + "indexFunction": "tokenize", + "indexOptions": { + "base_tokenizer": "simple", + "lower_case": "true", + "language": "und", + "with_position": "true", + "posting_block_size": "128" + } +} +``` + +`TEXT_INDEX` is added to `HoodieIndexVersion.getCurrentVersion`; the first +implementation returns `V1`. Changes to the definition or MDT record layout may +require a new `HoodieIndexVersion`. Analyzer and text-index file format versions +are kept separate from the index definition version. + +The first release creates indexes through Spark SQL or the Java index API. The +`HoodieIndexer` utility is extended to accept a dynamic text-index name and +refresh mode. `RefreshIndexCommand` submits the existing indexing action for an +active text index. A Hudi RS `create_text_index` API can be added when Hudi RS +supports the required MDT writes. + +### Hudi architecture + +`CREATE INDEX` writes a `HoodieIndexDefinition` and schedules the existing Hudi +indexing action for the dynamic MDT partition `text_index_<name>`. The generic +index executor resolves `TextIndexer` through `IndexerFactory`. `TextIndexer` +reads merged Hudi file slices, writes immutable index files to auxiliary +storage, and returns their descriptors and coverage for the executor to commit +to MDT. A `SEGMENT` record describes an auxiliary file set; it does not contain +the segment data. + +The index stays in `BUILDING` state until bootstrap covers the pinned snapshot. +It then becomes `ACTIVE` and can be used by the planner. Later writes may create +new file-slice identities that have not yet been indexed, so query planning +still checks coverage one slice at a time. + +Ingestion does not build text-index payloads. A new MOR log, COW replacement, +compaction, or clustering result changes the identity of the affected slice. +Until the next indexing-action refresh, the planner reads that slice normally. +The identity mismatch itself invalidates coverage; the writer does not need to +publish a separate invalidation record. + +```mermaid +flowchart LR + CREATE["CREATE INDEX"] + DEFINITION["HoodieIndexDefinition"] + DATA["Completed Hudi file slices"] + ACTION["Existing Hudi indexing action"] + INDEXER["TextIndexer<br/>TEXT_INDEX build procedure"] + FILES["Auxiliary storage<br/>immutable text-index files"] + MDT["text_index_<name> MDT partition<br/>MOR file groups containing control records"] + + CREATE --> DEFINITION + DEFINITION --> ACTION + ACTION --> INDEXER + DATA -->|"pinned completed snapshot"| INDEXER + INDEXER --> FILES + INDEXER -->|"return descriptors and coverage"| ACTION + ACTION -->|"commit control records"| MDT +``` + +A reader first selects the data file slices for the query, then looks up their +coverage in MDT. Exact matches use the index; all other slices use the normal +Hudi reader. Both paths return Hudi record addresses for row materialization. + +```mermaid +flowchart LR + QUERY["Spark SQL or Hudi RS Python"] + DATA["Eligible Hudi file slices"] + MDT["MDT MOR file groups<br/>coverage and segment descriptors"] + FILES["Auxiliary storage<br/>dictionaries and postings"] + PLANNER["Coverage planner"] + INDEX["Text-index reader"] + SCAN["Normal Hudi reader"] + ADDRESSES["Hudi record addresses"] + ROWS["Materialize Hudi rows"] + + QUERY --> PLANNER + DATA --> PLANNER + MDT --> PLANNER + PLANNER -->|"exact compatible coverage"| INDEX + PLANNER -->|"uncovered or incompatible"| SCAN + FILES -->|"referenced posting blocks"| INDEX + INDEX --> ADDRESSES + SCAN --> ADDRESSES + ADDRESSES --> ROWS +``` + +The index definition is stored in `.hoodie/.index/index.json`. Index files live +under the auxiliary path shown below, outside the dynamic MDT partition's MOR +file groups. An index file is visible only after its MDT records commit. + +### Metadata table integration + +`TEXT_INDEX` is added to `MetadataPartitionType` with the dynamic prefix +`text_index_`. Its MOR file groups receive incremental upserts for state, +descriptors, coverage, and tombstones. Dictionary and posting blocks remain in +the auxiliary index files. Partition lookup follows the secondary- and +expression-index code. `IndexerFactory` creates a `TextIndexer` that extends +`BaseIndexer`. + +`TextIndexer.buildInitialization` handles the first build of a `TEXT_INDEX` +partition. Ingestion still invokes the normal `Indexer.buildUpdate` dispatch for +available MDT partitions, but `TextIndexer.buildUpdate` returns no text-index +records and never loads the native builder. Exact file-slice identity detects +stale coverage at query time. + +The existing indexing action currently initializes a new MDT partition and then +expects ingestion writers to maintain it. Text search needs the same action to +refresh an already-active partition. This RFC adds an optional +`Indexer.buildRefresh(IndexRefreshContext)` hook and marks each index plan as +`INITIALIZE` or `REFRESH`. Existing indexers keep their current behavior. +`TextIndexer.buildRefresh` selects slices without exact coverage and produces +the replacement `SEGMENT`, `COVERAGE`, `HEAD`, and `TOMBSTONE` records. The +generic index executor remains responsible for timeline transitions, locking, +commit, and failure handling. No text-specific timeline action or service is +added. + +`HoodieMetadata.avsc` gets a tagged record named `HoodieTextIndexInfo` with four +record types: + +| Kind | Record key | Purpose | +| --- | --- | --- | +| `HEAD` | `head` | `BUILDING` or `ACTIVE` state, index-definition and text-index file format versions, analyzer identity, definition instant, latest publication instant, and aggregate statistics. | +| `SEGMENT` | `segment/<uuid>` | Descriptor for one auxiliary index segment: text-index file paths, sizes, checksums, statistics, source instant, and the covered file-slice ordinal map. | +| `COVERAGE` | `coverage/<encoded-partition>/<file-id>` | Exact file-slice identities and candidate segment references ordered by data instant. | +| `TOMBSTONE` | `tombstone/<uuid>` | Segment retirement instant and deletion eligibility. | + +The Avro record holds the index and format versions, analyzer fingerprint, +segment UUID, file-slice identities, ordinal mapping, summary statistics, file +descriptors, and optional tombstone instant. Dictionaries, postings, and +detailed term statistics stay in the index files. + +After partition pruning, the planner issues batched point lookups for +`coverage/<encoded-partition>/<file-id>` and loads only the referenced segment +descriptors. It may cache immutable descriptors for the life of the MDT +snapshot. Storage grows with the table's file-group count, while planning work +depends on the file groups selected by the query. A full-table query still +touches every file group, but uses normal MDT sharding and point lookups instead +of loading the control partition on the driver. + +The immutable files referenced by `SEGMENT` records use this default path: + +```text +<table>/.hoodie/metadata/.aux/text-index/ + <escaped-index-name>/<segment-uuid>/... +``` + +The metadata subsystem manages this path, but the path is not an MDT MOR +partition and its UUID directories are not MDT file groups. Readers validate +every descriptor path against the table UUID. An external storage root may be +supported later. Unpublished files are removed after the orphan grace period. + +### Hudi record identity and file-slice coverage + +The index addresses Hudi records directly. Version 1 requires a stable record +key and stores: + +```text +HudiTextIndexRecordAddress { Review Comment: is the storage format similar to secondary index? It stores each <record_key, file_slice> in a separate row? I feel like with text search, the cardinality could be a bigger issue than secondary indexes, do we need to handle that somehow? for instance, some text keywords might have a lot of references. Are we planning to handle that at the index level or we need to handle it in the storage level as well. -- 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]
