From 5f5eb0e41205405e889a373dccc011ad037ca01e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 Dec 2024 14:31:50 -0600 Subject: [PATCH] handle eq in an analyzed index by transforming it into a Match restriction in SingleColumnRelation.newEQRestriction. This eliminates the need for skipMerge and special cases in doMergeWith, and moves the issuing of warnings next to the place where the transformation occurs instead of doing it much later in RowFilterValidator (which is no longer needed) --- .../cassandra/cql3/SingleColumnRelation.java | 13 ++- .../ClusteringColumnRestrictions.java | 4 +- .../PartitionKeySingleRestrictionSet.java | 2 +- .../cql3/restrictions/RestrictionSet.java | 36 +++--- .../restrictions/SingleColumnRestriction.java | 24 +--- .../restrictions/StatementRestrictions.java | 4 +- .../org/apache/cassandra/db/ReadCommand.java | 2 - .../apache/cassandra/index/IndexRegistry.java | 25 ++--- .../cassandra/index/RowFilterValidator.java | 103 ------------------ .../index/SecondaryIndexManager.java | 6 - .../apache/cassandra/service/ClientWarn.java | 19 +++- 11 files changed, 62 insertions(+), 176 deletions(-) delete mode 100644 src/java/org/apache/cassandra/index/RowFilterValidator.java diff --git a/src/java/org/apache/cassandra/cql3/SingleColumnRelation.java b/src/java/org/apache/cassandra/cql3/SingleColumnRelation.java index a15251bdefa7..89ae1d38b8a3 100644 --- a/src/java/org/apache/cassandra/cql3/SingleColumnRelation.java +++ b/src/java/org/apache/cassandra/cql3/SingleColumnRelation.java @@ -23,6 +23,8 @@ import java.util.Objects; import org.apache.cassandra.db.marshal.VectorType; +import org.apache.cassandra.index.IndexRegistry; +import org.apache.cassandra.index.sai.analyzer.AnalyzerEqOperatorSupport; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.cql3.Term.Raw; @@ -33,6 +35,7 @@ import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.MapType; import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.service.ClientWarn; import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse; import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue; @@ -191,7 +194,15 @@ protected Restriction newEQRestriction(TableMetadata table, VariableSpecificatio if (mapKey == null) { Term term = toTerm(toReceivers(columnDef), value, table.keyspace, boundNames); - return new SingleColumnRestriction.EQRestriction(columnDef, term); + var analyzedIndex = IndexRegistry.obtain(table).supportsAnalyzedEq(columnDef); + if (analyzedIndex == null) + return new SingleColumnRestriction.EQRestriction(columnDef, term); + + ClientWarn.instance.warn(String.format(AnalyzerEqOperatorSupport.EQ_RESTRICTION_ON_ANALYZED_WARNING, + columnDef.toString(), + analyzedIndex.getIndexMetadata().name), + columnDef); + return new SingleColumnRestriction.AnalyzerMatchesRestriction(columnDef, term); } List receivers = toReceivers(columnDef); Term entryKey = toTerm(Collections.singletonList(receivers.get(0)), mapKey, table.keyspace, boundNames); diff --git a/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java index 092cc93cc7b7..0b33dfb16f69 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java @@ -189,7 +189,7 @@ public ClusteringColumnRestrictions.Builder addRestriction(Restriction restricti SingleRestriction lastRestriction = restrictions.lastRestriction(); ColumnMetadata lastRestrictionStart = lastRestriction.getFirstColumn(); ColumnMetadata newRestrictionStart = newRestriction.getFirstColumn(); - restrictions.addRestriction(newRestriction, isDisjunction, indexRegistry); + restrictions.addRestriction(newRestriction, isDisjunction); checkFalse(lastRestriction.isSlice() && newRestrictionStart.position() > lastRestrictionStart.position(), "Clustering column \"%s\" cannot be restricted (preceding column \"%s\" is restricted by a non-EQ relation)", @@ -203,7 +203,7 @@ public ClusteringColumnRestrictions.Builder addRestriction(Restriction restricti } else { - restrictions.addRestriction(newRestriction, isDisjunction, indexRegistry); + restrictions.addRestriction(newRestriction, isDisjunction); } return this; diff --git a/src/java/org/apache/cassandra/cql3/restrictions/PartitionKeySingleRestrictionSet.java b/src/java/org/apache/cassandra/cql3/restrictions/PartitionKeySingleRestrictionSet.java index 50e6ea481f09..faddd9d9ff06 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/PartitionKeySingleRestrictionSet.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/PartitionKeySingleRestrictionSet.java @@ -189,7 +189,7 @@ public PartitionKeyRestrictions build(IndexRegistry indexRegistry, boolean isDis if (restriction.isOnToken()) return buildWithTokens(restrictionSet, i, indexRegistry); - restrictionSet.addRestriction((SingleRestriction) restriction, isDisjunction, indexRegistry); + restrictionSet.addRestriction((SingleRestriction) restriction, isDisjunction); } return buildPartitionKeyRestrictions(restrictionSet); diff --git a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java index 92736c0d298a..f9dd22756fe0 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java @@ -413,45 +413,35 @@ private Builder() { } - public void addRestriction(SingleRestriction restriction, boolean isDisjunction, IndexRegistry indexRegistry) + public void addRestriction(SingleRestriction restriction, boolean isDisjunction) { List columnDefs = restriction.getColumnDefs(); if (isDisjunction) { // If this restriction is part of a disjunction query then we don't want - // to merge the restrictions (if that is possible), we just add the - // restriction to the set of restrictions for the column. + // to merge the restrictions, we just add the new restriction addRestrictionForColumns(columnDefs, restriction, null); } else { - // In some special cases such as EQ in analyzed index we need to skip merging the restriction, - // so we can send multiple EQ restrictions to the index. - if (restriction.skipMerge(indexRegistry)) + // ANDed together restrictions against the same columns should be merged. + Set existingRestrictions = getRestrictions(newRestrictions, columnDefs); + // Trivial case of no existing restrictions + if (existingRestrictions.isEmpty()) { addRestrictionForColumns(columnDefs, restriction, null); return; } + // Since we merge new restrictions into the existing ones at each pass, there should only be + // at most one existing restriction across the same columnDefs + assert existingRestrictions.size() == 1 : existingRestrictions; - // If this restriction isn't part of a disjunction then we need to get - // the set of existing restrictions for the column and merge them with the - // new restriction - Set existingRestrictions = getRestrictions(newRestrictions, columnDefs); - - SingleRestriction merged = restriction; - Set replacedRestrictions = new HashSet<>(); - - for (SingleRestriction existing : existingRestrictions) - { - if (!existing.skipMerge(indexRegistry)) - { - merged = existing.mergeWith(merged); - replacedRestrictions.add(existing); - } - } + // Perform the merge + SingleRestriction existing = existingRestrictions.iterator().next(); + var merged = existing.mergeWith(restriction); - addRestrictionForColumns(merged.getColumnDefs(), merged, replacedRestrictions); + addRestrictionForColumns(merged.getColumnDefs(), merged, Set.of(existing)); } } diff --git a/src/java/org/apache/cassandra/cql3/restrictions/SingleColumnRestriction.java b/src/java/org/apache/cassandra/cql3/restrictions/SingleColumnRestriction.java index 1cde8cfc00e4..ced7f9d31cdb 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/SingleColumnRestriction.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/SingleColumnRestriction.java @@ -198,24 +198,6 @@ public String toString() return String.format("EQ(%s)", term); } - @Override - public boolean skipMerge(IndexRegistry indexRegistry) - { - // We should skip merging this EQ if there is an analyzed index for this column that supports EQ, - // so there can be multiple EQs for the same column. - - if (indexRegistry == null) - return false; - - for (Index index : indexRegistry.listIndexes()) - { - if (index.supportsExpression(columnDef, Operator.ANALYZER_MATCHES) && - index.supportsExpression(columnDef, Operator.EQ)) - return true; - } - return false; - } - @Override public SingleRestriction doMergeWith(SingleRestriction otherRestriction) { @@ -1402,10 +1384,12 @@ public String toString() @Override public SingleRestriction doMergeWith(SingleRestriction otherRestriction) { - if (!(otherRestriction.isAnalyzerMatches())) + if (!otherRestriction.isAnalyzerMatches()) throw invalidRequest(CANNOT_BE_MERGED_ERROR, columnDef.name); - List otherValues = ((AnalyzerMatchesRestriction) otherRestriction).getValues(); + List otherValues = otherRestriction instanceof AnalyzerMatchesRestriction + ? ((AnalyzerMatchesRestriction) otherRestriction).getValues() + : List.of(((EQRestriction) otherRestriction).term); List newValues = new ArrayList<>(values.size() + otherValues.size()); newValues.addAll(values); newValues.addAll(otherValues); diff --git a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java index 7e92d3273077..4d1b4bb68a09 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java @@ -442,7 +442,7 @@ else if (def.isClusteringColumn() && nestingLevel == 0) } else { - nonPrimaryKeyRestrictionSet.addRestriction((SingleRestriction) restriction, element.isDisjunction(), indexRegistry); + nonPrimaryKeyRestrictionSet.addRestriction((SingleRestriction) restriction, element.isDisjunction()); } } } @@ -699,7 +699,7 @@ else if (indexOrderings.size() == 1) throw new InvalidRequestException(String.format(NON_CLUSTER_ORDERING_REQUIRES_INDEX_MESSAGE, restriction.getFirstColumn())); } - receiver.addRestriction(restriction, false, indexRegistry); + receiver.addRestriction(restriction, false); } } diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index a289987a1adc..c8b38695e004 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -383,8 +383,6 @@ static Index.QueryPlan findIndexQueryPlan(TableMetadata table, RowFilter rowFilt @Override public void maybeValidateIndexes() { - IndexRegistry.obtain(metadata()).validate(rowFilter()); - if (null != indexQueryPlan) indexQueryPlan.validate(this); } diff --git a/src/java/org/apache/cassandra/index/IndexRegistry.java b/src/java/org/apache/cassandra/index/IndexRegistry.java index cc6b0d103ea9..ec41a068d9ac 100644 --- a/src/java/org/apache/cassandra/index/IndexRegistry.java +++ b/src/java/org/apache/cassandra/index/IndexRegistry.java @@ -102,12 +102,6 @@ public Optional getBestIndexFor(RowFilter.Expression expression) public void validate(PartitionUpdate update) { } - - @Override - public void validate(RowFilter filter) - { - // no-op since it's an empty registry - } }; /** @@ -295,12 +289,6 @@ public Optional getBestIndexFor(RowFilter.Expression expression) public void validate(PartitionUpdate update) { } - - @Override - public void validate(RowFilter filter) - { - // no-op since it's an empty registry - } }; default void registerIndex(Index index) @@ -341,8 +329,6 @@ default Optional getAnalyzerFor(ColumnMetadata column, Operator */ void validate(PartitionUpdate update); - void validate(RowFilter filter); - /** * Returns the {@code IndexRegistry} associated to the specified table. * @@ -356,4 +342,15 @@ public static IndexRegistry obtain(TableMetadata table) return table.isVirtual() ? EMPTY : Keyspace.openAndGetStore(table).indexManager; } + + default Index supportsAnalyzedEq(ColumnMetadata cm) + { + for (Index index : listIndexes()) + { + if (index.supportsExpression(cm, Operator.ANALYZER_MATCHES) && + index.supportsExpression(cm, Operator.EQ)) + return index; + } + return null; + } } diff --git a/src/java/org/apache/cassandra/index/RowFilterValidator.java b/src/java/org/apache/cassandra/index/RowFilterValidator.java deleted file mode 100644 index fb70fbfc1452..000000000000 --- a/src/java/org/apache/cassandra/index/RowFilterValidator.java +++ /dev/null @@ -1,103 +0,0 @@ -/* - * Copyright DataStax, Inc. - * - * Licensed 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.cassandra.index; - -import java.util.HashSet; -import java.util.Set; -import java.util.StringJoiner; - -import org.apache.cassandra.cql3.Operator; -import org.apache.cassandra.db.filter.RowFilter; -import org.apache.cassandra.index.sai.analyzer.AnalyzerEqOperatorSupport; -import org.apache.cassandra.schema.ColumnMetadata; -import org.apache.cassandra.service.ClientWarn; - -/** - * Class for validating the index-related aspects of a {@link RowFilter}, without considering what index is actually used. - *

- * It will emit a client warning when a query has EQ restrictions on columns having an analyzed index. - */ -class RowFilterValidator -{ - private final Iterable allIndexes; - - private Set columns; - private Set indexes; - - private RowFilterValidator(Iterable allIndexes) - { - this.allIndexes = allIndexes; - } - - private void addEqRestriction(ColumnMetadata column) - { - for (Index index : allIndexes) - { - if (index.supportsExpression(column, Operator.EQ) && - index.supportsExpression(column, Operator.ANALYZER_MATCHES)) - { - if (columns == null) - columns = new HashSet<>(); - columns.add(column); - - if (indexes == null) - indexes = new HashSet<>(); - indexes.add(index); - } - } - } - - private void validate() - { - if (columns == null || indexes == null) - return; - - StringJoiner columnNames = new StringJoiner(", "); - StringJoiner indexNames = new StringJoiner(", "); - columns.forEach(column -> columnNames.add(column.name.toString())); - indexes.forEach(index -> indexNames.add(index.getIndexMetadata().name)); - - ClientWarn.instance.warn(String.format(AnalyzerEqOperatorSupport.EQ_RESTRICTION_ON_ANALYZED_WARNING, columnNames, indexNames)); - } - - /** - * Emits a client warning if the filter contains EQ restrictions on columns having an analyzed index. - * - * @param filter the filter to validate - * @param indexes the existing indexes - */ - public static void validate(RowFilter filter, Iterable indexes) - { - RowFilterValidator validator = new RowFilterValidator(indexes); - validate(filter.root(), validator); - validator.validate(); - } - - private static void validate(RowFilter.FilterElement element, RowFilterValidator validator) - { - for (RowFilter.Expression expression : element.expressions()) - { - if (expression.operator() == Operator.EQ) - validator.addEqRestriction(expression.column()); - } - - for (RowFilter.FilterElement child : element.children()) - { - validate(child, validator); - } - } -} diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java index 9954d1b3ee1c..21d633b30a39 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java +++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java @@ -1277,12 +1277,6 @@ public void validate(PartitionUpdate update) throws InvalidRequestException index.validate(update); } - @Override - public void validate(RowFilter filter) - { - RowFilterValidator.validate(filter, indexes.values()); - } - /* * IndexRegistry methods */ diff --git a/src/java/org/apache/cassandra/service/ClientWarn.java b/src/java/org/apache/cassandra/service/ClientWarn.java index 5a6a878681e1..38570a06d2b8 100644 --- a/src/java/org/apache/cassandra/service/ClientWarn.java +++ b/src/java/org/apache/cassandra/service/ClientWarn.java @@ -18,7 +18,9 @@ package org.apache.cassandra.service; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import io.netty.util.concurrent.FastThreadLocal; import org.apache.cassandra.concurrent.ExecutorLocal; @@ -45,10 +47,18 @@ public void set(State value) } public void warn(String text) + { + warn(text, null); + } + + /** + * Issue the given warning if this is the first time `key` is seen. + */ + public void warn(String text, Object key) { State state = warnLocal.get(); if (state != null) - state.add(text); + state.add(text, key); } public void captureWarnings() @@ -72,11 +82,16 @@ public void resetWarnings() public static class State { private final List warnings = new ArrayList<>(); + private final Set keysAdded = new HashSet<>(); - private void add(String warning) + private void add(String warning, Object key) { if (warnings.size() < FBUtilities.MAX_UNSIGNED_SHORT) + { + if (key != null && !keysAdded.add(key)) + return; warnings.add(maybeTruncate(warning)); + } } private static String maybeTruncate(String warning)