[core][spark][docs] Support row filters in full-text search

[zhuxiangyi]

Purpose

Full-text search on data-evolution tables cannot be combined with a row predicate today. Spark rejects it outright:

Full-text search does not support non-partition filters because full-text indexes
cannot apply row-id pre-filters before top-k ranking.

So the most common retrieval pattern — top-k documents for this query, among rows where category = 'x' / dt >= ... — forces users to over-fetch (limit × N) and filter on the engine side. That is slow and also wrong: the top-k is computed over the whole table, so rows that satisfy the filter but rank below k are silently dropped.

The reason given in the message no longer holds. NativeFullTextGlobalIndexReader already accepts a row-id bitmap (FullTextSearch.includeRowIds) and both read paths already pass one for deletion vectors and index coverage. Vector search already resolves withFilter(Predicate) through scalar global indexes into exactly such a bitmap (AbstractDataEvolutionVectorRead.preFilters). This PR closes the gap by wiring user predicates into the same mechanism, mirroring the vector design so users learn one rule.

What changes

Public API

• FullTextSearchBuilder.withFilter(Predicate). Partition predicates in the filter are extracted and applied as partition filters; the remaining predicates are evaluated before top-k ranking, so the result is the top-k among matching rows. A default implementation throws, keeping third-party builders source compatible (same pattern as VectorSearchBuilder.withOptions).
• HybridSearchBuilder.withFilter now reaches full-text routes. Previously the filter was applied to vector routes only, so a filtered hybrid query fused filtered and unfiltered candidates.

Scan (DataEvolutionFullTextScan)

• Scalar global index files (btree / bitmap / multivalue / fm) on the filtered columns are collected next to the full-text files. Each IndexFullTextSearchSplit carries the scalar files intersecting its range (scalarIndexFiles, serialized; split version bumped to 2, version 1 still deserializes).
• Rows whose filter columns are not covered by a scalar index are routed to the raw split following scalar-index.search-mode; in full-text-index.search-mode=fast those ranges are additionally kept inside the full-text coverage. RawFullTextSearchSplit carries the scalar files intersecting the raw ranges.

Read (DataEvolutionFullTextRead, RawFullTextReadImpl)

• The filter is resolved once per read through DataEvolutionGlobalIndexScanner into a row-id bitmap and AND-ed into every split's include set, next to the deletion-vector bitmap. The native call is unchanged, so limit is filter-then-rank inside the engine without over-fetching.
• The raw path bounds its scan with the same indexes (rawPreFilter, as vector search does), reads only the text column plus the filter columns, and evaluates the predicate row by row with executeFilter().
• In fast mode a filter no scalar index can evaluate excludes the indexed rows and logs a WARN once per query.

Coverage semantics (documented in full-text.mdx)

scalar-index.search-mode	filter columns indexed	filter columns not indexed
fast (default)	evaluated through the index	excluded, warning logged
full / detail	evaluated through the index	read raw and filtered row by row

A partially indexed conjunction (indexed = 1 AND unindexed = 2) is narrowed by the indexed members alone in fast mode, so the candidate set is a superset and the engine's row-level filter still applies (same contract as vector search); full mode is exact. Both are covered by tests.

Spark

• full_text_search(...) and hybrid_search(...) accept WHERE clauses on non-partition columns; PaimonBaseScan passes the pushed data filters to the builder instead of throwing. Predicates Spark cannot push down are applied after the search as for any scan.

Deliberately unchanged (follow-ups)

• Primary-key full-text search still rejects row filters, now with an explicit UnsupportedOperationException from the builder. It needs the per-file position mapping of PrimaryKeyFullTextBucketSearch and the DV precondition PrimaryKeyVectorScan has; separate PR.
• Flink sys.full_text_search and PyPaimon do not expose the parameter yet; the core API is in place for both. pypaimon hybrid search keeps its current rejection.

Tests

paimon-core — FullTextSearchBuilderTest (+12, 41 total) and PrimaryKeyFullTextSearchTest (+1):

• filter-then-rank: rows 0-2 score 1.0, rows 3-5 score 0.5; id >= 3, limit 2 returns two of {3,4,5} with their original scores, never the higher-scoring rows
• AND / OR / IN / always-true / zero-match predicates; two withFilter calls accumulate
• fast mode without a scalar index excludes rows and produces no raw split; full mode routes them to the raw path and returns the exact top-k
• partial scalar coverage: indexed range served by the index, the rest by the raw path, in one query
• full-text fast + scalar full: the raw scan never leaves the full-text coverage
• raw scan bounded by the scalar pre-filter (RawFullTextSearchSplit.scalarIndexFiles non-empty)
• partially indexed conjunction is a superset in fast, exact in full; OR with an unevaluable branch
• combined with deletion vectors; multiple full-text and btree index ranges, each split carrying only its own scalar file; partition predicate extracted from withFilter
• scan attaches scalar files only for filtered columns and does not mistake a btree on the text column for a full-text index; equality on the text column via btree and a full-text query on the same column coexist
• split serialization round-trips scalarIndexFiles and equality
• primary-key definition rejects withFilter on scan and read; the same table's non-PK column still takes the data-evolution path
• hybrid search applies the filter to a full-text route

paimon-full-text — NativeFullTextRowFilterTest against the real native engine: 2,000 rows, 8 categories, btree on category; for every category, withFilter(category = c), limit 20 equals the unfiltered full ranking restricted to that category (same rows, same BM25 scores, same cutoff).

paimon-spark-ut — FullTextSearchTest (+7) and HybridSearchTest: WHERE on a btree column, a bitmap string column, a partition + data column, with deletion vectors, fast vs full mode, a non-pushable predicate (id % 2 = 1) alone and mixed with a pushable one, and a hybrid full-text route with WHERE id = 1 (replaces the former "rejects non-partition filters" test).

Regression: paimon-core table.source.* + globalindex.** (346 tests) and the full paimon-full-text module pass. NativeFullTextGlobalIndexReaderTest.testCloseAttemptsAllResourcesWhenReaderCloseThrowsError fails identically on clean master (Mockito cannot mock the final native class in this environment) and is unrelated.

Performance

NativeFullTextRowFilterTest.benchmarkRowFilterStrategies (manual, -DextraJavaTestArgs=-Dpaimon.benchmark=true), native engine, 200,000 rows, 100 categories, limit = 10, best of 10 after 3 warm-ups, Apple M-series:

strategy	best	avg
no filter (baseline)	312 ms	319 ms
withFilter via btree, 1% selective	6.4 ms	6.8 ms
withFilter via btree, dense bitmap (~99% of rows)	307 ms	319 ms
over-fetch limit × 100 + client-side filter (previous workaround)	311 ms	326 ms
withFilter on an unindexed column, scalar-index.search-mode=full (raw scan of 200k rows)	129 ms	133 ms

• Handing the bitmap to the native engine costs nothing measurable: the dense-filter case matches the baseline within noise, so there is no regression for filters that keep most rows.
• A selective filter is ~50× faster than the baseline because the engine only scores rows in the bitmap.
• The workaround was never cheaper than the baseline: it already fetched every candidate.

While benchmarking I found that DataEvolutionFullTextRead.eval() asks the native reader for candidateLimit(rowRangeStart, rowRangeEnd) — the whole range — rather than the user limit, and materializes every candidate before topK(limit). That is what dominates the 312 ms baseline. It predates this PR (#8652) and is left as is here; I will open a separate issue/PR since it touches the compound-query contract (testCompoundFullTextSearchUsesFullLeafCandidatesBeforeFinalTopK).

API and Format

• New public method FullTextSearchBuilder.withFilter(Predicate) with a throwing default.
• IndexFullTextSearchSplit Java serialization version 1 → 2 (appends scalarIndexFiles; version 1 streams are still read). RawFullTextSearchSplit gains custom serialization for scalarIndexFiles. Splits are transient planning objects, no stored format changes.
• No index file format, manifest, or table option changes.
• Behaviour change: Spark full-text / hybrid queries with non-partition WHERE clauses now succeed instead of throwing.

Documentation

• docs/docs/multimodal-table/global-index/full-text.mdx: new Row Filters section (semantics, coverage table, partial-conjunction note), Spark SQL and Java examples.
• docs/docs/multimodal-table/global-index/hybrid-search.mdx: new Row Filters section.
• docs/docs/primary-key-table/global-index.mdx: the row-predicate limitation now scoped to primary-key full text, linking to the new section.