From a29d206532a44f99457af9ddd81d2bc3ed65e63e Mon Sep 17 00:00:00 2001
From: Branimir Lambov <branimir.lambov@datastax.com>
Date: Tue, 15 Sep 2015 18:57:43 +0300
Subject: [PATCH] Follow-up to CASSANDRA-8143

stops Schema initializing tables in client mode

patch by Branimir Lambov; reviewed by Aleksey Yeschenko for
CASSANDRA-8143
---
 .../org/apache/cassandra/config/Schema.java   |   9 +-
 .../io/sstable/format/SSTableReader.java      |   6 +
 .../sstable/format/ClientModeSSTableTest.java | 135 ++++++++++++++++++
 3 files changed, 147 insertions(+), 3 deletions(-)
 create mode 100644 test/unit/org/apache/cassandra/io/sstable/format/ClientModeSSTableTest.java

diff --git a/src/java/org/apache/cassandra/config/Schema.java b/src/java/org/apache/cassandra/config/Schema.java
index bcfac1b6f0..df4e984e77 100644
--- a/src/java/org/apache/cassandra/config/Schema.java
+++ b/src/java/org/apache/cassandra/config/Schema.java
@@ -91,8 +91,11 @@ public class Schema
      */
     public Schema()
     {
-        load(SchemaKeyspace.metadata());
-        load(SystemKeyspace.metadata());
+        if (!Config.isClientMode())
+        {
+            load(SchemaKeyspace.metadata());
+            load(SystemKeyspace.metadata());
+        }
     }
 
     /**
@@ -581,7 +584,7 @@ public class Schema
         Keyspace.clear(ksm.name);
         clearKeyspaceMetadata(ksm);
 
-        keyspace.writeOrder.awaitNewBarrier();
+        Keyspace.writeOrder.awaitNewBarrier();
 
         // force a new segment in the CL
         CommitLog.instance.forceRecycleAllSegments(droppedCfs);
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 04cdc21cda..ebf28a496c 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -370,6 +370,12 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
         return open(descriptor, components, cfs.metadata, false, false); // do not track hotness
     }
 
+    // use only for offline or "Standalone" operations
+    public static SSTableReader openNoValidation(Descriptor descriptor, CFMetaData metadata) throws IOException
+    {
+        return open(descriptor, componentsFor(descriptor), metadata, false, false); // do not track hotness
+    }
+
     /**
      * Open SSTable reader to be used in batch mode(such as sstableloader).
      *
diff --git a/test/unit/org/apache/cassandra/io/sstable/format/ClientModeSSTableTest.java b/test/unit/org/apache/cassandra/io/sstable/format/ClientModeSSTableTest.java
new file mode 100644
index 0000000000..661fcd52a1
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/format/ClientModeSSTableTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.io.sstable.format;
+
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import com.google.common.util.concurrent.Runnables;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.rows.SliceableUnfilteredRowIterator;
+import org.apache.cassandra.dht.ByteOrderedPartitioner;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.format.Version;
+
+/**
+ * Tests backwards compatibility for SSTables
+ */
+public class ClientModeSSTableTest
+{
+    public static final String LEGACY_SSTABLE_PROP = "legacy-sstable-root";
+    public static final String KSNAME = "Keyspace1";
+    public static final String CFNAME = "Standard1";
+
+    public static File LEGACY_SSTABLE_ROOT;
+
+    static CFMetaData metadata;
+
+    @BeforeClass
+    public static void defineSchema() throws ConfigurationException
+    {
+        Config.setClientMode(true);
+
+        metadata = CFMetaData.Builder.createDense(KSNAME, CFNAME, false, false)
+                                                .addPartitionKey("key", BytesType.instance)
+                                                .addClusteringColumn("column", BytesType.instance)
+                                                .addRegularColumn("value", BytesType.instance)
+                                                .withPartitioner(ByteOrderedPartitioner.instance)
+                                                .build();
+
+        String scp = System.getProperty(LEGACY_SSTABLE_PROP);
+        assert scp != null;
+        LEGACY_SSTABLE_ROOT = new File(scp).getAbsoluteFile();
+        assert LEGACY_SSTABLE_ROOT.isDirectory();
+    }
+
+    /**
+     * Get a descriptor for the legacy sstable at the given version.
+     */
+    protected Descriptor getDescriptor(String ver)
+    {
+        File directory = new File(LEGACY_SSTABLE_ROOT + File.separator + ver + File.separator + KSNAME);
+        return new Descriptor(ver, directory, KSNAME, CFNAME, 0, SSTableFormat.Type.LEGACY);
+    }
+
+    @Test
+    public void testVersions() throws Throwable
+    {
+        boolean notSkipped = false;
+
+        for (File version : LEGACY_SSTABLE_ROOT.listFiles())
+        {
+            if (!new File(LEGACY_SSTABLE_ROOT + File.separator + version.getName() + File.separator + KSNAME).isDirectory())
+                continue;
+            if (Version.validate(version.getName()) && SSTableFormat.Type.LEGACY.info.getVersion(version.getName()).isCompatible())
+            {
+                notSkipped = true;
+                testVersion(version.getName());
+            }
+        }
+
+        assert notSkipped;
+    }
+
+    public void testVersion(String version) throws Throwable
+    {
+        SSTableReader reader = null;
+        try
+        {
+            reader = SSTableReader.openNoValidation(getDescriptor(version), metadata);
+
+            ByteBuffer key = bytes(Integer.toString(100));
+
+            try (SliceableUnfilteredRowIterator iter = reader.iterator(metadata.decorateKey(key), ColumnFilter.selection(metadata.partitionColumns()), false, false))
+            {
+                assert iter.next().clustering().get(0).equals(key);
+            }
+        }
+        catch (Throwable e)
+        {
+            System.err.println("Failed to read " + version);
+            throw e;
+        }
+        finally
+        {
+            if (reader != null)
+            {
+                int globalTidyCount = SSTableReader.GlobalTidy.lookup.size();
+                reader.selfRef().release();
+                assert reader.selfRef().globalCount() == 0;
+
+                // await clean-up to complete if started.
+                ScheduledExecutors.nonPeriodicTasks.submit(Runnables.doNothing()).get();
+                // Ensure clean-up completed.
+                assert SSTableReader.GlobalTidy.lookup.size() < globalTidyCount;
+            }
+        }
+    }
+}