cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [2/6] git commit: Update CQLSSTableWriter to allow parallel writing of SSTables on the same table within the same JVM
Date Wed, 29 Oct 2014 19:02:05 GMT
Update CQLSSTableWriter to allow parallel writing of SSTables on the same table within the
same JVM

Patch by Carl Yeksigian, reviewed by Benjamin Lerer for CASSANDRA-7463


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

Branch: refs/heads/cassandra-2.1
Commit: f13ce558cc410f959634a6f0d31fcf7bd69be85d
Parents: 748b01d
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Wed Oct 29 13:57:25 2014 -0500
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Wed Oct 29 13:57:25 2014 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |  2 +
 .../io/sstable/AbstractSSTableSimpleWriter.java | 12 ++-
 .../cassandra/io/sstable/CQLSSTableWriter.java  | 38 +++++---
 .../io/sstable/CQLSSTableWriterTest.java        | 93 ++++++++++++++++++++
 4 files changed, 131 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/f13ce558/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d2cb003..9051b34 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -3,6 +3,8 @@
  * Pig: Remove errant LIMIT clause in CqlNativeStorage (CASSANDRA-8166)
  * Throw ConfigurationException when hsha is used with the default
    rpc_max_threads setting of 'unlimited' (CASSANDRA-8116)
+ * Allow concurrent writing of the same table in the same JVM using
+   CQLSSTableWriter (CASSANDRA-7463)
 
 
 2.0.11:

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f13ce558/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
index 2c6f82a..af1c43c 100644
--- a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -43,6 +44,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
     protected ColumnFamily columnFamily;
     protected ByteBuffer currentSuperColumn;
     protected final CounterId counterid = CounterId.generate();
+    protected static AtomicInteger generation = new AtomicInteger(0);
 
     public AbstractSSTableSimpleWriter(File directory, CFMetaData metadata, IPartitioner
partitioner)
     {
@@ -80,9 +82,15 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
                 return false;
             }
         });
-        int maxGen = 0;
+        int maxGen = generation.getAndIncrement();
         for (Descriptor desc : existing)
-            maxGen = Math.max(maxGen, desc.generation);
+        {
+            while (desc.generation > maxGen)
+            {
+                maxGen = generation.getAndIncrement();
+            }
+        }
+
         return new Descriptor(directory, keyspace, columnFamily, maxGen + 1, true).filenameFor(Component.DATA);
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f13ce558/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index 49a1259..93d3dcf 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -28,6 +28,7 @@ import java.util.List;
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
 
 import org.apache.cassandra.cql3.statements.*;
 import org.apache.cassandra.cql3.*;
@@ -335,18 +336,31 @@ public class CQLSSTableWriter implements Closeable
         {
             try
             {
-                this.schema = getStatement(schema, CreateTableStatement.class, "CREATE TABLE").left.getCFMetaData().rebuild();
-
-                // We need to register the keyspace/table metadata through Schema, otherwise
we won't be able to properly
-                // build the insert statement in using().
-                KSMetaData ksm = KSMetaData.newKeyspace(this.schema.ksName,
-                                                        AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
-                                                        ImmutableMap.of("replication_factor",
"1"),
-                                                        true,
-                                                        Collections.singleton(this.schema));
-
-                Schema.instance.load(ksm);
-                return this;
+                synchronized (CQLSSTableWriter.class)
+                {
+                    this.schema = getStatement(schema, CreateTableStatement.class, "CREATE
TABLE").left.getCFMetaData().rebuild();
+
+                    // We need to register the keyspace/table metadata through Schema, otherwise
we won't be able to properly
+                    // build the insert statement in using().
+                    KSMetaData ksm = Schema.instance.getKSMetaData(this.schema.ksName);
+                    if (ksm == null)
+                    {
+                        ksm = KSMetaData.newKeyspace(this.schema.ksName,
+                                AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+                                ImmutableMap.of("replication_factor", "1"),
+                                true,
+                                Collections.singleton(this.schema));
+                        Schema.instance.load(ksm);
+                    }
+                    else if (Schema.instance.getCFMetaData(this.schema.ksName, this.schema.cfName)
== null)
+                    {
+                        Schema.instance.load(this.schema);
+                        ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(),
Collections.singleton(this.schema)));
+                        Schema.instance.setKeyspaceDefinition(ksm);
+                        Keyspace.open(ksm.name).initCf(this.schema.cfId, this.schema.cfName,
false);
+                    }
+                    return this;
+                }
             }
             catch (RequestValidationException e)
             {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f13ce558/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
index de814e1..0f123a4 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -158,4 +158,97 @@ public class CQLSSTableWriterTest
         };
         assert tempdir.list(filterDataFiles).length > 1 : Arrays.toString(tempdir.list(filterDataFiles));
     }
+
+
+    private static final int NUMBER_WRITES_IN_RUNNABLE = 10;
+    private class WriterThread extends Thread
+    {
+        private final File dataDir;
+        private final int id;
+        public volatile Exception exception;
+
+        public WriterThread(File dataDir, int id)
+        {
+            this.dataDir = dataDir;
+            this.id = id;
+        }
+
+        @Override
+        public void run()
+        {
+            String schema = "CREATE TABLE cql_keyspace.table2 ("
+                    + "  k int,"
+                    + "  v int,"
+                    + "  PRIMARY KEY (k, v)"
+                    + ")";
+            String insert = "INSERT INTO cql_keyspace.table2 (k, v) VALUES (?, ?)";
+            CQLSSTableWriter writer = CQLSSTableWriter.builder()
+                    .inDirectory(dataDir)
+                    .forTable(schema)
+                    .withPartitioner(StorageService.instance.getPartitioner())
+                    .using(insert).build();
+
+            try
+            {
+                for (int i = 0; i < NUMBER_WRITES_IN_RUNNABLE; i++)
+                {
+                    writer.addRow(id, i);
+                }
+                writer.close();
+            }
+            catch (Exception e)
+            {
+                exception = e;
+            }
+        }
+    }
+
+    @Test
+    public void testConcurrentWriters() throws Exception
+    {
+        String KS = "cql_keyspace";
+        String TABLE = "table2";
+
+        File tempdir = Files.createTempDir();
+        File dataDir = new File(tempdir.getAbsolutePath() + File.separator + KS + File.separator
+ TABLE);
+        assert dataDir.mkdirs();
+
+        WriterThread[] threads = new WriterThread[5];
+        for (int i = 0; i < threads.length; i++)
+        {
+            WriterThread thread = new WriterThread(dataDir, i);
+            threads[i] = thread;
+            thread.start();
+        }
+
+        for (WriterThread thread : threads)
+        {
+            thread.join();
+            assert !thread.isAlive() : "Thread should be dead by now";
+            if (thread.exception != null)
+            {
+                throw thread.exception;
+            }
+        }
+
+        SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client()
+        {
+            public void init(String keyspace)
+            {
+                for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
+                    addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
+                setPartitioner(StorageService.getPartitioner());
+            }
+
+            public CFMetaData getCFMetaData(String keyspace, String cfName)
+            {
+                return Schema.instance.getCFMetaData(keyspace, cfName);
+            }
+        }, new OutputHandler.SystemOutput(false, false));
+
+        loader.stream().get();
+
+        UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace.table2;");
+        assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
+    }
 }


Mime
View raw message