Implement ORDER BY BM25

src/antlr/Lexer.g

@@ -227,6 +227,7 @@ K_DROPPED:     D R O P P E D;
 K_COLUMN:      C O L U M N;
 K_RECORD:      R E C O R D;
 K_ANN_OF:      A N N WS+ O F;
+K_BM25_OF:     'BM25' WS+ 'OF';
 
 // Case-insensitive alpha characters
 fragment A: ('a'|'A');

src/antlr/Parser.g

@@ -457,14 +457,18 @@ customIndexExpression [WhereClause.Builder clause]
     ;
 
 orderByClause[List<Ordering.Raw> orderings]
-    @init{
+    @init {
         Ordering.Direction direction = Ordering.Direction.ASC;
+        Ordering.Raw.Expression expr = null;
     }
-    : c=cident (K_ANN_OF t=term)? (K_ASC | K_DESC { direction = Ordering.Direction.DESC; })?
+    : c=cident
+        ( K_ANN_OF t=term { expr = new Ordering.Raw.Ann(c, t); }
+        | K_BM25_OF t=term { expr = new Ordering.Raw.Bm25(c, t); }
+        )?
+        (K_ASC | K_DESC { direction = Ordering.Direction.DESC; })?
     {
-        Ordering.Raw.Expression expr = (t == null)
-            ? new Ordering.Raw.SingleColumn(c)
-            : new Ordering.Raw.Ann(c, t);
+        if (expr == null)
+            expr = new Ordering.Raw.SingleColumn(c);
         orderings.add(new Ordering.Raw(expr, direction));
     }
     ;
@@ -1967,6 +1971,7 @@ basic_unreserved_keyword returns [String str]
         | K_COLUMN
         | K_RECORD
         | K_ANN_OF
+        | K_BM25_OF
         | K_OFFSET
         ) { $str = $k.text; }
     ;

src/java/org/apache/cassandra/cql3/GeoDistanceRelation.java

@@ -141,6 +141,12 @@ protected Restriction newAnnRestriction(TableMetadata table, VariableSpecificati
         throw invalidRequest("%s cannot be used with the GEO_DISTANCE function", operator());
     }
 
+    @Override
+    protected Restriction newBm25Restriction(TableMetadata table, VariableSpecifications boundNames)
+    {
+        throw invalidRequest("%s cannot be used with the GEO_DISTANCE function", operator());
+    }
+
     @Override
     protected Restriction newAnalyzerMatchesRestriction(TableMetadata table, VariableSpecifications boundNames)
     {

src/java/org/apache/cassandra/cql3/MultiColumnRelation.java

@@ -250,6 +250,12 @@ protected Restriction newAnnRestriction(TableMetadata table, VariableSpecificati
         throw invalidRequest("%s cannot be used for multi-column relations", operator());
     }
 
+    @Override
+    protected Restriction newBm25Restriction(TableMetadata table, VariableSpecifications boundNames)
+    {
+        throw invalidRequest("%s cannot be used for multi-column relations", operator());
+    }
+
     @Override
     protected Restriction newAnalyzerMatchesRestriction(TableMetadata table, VariableSpecifications boundNames)
     {

src/java/org/apache/cassandra/cql3/Operator.java

@@ -373,6 +373,20 @@ public boolean isSatisfiedBy(AbstractType<?> type, ByteBuffer leftOperand, ByteB
             return !LIKE.isSatisfiedBy(type, leftOperand, rightOperand, analyzer);
         }
     },
+    BM25(25)
+    {
+        @Override
+        public String toString()
+        {
+            return "BM25";
+        }
+
+        @Override
+        public boolean isSatisfiedBy(AbstractType<?> type, ByteBuffer leftOperand, ByteBuffer rightOperand, @Nullable Index.Analyzer analyzer)
+        {
+            throw new UnsupportedOperationException();
+        }
+    },
 
     /**
      * An operator that only performs matching against analyzed columns.

src/java/org/apache/cassandra/cql3/Ordering.java

@@ -20,6 +20,7 @@
 
 import org.apache.cassandra.cql3.restrictions.SingleColumnRestriction;
 import org.apache.cassandra.cql3.restrictions.SingleRestriction;
+import org.apache.cassandra.cql3.statements.SelectStatement;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 
@@ -48,6 +49,11 @@ public interface Expression
         SingleRestriction toRestriction();
 
         ColumnMetadata getColumn();
+
+        default boolean isScored()
+        {
+            return false;
+        }
     }
 
     /**
@@ -118,6 +124,54 @@ public ColumnMetadata getColumn()
         {
             return column;
         }
+
+        @Override
+        public boolean isScored()
+        {
+            return SelectStatement.ANN_USE_SYNTHETIC_SCORE;
+        }
+    }
+
+    /**
+     * An expression used in BM25 ordering.
+     * <code>ORDER BY column BM25 OF value</code>
+     */
+    public static class Bm25 implements Expression
+    {
+        final ColumnMetadata column;
+        final Term queryValue;
+        final Direction direction;
+
+        public Bm25(ColumnMetadata column, Term queryValue, Direction direction)
+        {
+            this.column = column;
+            this.queryValue = queryValue;
+            this.direction = direction;
+        }
+
+        @Override
+        public boolean hasNonClusteredOrdering()
+        {
+            return true;
+        }
+
+        @Override
+        public SingleRestriction toRestriction()
+        {
+            return new SingleColumnRestriction.Bm25Restriction(column, queryValue);
+        }
+
+        @Override
+        public ColumnMetadata getColumn()
+        {
+            return column;
+        }
+
+        @Override
+        public boolean isScored()
+        {
+            return true;
+        }
     }
 
     public enum Direction
@@ -190,6 +244,27 @@ public Ordering.Expression bind(TableMetadata table, VariableSpecifications boun
                 return new Ordering.Ann(column, value, direction);
             }
         }
+
+        public static class Bm25 implements Expression
+        {
+            final ColumnIdentifier columnId;
+            final Term.Raw queryValue;
+
+            Bm25(ColumnIdentifier column, Term.Raw queryValue)
+            {
+                this.columnId = column;
+                this.queryValue = queryValue;
+            }
+
+            @Override
+            public Ordering.Expression bind(TableMetadata table, VariableSpecifications boundNames, Direction direction)
+            {
+                ColumnMetadata column = table.getExistingColumn(columnId);
+                Term value = queryValue.prepare(table.keyspace, column);
+                value.collectMarkerSpecification(boundNames);
+                return new Ordering.Bm25(column, value, direction);
+            }
+        }
     }
 }
 

src/java/org/apache/cassandra/cql3/Relation.java

@@ -202,6 +202,8 @@ public final Restriction toRestriction(TableMetadata table, VariableSpecificatio
                 return newLikeRestriction(table, boundNames, relationType);
             case ANN:
                 return newAnnRestriction(table, boundNames);
+            case BM25:
+                return newBm25Restriction(table, boundNames);
             case ANALYZER_MATCHES:
                 return newAnalyzerMatchesRestriction(table, boundNames);
             default: throw invalidRequest("Unsupported \"!=\" relation: %s", this);
@@ -296,6 +298,11 @@ protected abstract Restriction newSliceRestriction(TableMetadata table,
      */
     protected abstract Restriction newAnnRestriction(TableMetadata table, VariableSpecifications boundNames);
 
+    /**
+     * Creates a new BM25 restriction instance.
+     */
+    protected abstract Restriction newBm25Restriction(TableMetadata table, VariableSpecifications boundNames);
+
     /**
      * Creates a new Analyzer Matches restriction instance.
      */

src/java/org/apache/cassandra/cql3/SingleColumnRelation.java

@@ -333,6 +333,14 @@ protected Restriction newAnnRestriction(TableMetadata table, VariableSpecificati
         return new SingleColumnRestriction.AnnRestriction(columnDef, term);
     }
 
+    @Override
+    protected Restriction newBm25Restriction(TableMetadata table, VariableSpecifications boundNames)
+    {
+        ColumnMetadata columnDef = table.getExistingColumn(entity);
+        Term term = toTerm(toReceivers(columnDef), value, table.keyspace, boundNames);
+        return new SingleColumnRestriction.AnnRestriction(columnDef, term);
+    }
+
     @Override
     protected Restriction newAnalyzerMatchesRestriction(TableMetadata table, VariableSpecifications boundNames)
     {

src/java/org/apache/cassandra/cql3/TokenRelation.java

@@ -138,7 +138,13 @@ protected Restriction newLikeRestriction(TableMetadata table, VariableSpecificat
     @Override
     protected Restriction newAnnRestriction(TableMetadata table, VariableSpecifications boundNames)
     {
-        throw invalidRequest("%s cannot be used for toekn relations", operator());
+        throw invalidRequest("%s cannot be used for token relations", operator());
+    }
+
+    @Override
+    protected Restriction newBm25Restriction(TableMetadata table, VariableSpecifications boundNames)
+    {
+        throw invalidRequest("%s cannot be used for token relations", operator());
     }
 
     @Override

src/java/org/apache/cassandra/cql3/restrictions/SingleColumnRestriction.java

@@ -1191,6 +1191,74 @@ public boolean isBoundedAnn()
         }
     }
 
+    public static final class Bm25Restriction extends SingleColumnRestriction
+    {
+        private final Term value;
+
+        public Bm25Restriction(ColumnMetadata columnDef, Term value)
+        {
+            super(columnDef);
+            this.value = value;
+        }
+
+        public ByteBuffer value(QueryOptions options)
+        {
+            return value.bindAndGet(options);
+        }
+
+        @Override
+        public void addFunctionsTo(List<Function> functions)
+        {
+            value.addFunctionsTo(functions);
+        }
+
+        @Override
+        MultiColumnRestriction toMultiColumnRestriction()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void addToRowFilter(RowFilter.Builder filter,
+                                   IndexRegistry indexRegistry,
+                                   QueryOptions options)
+        {
+            filter.add(columnDef, Operator.BM25, value.bindAndGet(options));
+        }
+
+        @Override
+        public MultiClusteringBuilder appendTo(MultiClusteringBuilder builder, QueryOptions options)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public String toString()
+        {
+            return String.format("BM25(%s)", value);
+        }
+
+        @Override
+        public SingleRestriction doMergeWith(SingleRestriction otherRestriction)
+        {
+            if (otherRestriction.isIndexBasedOrdering())
+                throw invalidRequest("%s cannot be restricted by multiple BM25 restrictions", columnDef.name);
+            throw invalidRequest("%s cannot be restricted by both BM25 and %s", columnDef.name, otherRestriction.toString());
+        }
+
+        @Override
+        protected boolean isSupportedBy(Index index)
+        {
+            return index.supportsExpression(columnDef, Operator.BM25);
+        }
+
+        @Override
+        public boolean isIndexBasedOrdering()
+        {
+            return true;
+        }
+    }
+
     /**
      * A Bounded ANN Restriction is one that uses a similarity score as the limiting factor for ANN instead of a number
      * of results.

src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java

@@ -682,7 +682,8 @@ else if (indexOrderings.size() == 1)
                 if (orderings.size() > 1)
                     throw new InvalidRequestException("Cannot combine clustering column ordering with non-clustering column ordering");
                 Ordering ordering = indexOrderings.get(0);
-                if (ordering.direction != Ordering.Direction.ASC && ordering.expression instanceof Ordering.Ann)
+                // TODO remove the instanceof with SelectStatement.ANN_USE_SYNTHETIC_SCORE.
+                if (ordering.direction != Ordering.Direction.ASC && (ordering.expression.isScored() || ordering.expression instanceof Ordering.Ann))
                     throw new InvalidRequestException("Descending ANN ordering is not supported");
                 if (!ENABLE_SAI_GENERAL_ORDER_BY && ordering.expression instanceof Ordering.SingleColumn)
                     throw new InvalidRequestException("SAI based ORDER BY on non-vector column is not supported");

src/java/org/apache/cassandra/cql3/selection/ColumnFilterFactory.java

@@ -38,9 +38,21 @@ abstract class ColumnFilterFactory
      */
     abstract ColumnFilter newInstance(List<Selector> selectors);
 
-    public static ColumnFilterFactory wildcard(TableMetadata table)
+    public static ColumnFilterFactory wildcard(TableMetadata table, Set<ColumnMetadata> orderingColumns)
     {
-        return new PrecomputedColumnFilter(ColumnFilter.all(table));
+        ColumnFilter cf;
+        if (orderingColumns.isEmpty())
+        {
+            cf = ColumnFilter.all(table);
+        }
+        else
+        {
+            ColumnFilter.Builder builder = ColumnFilter.selectionBuilder();
+            builder.addAll(table.regularAndStaticColumns());
+            builder.addAll(orderingColumns);
+            cf = builder.build();
+        }
+        return new PrecomputedColumnFilter(cf);
     }
 
     public static ColumnFilterFactory fromColumns(TableMetadata table,