[core] Support composite BTree prefix and range queries

docs/docs/multimodal-table/global-index/btree.mdx

@@ -165,18 +165,40 @@ The index stores typed tuples and one row-ID posting list for each distinct tupl
 Equality conditions on every key column use a composite point lookup, regardless
 of their order in the SQL predicate. This avoids expanding each column's posting
 list before intersecting them. Additional predicates remain row filters, and OR
-branches can each select a composite lookup. When several definitions match,
-selection prefers the one with more key columns.
+branches can each select a composite lookup.
 
-Eligible composite point queries honor `global-index.query-in-reader.enabled`.
+Composite indexes also support an equality prefix followed by a range on the next
+key column. For an index on `(category, item_number)`, the following conditions can
+use the index:
+
+| Condition | Index scan |
+| --- | --- |
+| `category = 'category-a'` | All tuples with this category prefix |
+| `category = 'category-a' AND item_number > 7` | A range within the category prefix |
+| `category = 'category-a' AND item_number BETWEEN 7 AND 9` | A range with inclusive endpoints within the prefix |
+| `category > 'category-a' AND item_number = 7` | A category range; the item-number condition remains a data filter |
+
+The range may use `>`, `>=`, `<`, `<=` or `BETWEEN`. Conditions after the first
+range column remain data filters. A condition only on `item_number` cannot use
+this composite index's left prefix.
+
+When several composite definitions match, selection prefers more bounded key
+columns, then more equality columns, then fewer total key columns. If only the
+leading column is bounded, a dedicated single-column index is preferred when
+its coverage includes the composite definition's coverage.
+
+Eligible composite queries honor `global-index.query-in-reader.enabled`.
 With reader evaluation enabled, supported index conditions narrow the candidate
 rows and remaining conditions are evaluated against the data.
 
 Composite indexes can coexist with single-column indexes on the same columns.
-Composite lookup currently requires equality on all key columns; conditions on
-only some key columns, ranges and IN predicates use available single-column
-indexes or ordinary scans. Vector and full-text search pre-filters use
-single-column indexes or their data fallback.
+Full-key equalities use point probes. Prefix and range scans use
+`btree-index.fallback-scan-max-size`, applied to the selected files in each index
+group. If any group exceeds the budget, the whole composite definition is
+declined and planning uses other indexes or an ordinary scan. Point probes are
+unaffected by this budget. Dedicated composite `IN` and `IS NULL` optimization is
+not supported yet. Vector and full-text search pre-filters use single-column
+indexes or their data fallback.
 
 In `full` and `detail` modes, rows outside the selected composite index's coverage
 remain eligible for data filtering. Calling the procedure again with the same

paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java

@@ -327,7 +327,7 @@ private CompletableFuture<Optional<GlobalIndexResult>> visitParallel(
         return visitSelectedFiles(selector.get(), visitor);
     }
 
-    private CompletableFuture<Optional<GlobalIndexResult>> visitSelectedFiles(
+    protected CompletableFuture<Optional<GlobalIndexResult>> visitSelectedFiles(
             Optional<List<GlobalIndexIOMeta>> selectedOpt,
             Function<R, Optional<GlobalIndexResult>> visitor) {
         if (!selectedOpt.isPresent()) {

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexerFactory.java

@@ -18,21 +18,15 @@
 
 package org.apache.paimon.globalindex.btree;
 
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.globalindex.CompositeKeySerializer;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
 import org.apache.paimon.globalindex.GlobalIndexer;
 import org.apache.paimon.globalindex.GlobalIndexerFactory;
 import org.apache.paimon.globalindex.KeySerializer;
 import org.apache.paimon.globalindex.SortedFileMetaSelector;
 import org.apache.paimon.options.Options;
-import org.apache.paimon.predicate.FieldRef;
-import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.RowType;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -50,25 +44,9 @@ public String identifier() {
     public List<GlobalIndexIOMeta> selectFiles(
             List<DataField> indexFields, Predicate predicate, List<GlobalIndexIOMeta> files) {
         if (indexFields.size() > 1) {
-            Optional<List<LeafPredicate>> matched =
-                    CompositeBTreePredicate.match(indexFields, predicate);
-            if (!matched.isPresent()) {
-                return files;
-            }
-            if (CompositeBTreePredicate.isContradictory(indexFields, predicate)) {
-                return Collections.emptyList();
-            }
-            if (files.stream().anyMatch(file -> file.metadata() == null)) {
-                return files;
-            }
-            RowType keyType = new RowType(indexFields);
-            KeySerializer serializer = new CompositeKeySerializer(keyType);
-            Object[] values = matched.get().stream().map(leaf -> leaf.literals().get(0)).toArray();
-            return new SortedFileMetaSelector(files, serializer)
-                    .visitEqual(
-                            new FieldRef(0, indexFields.get(0).name(), keyType),
-                            GenericRow.of(values))
-                    .orElse(files);
+            Optional<CompositeBTreePredicate.Plan> plan =
+                    CompositeBTreePredicate.plan(indexFields, predicate);
+            return plan.map(value -> value.selectFiles(files)).orElse(files);
         }
         return SortedFileMetaSelector.selectFiles(
                 predicate, files, KeySerializer.create(indexFields.get(0).type()));

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.globalindex.btree;
 
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
@@ -339,6 +340,38 @@ public Optional<GlobalIndexResult> visitEqual(Object literal) {
         return createResult(() -> pointQuery(literal));
     }
 
+    public Optional<GlobalIndexResult> visitComposite(CompositeBTreePredicate.Plan plan) {
+        return createResult(
+                () -> {
+                    if (plan.isEmpty()) {
+                        return new RoaringNavigableMap64();
+                    }
+                    if (plan.isPointLookup()) {
+                        return pointQuery(plan.pointKey());
+                    }
+                    RoaringNavigableMap64 result = new RoaringNavigableMap64();
+                    SstFileReader.SstFileIterator iterator = reader.createIterator();
+                    iterator.seekTo(
+                            key ->
+                                    plan.lower()
+                                            .compareKey(
+                                                    (InternalRow) keySerializer.deserialize(key)));
+                    BlockIterator batch;
+                    while ((batch = iterator.readBatch()) != null) {
+                        while (batch.hasNext()) {
+                            Map.Entry<MemorySlice, MemorySlice> entry = batch.next();
+                            InternalRow key =
+                                    (InternalRow) keySerializer.deserialize(entry.getKey());
+                            if (plan.upper().compareKey(key) > 0) {
+                                return result;
+                            }
+                            addRowIdsTo(entry.getValue(), result);
+                        }
+                    }
+                    return result;
+                });
+    }
+
     public Optional<GlobalIndexResult> visitGreaterThan(Object literal) {
         return createResult(() -> rangeQuery(literal, maxKey, false, true));
     }