You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by br...@apache.org on 2014/10/30 16:50:22 UTC
[4/4] git commit: Update CQLSSTableWriter to allow parallel writing
of SSTables on the same table within the same JVM
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/fbb6b8ab
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fbb6b8ab
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fbb6b8ab
Branch: refs/heads/trunk
Commit: fbb6b8abfa359be3475ba55912d939096637b8f8
Parents: c4c9303
Author: Brandon Williams <br...@apache.org>
Authored: Thu Oct 30 10:49:42 2014 -0500
Committer: Brandon Williams <br...@apache.org>
Committed: Thu Oct 30 10:49:42 2014 -0500
----------------------------------------------------------------------
.../io/sstable/AbstractSSTableSimpleWriter.java | 12 ++-
.../cassandra/io/sstable/CQLSSTableWriter.java | 41 +++++----
.../io/sstable/CQLSSTableWriterTest.java | 93 ++++++++++++++++++++
3 files changed, 127 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbb6b8ab/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 11d6d5e..2854308 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;
@@ -45,6 +46,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
protected ByteBuffer currentSuperColumn;
protected final CounterId counterid = CounterId.generate();
private SSTableFormat.Type formatType = DatabaseDescriptor.getSSTableFormat();
+ protected static AtomicInteger generation = new AtomicInteger(0);
public AbstractSSTableSimpleWriter(File directory, CFMetaData metadata, IPartitioner partitioner)
@@ -83,9 +85,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, Descriptor.Type.TEMP, fmt).filenameFor(Component.DATA);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbb6b8ab/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 f7d467e..d4b4eab 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.*;
@@ -338,25 +339,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().
- if (Schema.instance.getKSMetaData(this.schema.ksName) == null)
- {
- 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);
- }
- else if (Schema.instance.getCFMetaData(this.schema.ksName, this.schema.cfName) == null)
+ synchronized (CQLSSTableWriter.class)
{
- Schema.instance.load(this.schema);
- }
+ this.schema = getStatement(schema, CreateTableStatement.class, "CREATE TABLE").left.getCFMetaData().rebuild();
- return this;
+ // 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/fbb6b8ab/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 250545a..2a41794 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -163,4 +163,97 @@ public class CQLSSTableWriterTest
};
assert dataDir.list(filterDataFiles).length > 1 : Arrays.toString(dataDir.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());
+ }
}