Add PrimaryIdTable as a virtual table for querying SSTable metadata

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

@@ -77,7 +77,6 @@ public ClusteringColumnRestrictions mergeWith(Restriction restriction, @Nullable
 
             ColumnMetadata lastRestrictionStart = lastRestriction.firstColumn();
             ColumnMetadata newRestrictionStart = restriction.firstColumn();
-
             checkFalse(lastRestriction.isSlice() && newRestrictionStart.position() > lastRestrictionStart.position(),
                        "Clustering column \"%s\" cannot be restricted (preceding column \"%s\" is restricted by a non-EQ relation)",
                        newRestrictionStart.name,

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

@@ -345,7 +345,7 @@ else if (operator.requiresIndexing())
             }
             else
             {
-                if (!allowFiltering && requiresAllowFilteringIfNotSpecified())
+                if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table))
                     throw invalidRequest(allowFilteringMessage(state));
             }
 
@@ -356,12 +356,12 @@ else if (operator.requiresIndexing())
             validateSecondaryIndexSelections();
     }
 
-    public boolean requiresAllowFilteringIfNotSpecified()
+    public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata)
     {
-        if (!table.isVirtual())
+        if (!metadata.isVirtual())
             return true;
 
-        VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(table.id);
+        VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(metadata.id);
         assert tableNullable != null;
         return !tableNullable.allowFilteringImplicitly();
     }
@@ -568,7 +568,7 @@ private void processPartitionKeyRestrictions(ClientState state, boolean hasQueri
             // components must have a EQ. Only the last partition key component can be in IN relation.
             if (partitionKeyRestrictions.needFiltering())
             {
-                if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified())
+                if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified(table))
                     throw new InvalidRequestException(allowFilteringMessage(state));
 
                 isKeyRange = true;

src/java/org/apache/cassandra/cql3/statements/SelectStatement.java

@@ -88,6 +88,7 @@
 import org.apache.commons.lang3.builder.ToStringStyle;
 
 import static java.lang.String.format;
+import static org.apache.cassandra.cql3.restrictions.StatementRestrictions.requiresAllowFilteringIfNotSpecified;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
@@ -1357,7 +1358,7 @@ private StatementRestrictions prepareRestrictions(ClientState state,
                                              boundNames,
                                              orderings,
                                              selectsOnlyStaticColumns,
-                                             parameters.allowFiltering,
+                                             parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata),
                                              forView);
         }
 
@@ -1583,7 +1584,7 @@ private void checkNeedsFiltering(TableMetadata table, StatementRestrictions rest
             {
                 // We will potentially filter data if the row filter is not the identity and there isn't any index group
                 // supporting all the expressions in the filter.
-                if (restrictions.requiresAllowFilteringIfNotSpecified())
+                if (requiresAllowFilteringIfNotSpecified(table))
                     checkFalse(restrictions.needFiltering(table), StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE);
             }
         }

src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java

@@ -565,7 +565,7 @@ public PartitionIterator execute(ConsistencyLevel consistency, ClientState state
         public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
         {
             VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
-            UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter());
+            UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter(), rowFilter());
             return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
         }
 

src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java

@@ -1395,7 +1395,7 @@ public PartitionIterator execute(ConsistencyLevel consistency, ClientState state
         public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
         {
             VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
-            UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter());
+            UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter(), rowFilter());
             return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
         }
 

src/java/org/apache/cassandra/db/virtual/AbstractVirtualTable.java

@@ -29,6 +29,7 @@
 import org.apache.cassandra.db.PartitionPosition;
 import org.apache.cassandra.db.filter.ClusteringIndexFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.filter.RowFilter;
 import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator;
@@ -75,7 +76,7 @@ public DataSet data(DecoratedKey partitionKey)
     }
 
     @Override
-    public final UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter)
+    public final UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter)
     {
         Partition partition = data(partitionKey).getPartition(partitionKey);
 
@@ -88,7 +89,7 @@ public final UnfilteredPartitionIterator select(DecoratedKey partitionKey, Clust
     }
 
     @Override
-    public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter)
+    public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
     {
         DataSet data = data();
 

src/java/org/apache/cassandra/db/virtual/CollectionVirtualTableAdapter.java

@@ -51,6 +51,7 @@
 import org.apache.cassandra.db.EmptyIterators;
 import org.apache.cassandra.db.filter.ClusteringIndexFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.filter.RowFilter;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.BooleanType;
 import org.apache.cassandra.db.marshal.ByteType;
@@ -307,7 +308,8 @@ else if (partitionKeyTypes.size() > 1)
     @Override
     public UnfilteredPartitionIterator select(DecoratedKey partitionKey,
                                               ClusteringIndexFilter clusteringFilter,
-                                              ColumnFilter columnFilter)
+                                              ColumnFilter columnFilter,
+                                              RowFilter rowFilter)
     {
         if (!data.iterator().hasNext())
             return EmptyIterators.unfilteredPartition(metadata);
@@ -348,7 +350,7 @@ public UnfilteredPartitionIterator select(DecoratedKey partitionKey,
     }
 
     @Override
-    public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter)
+    public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
     {
         return createPartitionIterator(metadata, new AbstractIterator<>()
         {