Change schema change response in native protocol v3

patch by slebresne; reviewed by thobbs for CASSANDRA-7413

CHANGES.txt

@@ -1,4 +1,4 @@
-2.1.0
+2.1.0-rc2
  * Modify reconcile logic to always pick a tombstone over a counter cell
    (CASSANDRA-7346)
  * Avoid incremental compaction on Windows (CASSANDRA-7365)
@@ -16,6 +16,7 @@
  * RefCount native frames from netty (CASSANDRA-7245)
  * Use tarball dir instead of /var for default paths (CASSANDRA-7136)
  * Remove rows_per_partition_to_cache keyword (CASSANDRA-7193)
+ * Fix schema change response in native protocol v3 (CASSANDRA-7413)
 Merged from 2.0:
  * Fix assertion error in CL.ANY timeout handling (CASSANDRA-7364)
  * Add per-CF range read request latency metrics (CASSANDRA-7338)

doc/native_protocol_v3.spec

@@ -639,17 +639,10 @@ Table of Contents
 4.2.5.5. Schema_change
 
   The result to a schema altering query (creation/update/drop of a
-  keyspace/table/index). The body (after the kind [int]) is composed of 3
+  keyspace/table/index). The body (after the kind [int]) is the same
-  [string]:
+  as the body for a "SCHEMA_CHANGE" event, so 3 strings:
-    <change><keyspace><table>
+    <change_type><target><options>
-  where:
+  Please refer to the section 4.2.6 below for the meaning of those fields.
-    - <change> describe the type of change that has occured. It can be one of
-      "CREATED", "UPDATED" or "DROPPED".
-    - <keyspace> is the name of the affected keyspace or the keyspace of the
-      affected table.
-    - <table> is the name of the affected table. <table> will be empty (i.e.
-      the empty string "") if the change was affecting a keyspace and not a
-      table.
 
   Note that queries to create and drop an index are considered as change
   updating the table the index is on.
@@ -674,11 +667,11 @@ Table of Contents
       concerned node.
     - "SCHEMA_CHANGE": events related to schema change. After the event type,
       the rest of the message will be <change_type><target><options> where:
-        - <change_type> is the type of changed involved. It will be one of
+        - <change_type> is a [string] representing the type of changed involved.
-          "CREATED", "UPDATED" or "DROPPED".
+          It will be one of "CREATED", "UPDATED" or "DROPPED".
-        - <target> can be one of "KEYSPACE", "TABLE" or "TYPE" and describes
+        - <target> is a [string] that can be one of "KEYSPACE", "TABLE" or "TYPE"
-          what has been modified ("TYPE" stands for modifications related to
+          and describes what has been modified ("TYPE" stands for modifications
-          user types).
+          related to user types).
         - <options> depends on the preceding <target>. If <target> is
           "KEYSPACE", then <options> will be a single [string] representing the
           keyspace changed. Otherwise, if <target> is "TABLE" or "TYPE", then
@@ -916,6 +909,6 @@ Table of Contents
   * QUERY, EXECUTE and BATCH messages can now optionally provide the names for the values of the
     query. As this feature is optionally enabled by clients, implementing it is at the discretion of the
     client.
-  * The format of "SCHEMA_CHANGE" notifications has been modified, and now includes changes related to
+  * The format of "Schema_change" results (Section 4.2.5.5) and "SCHEMA_CHANGE" events (Section 4.2.6)
-    user types.
+    has been modified, and now includes changes related to user types.
 

src/java/org/apache/cassandra/cql3/UTName.java

@@ -50,6 +50,11 @@ public class UTName
         return utName.bytes;
     }
 
+    public String getStringTypeName()
+    {
+        return utName.toString();
+    }
+
     @Override
     public String toString()
     {

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

@@ -27,7 +27,7 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class AlterKeyspaceStatement extends SchemaAlteringStatement
 {
@@ -89,8 +89,8 @@ public class AlterKeyspaceStatement extends SchemaAlteringStatement
         MigrationManager.announceKeyspaceUpdate(attrs.asKSMetadataUpdate(ksm), isLocalOnly);
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, keyspace());
     }
 }

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

@@ -29,7 +29,7 @@ import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily;
 
@@ -249,8 +249,8 @@ public class AlterTableStatement extends SchemaAlteringStatement
                              validator);
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 }

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

@@ -28,7 +28,7 @@ import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public abstract class AlterTypeStatement extends SchemaAlteringStatement
 {
@@ -78,9 +78,9 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
         // It doesn't really change anything anyway.
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TYPE, keyspace(), name.getStringTypeName());
     }
 
     @Override

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

@@ -35,7 +35,7 @@ import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 /** A <code>CREATE INDEX</code> statement parsed from a CQL query. */
 public class CreateIndexStatement extends SchemaAlteringStatement
@@ -150,9 +150,9 @@ public class CreateIndexStatement extends SchemaAlteringStatement
         MigrationManager.announceColumnFamilyUpdate(cfm, false, isLocalOnly);
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
         // Creating an index is akin to updating the CF
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 }

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

@@ -30,7 +30,7 @@ import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 /** A <code>CREATE KEYSPACE</code> statement parsed from a CQL query. */
 public class CreateKeyspaceStatement extends SchemaAlteringStatement
@@ -110,8 +110,8 @@ public class CreateKeyspaceStatement extends SchemaAlteringStatement
         }
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.CREATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.CREATED, keyspace());
     }
 }

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

@@ -38,7 +38,7 @@ import org.apache.cassandra.io.compress.CompressionParameters;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.CqlResult;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 /** A <code>CREATE TABLE</code> parsed from a CQL query statement. */
@@ -118,9 +118,9 @@ public class CreateTableStatement extends SchemaAlteringStatement
         }
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.CREATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.CREATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 
     /**

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

@@ -30,7 +30,7 @@ import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 import org.apache.cassandra.triggers.TriggerExecutor;
 
 public class CreateTriggerStatement extends SchemaAlteringStatement
@@ -73,8 +73,8 @@ public class CreateTriggerStatement extends SchemaAlteringStatement
         MigrationManager.announceColumnFamilyUpdate(cfm, false, isLocalOnly);
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 }

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

@@ -29,7 +29,7 @@ import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class CreateTypeStatement extends SchemaAlteringStatement
 {
@@ -87,9 +87,9 @@ public class CreateTypeStatement extends SchemaAlteringStatement
         }
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.CREATED, Event.SchemaChange.Target.TYPE, keyspace(), name.getStringTypeName());
     }
 
     @Override

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

@@ -26,7 +26,7 @@ import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class DropIndexStatement extends SchemaAlteringStatement
 {
@@ -54,10 +54,10 @@ public class DropIndexStatement extends SchemaAlteringStatement
         // validated in findIndexedCf()
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
         // Dropping an index is akin to updating the CF
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 
     public void announceMigration(boolean isLocalOnly) throws InvalidRequestException, ConfigurationException

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

@@ -25,7 +25,7 @@ import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class DropKeyspaceStatement extends SchemaAlteringStatement
 {
@@ -68,8 +68,8 @@ public class DropKeyspaceStatement extends SchemaAlteringStatement
         }
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.DROPPED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.DROPPED, keyspace());
     }
 }

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

@@ -24,7 +24,7 @@ import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class DropTableStatement extends SchemaAlteringStatement
 {
@@ -67,8 +67,8 @@ public class DropTableStatement extends SchemaAlteringStatement
         }
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.DROPPED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.DROPPED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 }

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

@@ -29,7 +29,7 @@ import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class DropTriggerStatement extends SchemaAlteringStatement
 {
@@ -62,8 +62,8 @@ public class DropTriggerStatement extends SchemaAlteringStatement
         MigrationManager.announceColumnFamilyUpdate(cfm, false, isLocalOnly);
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.UPDATED, Event.SchemaChange.Target.TABLE, keyspace(), columnFamily());
     }
 }

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

@@ -24,7 +24,7 @@ import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
-import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.transport.Event;
 
 public class DropTypeStatement extends SchemaAlteringStatement
 {
@@ -126,9 +126,9 @@ public class DropTypeStatement extends SchemaAlteringStatement
         return false;
     }
 
-    public ResultMessage.SchemaChange.Change changeType()
+    public Event.SchemaChange changeEvent()
     {
-        return ResultMessage.SchemaChange.Change.UPDATED;
+        return new Event.SchemaChange(Event.SchemaChange.Change.DROPPED, Event.SchemaChange.Target.TYPE, keyspace(), name.getStringTypeName());
     }
 
     @Override

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

@@ -23,6 +23,7 @@ import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.Event;
 import org.apache.cassandra.transport.messages.ResultMessage;
 
 /**
@@ -62,15 +63,14 @@ public abstract class SchemaAlteringStatement extends CFStatement implements CQL
         return new Prepared(this);
     }
 
-    public abstract ResultMessage.SchemaChange.Change changeType();
+    public abstract Event.SchemaChange changeEvent();
 
     public abstract void announceMigration(boolean isLocalOnly) throws RequestValidationException;
 
     public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException
     {
         announceMigration(false);
-        String tableName = cfName == null || columnFamily() == null ? "" : columnFamily();
+        return new ResultMessage.SchemaChange(changeEvent());
-        return new ResultMessage.SchemaChange(changeType(), keyspace(), tableName);
     }
 
     public ResultMessage executeInternal(QueryState state, QueryOptions options)
@@ -78,8 +78,7 @@ public abstract class SchemaAlteringStatement extends CFStatement implements CQL
         try
         {
             announceMigration(true);
-            String tableName = cfName == null || columnFamily() == null ? "" : columnFamily();
+            return new ResultMessage.SchemaChange(changeEvent());
-            return new ResultMessage.SchemaChange(changeType(), keyspace(), tableName);
         }
         catch (RequestValidationException e)
         {

src/java/org/apache/cassandra/transport/Event.java

@@ -226,7 +226,7 @@ public abstract class Event
         }
 
         // Assumes the type has already been deserialized
-        private static SchemaChange deserializeEvent(ByteBuf cb, int version)
+        public static SchemaChange deserializeEvent(ByteBuf cb, int version)
         {
             Change change = CBUtil.readEnumValue(Change.class, cb);
             if (version >= 3)
@@ -244,7 +244,7 @@ public abstract class Event
             }
         }
 
-        protected void serializeEvent(ByteBuf dest, int version)
+        public void serializeEvent(ByteBuf dest, int version)
         {
             if (version >= 3)
             {
@@ -262,7 +262,7 @@ public abstract class Event
             }
         }
 
-        protected int eventSerializedSize(int version)
+        public int eventSerializedSize(int version)
         {
             if (version >= 3)
             {

src/java/org/apache/cassandra/transport/messages/ResultMessage.java

@@ -344,56 +344,33 @@ public abstract class ResultMessage extends Message.Response
 
     public static class SchemaChange extends ResultMessage
     {
-        public enum Change { CREATED, UPDATED, DROPPED }
+        public final Event.SchemaChange change;
 
-        public final Change change;
+        public SchemaChange(Event.SchemaChange change)
-        public final String keyspace;
-        public final String columnFamily;
-
-        public SchemaChange(Change change, String keyspace)
-        {
-            this(change, keyspace, "");
-        }
-
-        public SchemaChange(Change change, String keyspace, String columnFamily)
         {
             super(Kind.SCHEMA_CHANGE);
             this.change = change;
-            this.keyspace = keyspace;
-            this.columnFamily = columnFamily;
         }
 
         public static final Message.Codec<ResultMessage> subcodec = new Message.Codec<ResultMessage>()
         {
             public ResultMessage decode(ByteBuf body, int version)
             {
-                Change change = CBUtil.readEnumValue(Change.class, body);
+                return new SchemaChange(Event.SchemaChange.deserializeEvent(body, version));
-                String keyspace = CBUtil.readString(body);
-                String columnFamily = CBUtil.readString(body);
-                return new SchemaChange(change, keyspace, columnFamily);
-
             }
 
             public void encode(ResultMessage msg, ByteBuf dest, int version)
             {
                 assert msg instanceof SchemaChange;
                 SchemaChange scm = (SchemaChange)msg;
-
+                scm.change.serializeEvent(dest, version);
-                CBUtil.writeEnumValue(scm.change, dest);
-                CBUtil.writeString(scm.keyspace, dest);
-                CBUtil.writeString(scm.columnFamily, dest);
             }
 
             public int encodedSize(ResultMessage msg, int version)
             {
                 assert msg instanceof SchemaChange;
                 SchemaChange scm = (SchemaChange)msg;
-
+                return scm.change.eventSerializedSize(version);
-                int size = 0;
-                size += CBUtil.sizeOfEnumValue(scm.change);
-                size += CBUtil.sizeOfString(scm.keyspace);
-                size += CBUtil.sizeOfString(scm.columnFamily);
-                return size;
             }
         };
 
@@ -405,7 +382,7 @@ public abstract class ResultMessage extends Message.Response
         @Override
         public String toString()
         {
-            return "RESULT schema change " + change + " on " + keyspace + (columnFamily.isEmpty() ? "" : "." + columnFamily);
+            return "RESULT schema change " + change;
         }
     }
 }