Remote vtables

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	url = https://github.com/belliottsmith/cassandra-accord.git
+	branch = rebootstrap-sq

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

@@ -156,13 +156,13 @@ public static StatementRestrictions empty(StatementType type, TableMetadata tabl
         return new StatementRestrictions(type, table, IndexHints.NONE, false);
     }
 
-    private StatementRestrictions(StatementType type, TableMetadata table, IndexHints indexHints, boolean allowFiltering)
+    private StatementRestrictions(StatementType type, TableMetadata table, IndexHints indexHints, boolean allowFilteringOfPrimaryKeys)
     {
         this.type = type;
         this.table = table;
         this.indexHints = indexHints;
         this.partitionKeyRestrictions = new PartitionKeyRestrictions(table.partitionKeyAsClusteringComparator());
-        this.clusteringColumnsRestrictions = new ClusteringColumnRestrictions(table, allowFiltering);
+        this.clusteringColumnsRestrictions = new ClusteringColumnRestrictions(table, allowFilteringOfPrimaryKeys);
         this.nonPrimaryKeyRestrictions = RestrictionSet.empty();
         this.notNullColumns = new HashSet<>();
     }
@@ -370,7 +370,7 @@ else if (operator.requiresIndexing())
             }
             else
             {
-                if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table))
+                if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table, false))
                     throw invalidRequest(allowFilteringMessage(state));
             }
 
@@ -381,14 +381,14 @@ else if (operator.requiresIndexing())
             validateSecondaryIndexSelections();
     }
 
-    public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata)
+    public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata, boolean isPrimaryKey)
     {
         if (!metadata.isVirtual())
             return true;
 
         VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(metadata.id);
         assert tableNullable != null;
-        return !tableNullable.allowFilteringImplicitly();
+        return isPrimaryKey ? !tableNullable.allowFilteringPrimaryKeysImplicitly() : !tableNullable.allowFilteringImplicitly();
     }
 
     private void addRestriction(Restriction restriction, IndexRegistry indexRegistry, IndexHints indexHints)
@@ -593,7 +593,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(table))
+                if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified(table, true))
                     throw new InvalidRequestException(allowFilteringMessage(state));
 
                 isKeyRange = true;

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

@@ -1474,7 +1474,7 @@ private StatementRestrictions prepareRestrictions(ClientState state,
                                              boundNames,
                                              orderings,
                                              selectsOnlyStaticColumns,
-                                             parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata),
+                                             parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata, true),
                                              forView);
         }
 
@@ -1700,7 +1700,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 (requiresAllowFilteringIfNotSpecified(table))
+                if (requiresAllowFilteringIfNotSpecified(table, true))
                     checkFalse(restrictions.needFiltering(table), StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE);
             }
         }

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

@@ -647,7 +647,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(), rowFilter());
+            UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter(), rowFilter(), limits());
             return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
         }
 

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

@@ -1516,7 +1516,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(), rowFilter());
+            UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter(), rowFilter(), limits());
             return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
         }
 

src/java/org/apache/cassandra/db/filter/ColumnFilter.java

@@ -33,6 +33,7 @@
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.btree.BTree;
 
 /**
  * Represents which (non-PK) columns (and optionally which sub-part of a column for complex columns) are selected
@@ -64,7 +65,6 @@
  */
 public abstract class ColumnFilter
 {
-
     public static final ColumnFilter NONE = selection(RegularAndStaticColumns.NONE);
 
     public static final Serializer serializer = new Serializer();
@@ -305,6 +305,11 @@ public boolean isWildcard()
         return false;
     }
 
+    /**
+     * Rebinds matching columns into a new filter; ignores any missing but fails if any are a different type
+     */
+    public abstract ColumnFilter rebind(TableMetadata newTable);
+
     /**
      * Returns the CQL string corresponding to this {@code ColumnFilter}.
      *
@@ -630,6 +635,12 @@ public boolean isWildcard()
             return true;
         }
 
+        @Override
+        public ColumnFilter rebind(TableMetadata newTable)
+        {
+            return new WildCardColumnFilter(ColumnFilter.rebind(newTable, fetchedAndQueried));
+        }
+
         @Override
         protected SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections()
         {
@@ -779,6 +790,17 @@ public Tester newTester(ColumnMetadata column)
             return new Tester(fetchingStrategy.fetchesAllColumns(column.isStatic()), s.iterator());
         }
 
+        @Override
+        public ColumnFilter rebind(TableMetadata newTable)
+        {
+            RegularAndStaticColumns queried = ColumnFilter.rebind(newTable, this.queried);
+            RegularAndStaticColumns fetched = this.queried == this.fetched ? queried : ColumnFilter.rebind(newTable, this.fetched);
+            SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections = this.subSelections;
+            if (subSelections != null)
+                subSelections = TreeMultimap.create(subSelections);
+            return new SelectionColumnFilter(fetchingStrategy, queried, fetched, subSelections);
+        }
+
         @Override
         protected SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections()
         {
@@ -1003,4 +1025,26 @@ private long subSelectionsSerializedSize(SortedSetMultimap<ColumnIdentifier, Col
             return size;
         }
     }
+
+    private static RegularAndStaticColumns rebind(TableMetadata newTable, RegularAndStaticColumns columns)
+    {
+        return new RegularAndStaticColumns(rebind(newTable, columns.statics), rebind(newTable, columns.regulars));
+    }
+
+    private static Columns rebind(TableMetadata newTable, Columns columns)
+    {
+        if (columns.isEmpty())
+            return columns;
+
+        try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
+        {
+            for (ColumnMetadata in : columns)
+            {
+                ColumnMetadata out = newTable.getColumn(in.name);
+                if (out != null)
+                    builder.add(out);
+            }
+            return Columns.from(builder);
+        }
+    }
 }

src/java/org/apache/cassandra/db/filter/RowFilter.java

@@ -141,7 +141,7 @@ public void addCustomIndexExpression(TableMetadata metadata, IndexMetadata targe
         add(new CustomExpression(metadata, targetIndex, value));
     }
 
-    private void add(Expression expression)
+    public void add(Expression expression)
     {
         expression.validate();
         expressions.add(expression);
@@ -549,6 +549,28 @@ public void validateForIndexing()
                        "Index expression values may not be larger than 64K");
         }
 
+        /**
+         * Rebind this expression to a table metadata that is expected to have equivalent columns.
+         * If any referenced column is missing, returns null;
+         * if any referenced column has a different type throws an exception
+         */
+        public Expression rebind(TableMetadata newTable)
+        {
+            throw new UnsupportedOperationException("Expression " + toString(true) + " does not support rebinding to another table definition");
+        }
+
+        protected static ColumnMetadata rebind(ColumnMetadata in, TableMetadata newTable)
+        {
+            ColumnMetadata out = newTable.getColumn(in.name);
+            if (out == null)
+                return null;
+
+            if (!out.type.equals(in.type) && !out.type.isCompatibleWith(in.type) || !in.type.isCompatibleWith(out.type))
+                throw new IllegalArgumentException("The provided TableMetadata is not compatible with the expression");
+
+            return out;
+        }
+
         /**
          * Returns whether the provided row satisfied this expression or not.
          *
@@ -734,6 +756,16 @@ public static class SimpleExpression extends Expression
             super(column, operator, value);
         }
 
+        @Override
+        public Expression rebind(TableMetadata newTable)
+        {
+            ColumnMetadata out = rebind(column, newTable);
+            if (out == null)
+                return null;
+
+            return new SimpleExpression(out, operator, value);
+        }
+
         @Override
         public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
@@ -853,6 +885,16 @@ public void validate() throws InvalidRequestException
             checkBindValueSet(value, "Unsupported unset map value for column %s", column.name);
         }
 
+        @Override
+        public Expression rebind(TableMetadata newTable)
+        {
+            ColumnMetadata out = rebind(column, newTable);
+            if (out == null)
+                return null;
+
+            return new MapElementExpression(out, key, operator, value);
+        }
+
         @Override
         public ByteBuffer getIndexValue()
         {
@@ -978,6 +1020,12 @@ protected Kind kind()
             return Kind.CUSTOM;
         }
 
+        @Override
+        public Expression rebind(TableMetadata newTable)
+        {
+            return new CustomExpression(table, targetIndex, value);
+        }
+
         // Filtering by custom expressions isn't supported yet, so just accept any row
         @Override
         public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)

src/java/org/apache/cassandra/db/marshal/TxnIdUtf8Type.java

@@ -48,6 +48,12 @@ public <V> void validate(V value, ValueAccessor<V> accessor) throws MarshalExcep
 
     String describe() { return "TxnId"; }
 
+    @Override
+    public boolean isEmptyValueMeaningless()
+    {
+        return true;
+    }
+
     @Override
     public TypeSerializer<String> getSerializer()
     {