cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ifesdj...@apache.org
Subject [6/6] cassandra git commit: Merge branch 'cassandra-3.11' into trunk
Date Thu, 15 Jun 2017 17:38:29 GMT
Merge branch 'cassandra-3.11' into trunk


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/a07d327b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/a07d327b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/a07d327b

Branch: refs/heads/trunk
Commit: a07d327be86783137b7ae46a7722ac41cbebdc31
Parents: f5531e1 2a0890d
Author: Alex Petrov <oleksandr.petrov@gmail.com>
Authored: Thu Jun 15 19:36:30 2017 +0200
Committer: Alex Petrov <oleksandr.petrov@gmail.com>
Committed: Thu Jun 15 19:36:30 2017 +0200

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../cassandra/db/filter/ColumnFilter.java       | 121 +++++++++++++++----
 .../apache/cassandra/net/MessagingService.java  |   3 +-
 .../cassandra/db/filter/ColumnFilterTest.java   |  83 +++++++++++++
 4 files changed, 182 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/a07d327b/CHANGES.txt
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a07d327b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index b568704,37da86a..dcd93e8
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@@ -20,8 -20,6 +20,9 @@@ package org.apache.cassandra.db.filter
  import java.io.IOException;
  import java.util.*;
  
++import com.google.common.annotations.VisibleForTesting;
 +import com.google.common.collect.Iterables;
 +import com.google.common.collect.Iterators;
  import com.google.common.collect.SortedSetMultimap;
  import com.google.common.collect.TreeMultimap;
  
@@@ -66,25 -63,23 +67,59 @@@ public class ColumnFilte
  {
      public static final Serializer serializer = new Serializer();
  
 -    // True if _fetched_ is all the columns, in which case metadata must not be null. If
false,
 -    // then _fetched_ == _queried_ and we only store _queried_.
 -    private final boolean isFetchAll;
 +    // True if _fetched_ includes all regular columns (and any static in _queried_), in
which case metadata must not be
 +    // null. If false, then _fetched_ == _queried_ and we only store _queried_.
-     private final boolean fetchAllRegulars;
- 
-     private final TableMetadata metadata; // can be null if !isFetchAll
++    public final boolean fetchAllRegulars;
  
 -    private final PartitionColumns fetched;
 -    private final PartitionColumns queried; // can be null if isFetchAll and _fetched_ ==
_queried_
++    private final RegularAndStaticColumns fetched;
 +    private final RegularAndStaticColumns queried; // can be null if fetchAllRegulars, to
represent a wildcard query (all
-                                             // static and regular columns are both _fetched_
and _queried_).
++                                                   // static and regular columns are both
_fetched_ and _queried_).
      private final SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections;
// can be null
  
 -    private ColumnFilter(boolean isFetchAll,
 -                         PartitionColumns fetched,
 -                         PartitionColumns queried,
 +    private ColumnFilter(boolean fetchAllRegulars,
 +                         TableMetadata metadata,
 +                         RegularAndStaticColumns queried,
                           SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
      {
 -        assert !isFetchAll || fetched != null;
 -        assert isFetchAll || queried != null;
 -        this.isFetchAll = isFetchAll;
 -        this.fetched = isFetchAll ? fetched : queried;
 +        assert !fetchAllRegulars || metadata != null;
 +        assert fetchAllRegulars || queried != null;
 +        this.fetchAllRegulars = fetchAllRegulars;
-         this.metadata = metadata;
++
++        if (fetchAllRegulars)
++        {
++            RegularAndStaticColumns all = metadata.regularAndStaticColumns();
++            if (queried == null)
++            {
++                this.fetched = this.queried = all;
++            }
++            else
++            {
++                this.fetched = all.statics.isEmpty()
++                               ? all
++                               : new RegularAndStaticColumns(queried.statics, all.regulars);
++                this.queried = queried;
++            }
++        }
++        else
++        {
++            this.fetched = this.queried = queried;
++        }
++
++        this.subSelections = subSelections;
++    }
++
++    /**
++     * Used on replica for deserialisation
++     */
++    private ColumnFilter(boolean fetchAllRegulars,
++                         RegularAndStaticColumns fetched,
++                         RegularAndStaticColumns queried,
++                         SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
++    {
++        assert !fetchAllRegulars || fetched != null;
++        assert fetchAllRegulars || queried != null;
++        this.fetchAllRegulars = fetchAllRegulars;
++        this.fetched = fetchAllRegulars ? fetched : queried;
          this.queried = queried;
          this.subSelections = subSelections;
      }
@@@ -104,9 -99,9 +139,9 @@@
       * preserve CQL semantic (see class javadoc). This is ok for some internal queries however
(and
       * for #6588 if/when we implement it).
       */
 -    public static ColumnFilter selection(PartitionColumns columns)
 +    public static ColumnFilter selection(RegularAndStaticColumns columns)
      {
--        return new ColumnFilter(false, null, columns, null);
++        return new ColumnFilter(false, (TableMetadata) null, columns, null);
      }
  
  	/**
@@@ -123,17 -118,9 +158,9 @@@
       *
       * @return the columns to fetch for this filter.
       */
 -    public PartitionColumns fetchedColumns()
 +    public RegularAndStaticColumns fetchedColumns()
      {
-         if (!fetchAllRegulars)
-             return queried;
- 
-         // We always fetch all regulars, but only fetch the statics in queried. Unless queried
== null, in which
-         // case it's a wildcard and we fetch everything.
-         RegularAndStaticColumns all = metadata.regularAndStaticColumns();
-         return queried == null || all.statics.isEmpty()
-              ? all
-              : new RegularAndStaticColumns(queried.statics, all.regulars);
+         return fetched;
      }
  
      /**
@@@ -141,28 -128,14 +168,27 @@@
       * <p>
       * Note that this is in general not all the columns that are fetched internally (see
{@link #fetchedColumns}).
       */
 -    public PartitionColumns queriedColumns()
 +    public RegularAndStaticColumns queriedColumns()
      {
-         assert queried != null || fetchAllRegulars;
-         return queried == null ? metadata.regularAndStaticColumns() : queried;
 -        return queried == null ? fetched : queried;
++        return queried;
      }
  
 -    public boolean fetchesAllColumns()
 +    /**
 +     * Wether all the (regular or static) columns are fetched by this filter.
 +     * <p>
 +     * Note that this method is meant as an optimization but a negative return
 +     * shouldn't be relied upon strongly: this can return {@code false} but
 +     * still have all the columns fetches if those were manually selected by the
 +     * user. The goal here is to cheaply avoid filtering things on wildcard
 +     * queries, as those are common.
 +     *
 +     * @param isStatic whether to check for static columns or not. If {@code true},
 +     * the method returns if all static columns are fetched, otherwise it checks
 +     * regular columns.
 +     */
 +    public boolean fetchesAllColumns(boolean isStatic)
      {
 -        return isFetchAll;
 +        return isStatic ? queried == null : fetchAllRegulars;
      }
  
      /**
@@@ -432,9 -356,26 +458,26 @@@
      }
  
      @Override
+     public boolean equals(Object other)
+     {
+         if (other == this)
+             return true;
+ 
+         if (!(other instanceof ColumnFilter))
+             return false;
+ 
+         ColumnFilter otherCf = (ColumnFilter) other;
+ 
 -        return otherCf.isFetchAll == this.isFetchAll &&
++        return otherCf.fetchAllRegulars == this.fetchAllRegulars &&
+                Objects.equals(otherCf.fetched, this.fetched) &&
+                Objects.equals(otherCf.queried, this.queried) &&
+                Objects.equals(otherCf.subSelections, this.subSelections);
+     }
+ 
+     @Override
      public String toString()
      {
 -        if (isFetchAll)
 +        if (fetchAllRegulars && queried == null)
              return "*";
  
          if (queried.isEmpty())
@@@ -476,8 -417,8 +519,8 @@@
  
      public static class Serializer
      {
-         private static final int FETCH_ALL_MASK       = 0x01;
 -        private static final int IS_FETCH_ALL_MASK       = 0x01;
--        private static final int HAS_QUERIED_MASK      = 0x02;
++        private static final int FETCH_ALL_MASK          = 0x01;
++        private static final int HAS_QUERIED_MASK        = 0x02;
          private static final int HAS_SUB_SELECTIONS_MASK = 0x04;
  
          private static int makeHeaderByte(ColumnFilter selection)
@@@ -487,33 -428,16 +530,39 @@@
                   | (selection.subSelections != null ? HAS_SUB_SELECTIONS_MASK : 0);
          }
  
-         private static ColumnFilter maybeUpdateForBackwardCompatility(ColumnFilter selection,
int version)
++        @VisibleForTesting
++        public static ColumnFilter maybeUpdateForBackwardCompatility(ColumnFilter selection,
int version)
 +        {
 +            if (version > MessagingService.VERSION_30 || !selection.fetchAllRegulars
|| selection.queried == null)
 +                return selection;
 +
 +            // The meaning of fetchAllRegulars changed (at least when queried != null) due
to CASSANDRA-12768: in
 +            // pre-4.0 it means that *all* columns are fetched, not just the regular ones,
and so 3.0/3.X nodes
 +            // would send us more than we'd like. So instead recreating a filter that correspond
to what we
 +            // actually want (it's a tiny bit less efficient as we include all columns manually
and will mark as
 +            // queried some columns that are actually only fetched, but it's fine during
upgrade).
 +            // More concretely, we replace our filter by a non-fetch-all one that queries
every columns that our
 +            // current filter fetches.
-             Columns allRegulars = selection.metadata.regularColumns();
 +            Set<ColumnMetadata> queriedStatic = new HashSet<>();
 +            Iterables.addAll(queriedStatic, Iterables.filter(selection.queried, ColumnMetadata::isStatic));
 +            return new ColumnFilter(false,
-                                     null,
-                                     new RegularAndStaticColumns(Columns.from(queriedStatic),
allRegulars),
++                                    (TableMetadata) null,
++                                    new RegularAndStaticColumns(Columns.from(queriedStatic),
selection.fetched.regulars),
 +                                    selection.subSelections);
 +        }
 +
          public void serialize(ColumnFilter selection, DataOutputPlus out, int version) throws
IOException
          {
 +            selection = maybeUpdateForBackwardCompatility(selection, version);
 +
              out.writeByte(makeHeaderByte(selection));
  
 -            if (version >= MessagingService.VERSION_3014 && selection.isFetchAll)
++            if (version >= MessagingService.VERSION_3014 && selection.fetchAllRegulars)
+             {
+                 Columns.serializer.serialize(selection.fetched.statics, out);
+                 Columns.serializer.serialize(selection.fetched.regulars, out);
+             }
+ 
              if (selection.queried != null)
              {
                  Columns.serializer.serialize(selection.queried.statics, out);
@@@ -535,7 -459,23 +584,23 @@@
              boolean hasQueried = (header & HAS_QUERIED_MASK) != 0;
              boolean hasSubSelections = (header & HAS_SUB_SELECTIONS_MASK) != 0;
  
 -            PartitionColumns fetched = null;
 -            PartitionColumns queried = null;
++            RegularAndStaticColumns fetched = null;
 +            RegularAndStaticColumns queried = null;
+ 
+             if (isFetchAll)
+             {
+                 if (version >= MessagingService.VERSION_3014)
+                 {
+                     Columns statics = Columns.serializer.deserialize(in, metadata);
+                     Columns regulars = Columns.serializer.deserialize(in, metadata);
 -                    fetched = new PartitionColumns(statics, regulars);
++                    fetched = new RegularAndStaticColumns(statics, regulars);
+                 }
+                 else
+                 {
 -                    fetched = metadata.partitionColumns();
++                    fetched = metadata.regularAndStaticColumns();
+                 }
+             }
+ 
              if (hasQueried)
              {
                  Columns statics = Columns.serializer.deserialize(in, metadata);
@@@ -555,24 -495,19 +620,30 @@@
                  }
              }
  
 +            // Same concern than in serialize/serializedSize: we should be wary of the change
in meaning for isFetchAll.
 +            // If we get a filter with isFetchAll from 3.0/3.x, it actually expects all
static columns to be fetched,
 +            // make sure we do that (note that if queried == null, that's already what we
do).
 +            // Note that here again this will make us do a bit more work that necessary,
namely we'll _query_ all
 +            // statics even though we only care about _fetching_ them all, but that's a
minor inefficiency, so fine
 +            // during upgrade.
 +            if (version <= MessagingService.VERSION_30 && isFetchAll &&
queried != null)
 +                queried = new RegularAndStaticColumns(metadata.staticColumns(), queried.regulars);
 +
-             return new ColumnFilter(isFetchAll, isFetchAll ? metadata : null, queried, subSelections);
+             return new ColumnFilter(isFetchAll, fetched, queried, subSelections);
          }
  
          public long serializedSize(ColumnFilter selection, int version)
          {
 +            selection = maybeUpdateForBackwardCompatility(selection, version);
 +
              long size = 1; // header byte
  
 -            if (version >= MessagingService.VERSION_3014 && selection.isFetchAll)
++            if (version >= MessagingService.VERSION_3014 && selection.fetchAllRegulars)
+             {
+                 size += Columns.serializer.serializedSize(selection.fetched.statics);
+                 size += Columns.serializer.serializedSize(selection.fetched.regulars);
+             }
+ 
              if (selection.queried != null)
              {
                  size += Columns.serializer.serializedSize(selection.queried.statics);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a07d327b/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/net/MessagingService.java
index b3e7b61,032dc03..41771e7
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@@ -93,9 -89,17 +93,10 @@@ public final class MessagingService imp
      public static final String MBEAN_NAME = "org.apache.cassandra.net:type=MessagingService";
  
      // 8 bits version, so don't waste versions
 -    public static final int VERSION_12 = 6;
 -    public static final int VERSION_20 = 7;
 -    public static final int VERSION_21 = 8;
 -    public static final int VERSION_22 = 9;
      public static final int VERSION_30 = 10;
-     public static final int VERSION_40 = 11;
+     public static final int VERSION_3014 = 11;
 -    public static final int current_version = FORCE_3_0_PROTOCOL_VERSION ? VERSION_30 :
VERSION_3014;
++    public static final int VERSION_40 = 12;
 +    public static final int current_version = VERSION_40;
  
      public static final String FAILURE_CALLBACK_PARAM = "CAL_BAC";
      public static final byte[] ONE_BYTE = new byte[1];

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a07d327b/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java
index 0000000,db06d20..eee2ad5
mode 000000,100644..100644
--- a/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java
+++ b/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java
@@@ -1,0 -1,70 +1,83 @@@
+ /*
+  * 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.filter;
+ 
+ import org.junit.Test;
+ 
+ import junit.framework.Assert;
 -import org.apache.cassandra.config.CFMetaData;
 -import org.apache.cassandra.config.ColumnDefinition;
+ import org.apache.cassandra.db.marshal.Int32Type;
+ import org.apache.cassandra.dht.Murmur3Partitioner;
+ import org.apache.cassandra.io.util.DataInputBuffer;
+ import org.apache.cassandra.io.util.DataInputPlus;
+ import org.apache.cassandra.io.util.DataOutputBuffer;
+ import org.apache.cassandra.net.MessagingService;
++import org.apache.cassandra.schema.ColumnMetadata;
++import org.apache.cassandra.schema.TableMetadata;
+ import org.apache.cassandra.utils.ByteBufferUtil;
+ 
+ public class ColumnFilterTest
+ {
+     final static ColumnFilter.Serializer serializer = new ColumnFilter.Serializer();
+ 
+     @Test
+     public void columnFilterSerialisationRoundTrip() throws Exception
+     {
 -        CFMetaData metadata = CFMetaData.Builder.create("ks", "table")
 -                                                .withPartitioner(Murmur3Partitioner.instance)
 -                                                .addPartitionKey("pk", Int32Type.instance)
 -                                                .addClusteringColumn("ck", Int32Type.instance)
 -                                                .addRegularColumn("v1", Int32Type.instance)
 -                                                .addRegularColumn("v2", Int32Type.instance)
 -                                                .addRegularColumn("v3", Int32Type.instance)
 -                                                .build();
++        TableMetadata metadata = TableMetadata.builder("ks", "table")
++                                              .partitioner(Murmur3Partitioner.instance)
++                                              .addPartitionKeyColumn("pk", Int32Type.instance)
++                                              .addClusteringColumn("ck", Int32Type.instance)
++                                              .addRegularColumn("v1", Int32Type.instance)
++                                              .addRegularColumn("v2", Int32Type.instance)
++                                              .addRegularColumn("v3", Int32Type.instance)
++                                              .build();
+ 
 -        ColumnDefinition v1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v1"));
++        ColumnMetadata v1 = metadata.getColumn(ByteBufferUtil.bytes("v1"));
+ 
 -        testRoundTrip(ColumnFilter.all(metadata), metadata, MessagingService.VERSION_30);
++        ColumnFilter columnFilter;
++
++        columnFilter = ColumnFilter.all(metadata);
++        testRoundTrip(columnFilter, ColumnFilter.Serializer.maybeUpdateForBackwardCompatility(columnFilter,
MessagingService.VERSION_30), metadata, MessagingService.VERSION_30);
+         testRoundTrip(ColumnFilter.all(metadata), metadata, MessagingService.VERSION_3014);
++        testRoundTrip(ColumnFilter.all(metadata), metadata, MessagingService.VERSION_40);
++
++        testRoundTrip(ColumnFilter.selection(metadata.regularAndStaticColumns().without(v1)),
metadata, MessagingService.VERSION_30);
++        testRoundTrip(ColumnFilter.selection(metadata.regularAndStaticColumns().without(v1)),
metadata, MessagingService.VERSION_3014);
++        testRoundTrip(ColumnFilter.selection(metadata.regularAndStaticColumns().without(v1)),
metadata, MessagingService.VERSION_40);
+ 
 -        testRoundTrip(ColumnFilter.selection(metadata.partitionColumns().without(v1)), metadata,
MessagingService.VERSION_30);
 -        testRoundTrip(ColumnFilter.selection(metadata.partitionColumns().without(v1)), metadata,
MessagingService.VERSION_3014);
++        columnFilter = ColumnFilter.selection(metadata, metadata.regularAndStaticColumns().without(v1));
++        testRoundTrip(columnFilter, ColumnFilter.Serializer.maybeUpdateForBackwardCompatility(columnFilter,
MessagingService.VERSION_30), metadata, MessagingService.VERSION_30);
++        testRoundTrip(ColumnFilter.selection(metadata, metadata.regularAndStaticColumns().without(v1)),
metadata, MessagingService.VERSION_3014);
++        testRoundTrip(ColumnFilter.selection(metadata, metadata.regularAndStaticColumns().without(v1)),
metadata, MessagingService.VERSION_40);
++    }
+ 
 -        testRoundTrip(ColumnFilter.selection(metadata, metadata.partitionColumns().without(v1)),
metadata, MessagingService.VERSION_30);
 -        testRoundTrip(ColumnFilter.selection(metadata, metadata.partitionColumns().without(v1)),
metadata, MessagingService.VERSION_3014);
++    static void testRoundTrip(ColumnFilter columnFilter, TableMetadata metadata, int version)
throws Exception
++    {
++        testRoundTrip(columnFilter, columnFilter, metadata, version);
+     }
+ 
 -    static void testRoundTrip(ColumnFilter columnFilter, CFMetaData metadata, int version)
throws Exception
++    static void testRoundTrip(ColumnFilter columnFilter, ColumnFilter expected, TableMetadata
metadata, int version) throws Exception
+     {
+         DataOutputBuffer output = new DataOutputBuffer();
+         serializer.serialize(columnFilter, output, version);
+         Assert.assertEquals(serializer.serializedSize(columnFilter, version), output.position());
+         DataInputPlus input = new DataInputBuffer(output.buffer(), false);
 -        Assert.assertEquals(serializer.deserialize(input, version, metadata), columnFilter);
++        ColumnFilter deserialized = serializer.deserialize(input, version, metadata);
++        Assert.assertEquals(deserialized, expected);
+     }
+ }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message