CFS.setCompressionParameters() method can affect schema globally

Patch by Alex Petrov; reviewed by Aleksey Yeschenko for CASSANDRA-12949

src/java/org/apache/cassandra/config/CFMetaData.java

@@ -102,6 +102,7 @@ public final class CFMetaData
     private volatile Map<ByteBuffer, DroppedColumn> droppedColumns = new HashMap<>();
     private volatile Triggers triggers = Triggers.none();
     private volatile Indexes indexes = Indexes.none();
+    private volatile CompressionParams localCompressionParams;
 
     /*
      * All CQL3 columns definition are stored in the columnMetadata map.
@@ -152,6 +153,18 @@ public final class CFMetaData
 
     public ColumnDefinition superColumnKeyColumn() { return superCfKeyColumn; }
 
+    public void setLocalCompressionParams(CompressionParams params)
+    {
+        this.localCompressionParams = params;
+    }
+
+    public CompressionParams compressionParams()
+    {
+        if (localCompressionParams == null)
+            return params.compression;
+        return localCompressionParams;
+    }
+
     /*
      * All of these methods will go away once CFMetaData becomes completely immutable.
      */
@@ -338,6 +351,8 @@ public final class CFMetaData
     // are kept because they are often useful in a different format.
     private void rebuild()
     {
+        this.localCompressionParams = null;
+
         // A non-compact copy will be created lazily
         this.nonCompactCopy = null;
 

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

@@ -339,15 +339,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public Map<String,String> getCompressionParameters()
     {
-        return metadata.params.compression.asMap();
+        return metadata.compressionParams().asMap();
     }
 
     public void setCompressionParameters(Map<String,String> opts)
     {
         try
         {
-            metadata.compression(CompressionParams.fromMap(opts));
-            metadata.params.compression.validate();
+            CompressionParams newParams = CompressionParams.fromMap(opts);
+            newParams.validate();
+            metadata.setLocalCompressionParams(newParams);
         }
         catch (ConfigurationException e)
         {
@@ -355,6 +356,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
     }
 
+    public void setCompressionParametersJson(String options)
+    {
+        setCompressionParameters(FBUtilities.fromJsonMap(options));
+    }
+
+    public String getCompressionParametersJson()
+    {
+        return FBUtilities.json(getCompressionParameters());
+    }
+
     private ColumnFamilyStore(Keyspace keyspace,
                               String columnFamilyName,
                               int generation,

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

@@ -93,12 +93,14 @@ public interface ColumnFamilyStoreMBean
      * Get the compression parameters
      */
     public Map<String,String> getCompressionParameters();
+    public String getCompressionParametersJson();
 
     /**
      * Set the compression parameters
      * @param opts map of string names to values
      */
     public void setCompressionParameters(Map<String,String> opts);
+    public void setCompressionParametersJson(String opts);
 
     /**
      * Set new crc check chance

src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java

@@ -137,7 +137,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
         if (metadata.params.bloomFilterFpChance < 1.0)
             components.add(Component.FILTER);
 
-        if (metadata.params.compression.isEnabled())
+        if (metadata.compressionParams().isEnabled())
         {
             components.add(Component.COMPRESSION_INFO);
         }

src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java

@@ -72,7 +72,7 @@ public class BigTableWriter extends SSTableWriter
         {
             dataFile = SequentialWriter.open(getFilename(),
                                              descriptor.filenameFor(Component.COMPRESSION_INFO),
-                                             metadata.params.compression,
+                                             metadata.compressionParams(),
                                              metadataCollector);
             dbuilder = SegmentedFile.getCompressedBuilder((CompressedSequentialWriter) dataFile);
         }

test/distributed/org/apache/cassandra/distributed/test/AlterTest.java

@@ -0,0 +1,60 @@
+/*
+ * 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.distributed.test;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+
+public class AlterTest extends TestBaseImpl
+{
+    @Test
+    public void getAndSetCompressionParametersTest() throws Throwable
+    {
+        try (ICluster<IInvokableInstance> cluster = init(builder().withNodes(2)
+                                                                  .start()))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));");
+            cluster.stream().forEach((i) -> {
+                i.acceptsOnInstance((IIsolatedExecutor.SerializableConsumer<String>) (ks) -> {
+                    Keyspace.open(ks)
+                            .getColumnFamilyStore("tbl")
+                            .setCompressionParametersJson("{\"chunk_length_in_kb\": \"128\"," +
+                                                          " \"class\": \"org.apache.cassandra.io.compress.LZ4Compressor\"}");
+                    Assert.assertTrue(Keyspace.open(ks)
+                                              .getColumnFamilyStore("tbl")
+                                              .getCompressionParametersJson().contains("128"));
+                }).accept(KEYSPACE);
+            });
+            cluster.schemaChange("ALTER TABLE " + KEYSPACE + ".tbl ADD v2 int");
+
+            cluster.stream().forEach((i) -> {
+                i.acceptsOnInstance((IIsolatedExecutor.SerializableConsumer<String>) (ks) -> {
+                    Assert.assertFalse(Keyspace.open(ks)
+                                               .getColumnFamilyStore("tbl")
+                                               .getCompressionParametersJson().contains("128"));
+                }).accept(KEYSPACE);
+            });
+        }
+    }
+}