RemoteToLocalVirtualKeyspace: supporting access to all nodes' local virtual tables from any node in the cluster

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20900

Author: belliottsmith

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/virtual/AccordDebugRemoteKeyspace.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.db.virtual;
+
+import org.apache.cassandra.schema.SchemaConstants;
+
+public class AccordDebugRemoteKeyspace extends RemoteToLocalVirtualKeyspace
+{
+    public static final AccordDebugRemoteKeyspace instance = new AccordDebugRemoteKeyspace(SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.instance);
+
+    public AccordDebugRemoteKeyspace(String name, VirtualKeyspace wrap)
+    {
+        super(name, wrap);
+    }
+}

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

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.db.virtual;
+
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+public class RemoteToLocalVirtualKeyspace extends VirtualKeyspace
+{
+    public RemoteToLocalVirtualKeyspace(String name, VirtualKeyspace wrap)
+    {
+        this(name, wrap, ignore -> true);
+    }
+
+    public RemoteToLocalVirtualKeyspace(String name, VirtualKeyspace wrap, Predicate<VirtualTable> include)
+    {
+        super(name, wrap.tables().stream().filter(include).map(vt -> new RemoteToLocalVirtualTable(name, vt)).collect(Collectors.toList()));
+    }
+}