You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yu...@apache.org on 2015/03/05 18:52:46 UTC
[1/6] cassandra git commit: Fix CQLSSTableWriter throwing exception
and spawning threads
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.0 e56d9efb7 -> fb67c41ad
refs/heads/cassandra-2.1 b1c2536cc -> ecf48dd79
refs/heads/trunk ac75ce3e9 -> 258e59fcf
Fix CQLSSTableWriter throwing exception and spawning threads
patch by Benjamin Lerer; reviewed by yukim for CASSANDRA-8808
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/fb67c41a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fb67c41a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fb67c41a
Branch: refs/heads/cassandra-2.0
Commit: fb67c41ad7faeff7a5f33e9e0bca6493a3febe89
Parents: e56d9ef
Author: blerer <be...@datastax.com>
Authored: Tue Mar 3 12:00:57 2015 +0100
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:41:50 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cql3/statements/UpdateStatement.java | 23 ++++
.../cassandra/io/sstable/CQLSSTableWriter.java | 50 ++++++--
.../io/sstable/CQLSSTableWriterClientTest.java | 116 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 31 +++--
5 files changed, 198 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4e34c9e..faa14d5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -27,6 +27,8 @@
* 'nodetool info' prints exception against older node (CASSANDRA-8796)
* Ensure SSTableSimpleUnsortedWriter.close() terminates if
disk writer has crashed (CASSANDRA-8807)
+ * Fix CQLSSTableWriter throwing exception and spawning threads
+ (CASSANDRA-8808)
2.0.12:
* Use more efficient slice size for querying internal secondary
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 594b5db..f34edaf 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -50,6 +50,15 @@ public class UpdateStatement extends ModificationStatement
public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
throws InvalidRequestException
{
+ addUpdateForKey(cf, key, builder, params, true);
+ }
+
+ public void addUpdateForKey(ColumnFamily cf,
+ ByteBuffer key,
+ ColumnNameBuilder builder,
+ UpdateParameters params,
+ boolean validateIndexedColumns) throws InvalidRequestException
+ {
CFDefinition cfDef = cfm.getCfDef();
if (builder.getLength() > FBUtilities.MAX_UNSIGNED_SHORT)
@@ -106,6 +115,20 @@ public class UpdateStatement extends ModificationStatement
update.execute(key, cf, builder.copy(), params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
+ * Checks that the value of the indexed columns is valid.
+ *
+ * @param cf the column family
+ * @throws InvalidRequestException if one of the values is invalid
+ */
+ private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 8006112..fb4c186 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -37,9 +37,9 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.RequestValidationException;
-import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.Allocator;
@@ -77,6 +77,11 @@ import org.apache.cassandra.utils.Pair;
*/
public class CQLSSTableWriter implements Closeable
{
+ static
+ {
+ Config.setClientMode(true);
+ }
+
private final AbstractSSTableSimpleWriter writer;
private final UpdateStatement insert;
private final List<ColumnSpecification> boundNames;
@@ -215,7 +220,7 @@ public class CQLSSTableWriter implements Closeable
{
if (writer.currentKey() == null || !key.equals(writer.currentKey().key))
writer.newRow(key);
- insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params);
+ insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params, false);
}
return this;
}
@@ -345,19 +350,11 @@ public class CQLSSTableWriter implements Closeable
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);
+ createKeyspaceWithColumnFamily(this.schema);
}
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);
+ addColumnFamilyToKeyspace(ksm, this.schema);
}
return this;
}
@@ -369,6 +366,35 @@ public class CQLSSTableWriter implements Closeable
}
/**
+ * Adds the specified column family to the specified keyspace.
+ *
+ * @param ksm the keyspace meta data
+ * @param cfm the column family meta data
+ */
+ private static void addColumnFamilyToKeyspace(KSMetaData ksm, CFMetaData cfm)
+ {
+ ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
+ Schema.instance.load(cfm);
+ Schema.instance.setKeyspaceDefinition(ksm);
+ }
+
+ /**
+ * Creates a keyspace for the specified column family.
+ *
+ * @param cfm the column family
+ * @throws ConfigurationException if a problem occurs while creating the keyspace.
+ */
+ private static void createKeyspaceWithColumnFamily(CFMetaData cfm) throws ConfigurationException
+ {
+ KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
+ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+ ImmutableMap.of("replication_factor", "1"),
+ true,
+ Collections.singleton(cfm));
+ Schema.instance.load(ksm);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
new file mode 100644
index 0000000..d10c9fb
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.util.Arrays;
+
+import com.google.common.io.Files;
+
+import org.junit.*;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.util.FileUtils;
+
+import static org.junit.Assert.assertEquals;
+
+import static org.junit.Assert.assertTrue;
+
+public class CQLSSTableWriterClientTest
+{
+ private File testDirectory;
+
+ @Before
+ public void setUp()
+ {
+ this.testDirectory = Files.createTempDir();
+ }
+
+ @After
+ public void tearDown()
+ {
+ FileUtils.deleteRecursive(this.testDirectory);
+ }
+
+ @AfterClass
+ public static void cleanup() throws Exception
+ {
+ Config.setClientMode(false);
+ }
+
+ @Test
+ public void testWriterInClientMode() throws IOException, InvalidRequestException
+ {
+ final String TABLE1 = "table1";
+ final String TABLE2 = "table2";
+
+ String schema = "CREATE TABLE client_test.%s ("
+ + " k int PRIMARY KEY,"
+ + " v1 text,"
+ + " v2 int"
+ + ")";
+ String insert = "INSERT INTO client_test.%s (k, v1, v2) VALUES (?, ?, ?)";
+
+ CQLSSTableWriter writer = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE1))
+ .using(String.format(insert, TABLE1)).build();
+
+ CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE2))
+ .using(String.format(insert, TABLE2)).build();
+
+ writer.addRow(0, "A", 0);
+ writer2.addRow(0, "A", 0);
+ writer.addRow(1, "B", 1);
+ writer2.addRow(1, "B", 1);
+ writer.close();
+ writer2.close();
+
+ assertContainsDataFiles(this.testDirectory, "client_test-table1", "client_test-table2");
+ }
+
+ /**
+ * Checks that the specified directory contains the files with the specified prefixes.
+ *
+ * @param directory the directory containing the data files
+ * @param prefixes the file prefixes
+ */
+ private static void assertContainsDataFiles(File directory, String... prefixes)
+ {
+ FilenameFilter filter = new FilenameFilter()
+ {
+ @Override
+ public boolean accept(File dir, String name)
+ {
+ return name.endsWith("-Data.db");
+ }
+ };
+
+ File[] dataFiles = directory.listFiles(filter);
+ Arrays.sort(dataFiles);
+
+ assertEquals(dataFiles.length, prefixes.length);
+ for (int i = 0; i < dataFiles.length; i++)
+ assertTrue(dataFiles[i].toString().contains(prefixes[i]));
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 0f123a4..0922502 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -25,24 +25,25 @@ import java.util.Iterator;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Files;
+
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import org.apache.cassandra.SchemaLoader;
-import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.OutputHandler;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
public class CQLSSTableWriterTest
{
@BeforeClass
@@ -51,6 +52,12 @@ public class CQLSSTableWriterTest
StorageService.instance.initServer();
}
+ @AfterClass
+ public static void tearDown()
+ {
+ Config.setClientMode(false);
+ }
+
@Test
public void testUnsortedWriter() throws Exception
{
@@ -176,12 +183,12 @@ public class CQLSSTableWriterTest
@Override
public void run()
{
- String schema = "CREATE TABLE cql_keyspace.table2 ("
+ String schema = "CREATE TABLE cql_keyspace2.table2 ("
+ " k int,"
+ " v int,"
+ " PRIMARY KEY (k, v)"
+ ")";
- String insert = "INSERT INTO cql_keyspace.table2 (k, v) VALUES (?, ?)";
+ String insert = "INSERT INTO cql_keyspace2.table2 (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
@@ -206,7 +213,7 @@ public class CQLSSTableWriterTest
@Test
public void testConcurrentWriters() throws Exception
{
- String KS = "cql_keyspace";
+ String KS = "cql_keyspace2";
String TABLE = "table2";
File tempdir = Files.createTempDir();
@@ -235,7 +242,7 @@ public class CQLSSTableWriterTest
{
public void init(String keyspace)
{
- for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
+ for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace2"))
addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
setPartitioner(StorageService.getPartitioner());
}
@@ -248,7 +255,7 @@ public class CQLSSTableWriterTest
loader.stream().get();
- UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace.table2;");
+ UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace2.table2;");
assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
}
}
[6/6] cassandra git commit: Merge branch 'cassandra-2.1' into trunk
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/258e59fc
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/258e59fc
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/258e59fc
Branch: refs/heads/trunk
Commit: 258e59fcf3eae9f808e9c37c0aee2c54458e3cf9
Parents: ac75ce3 ecf48dd
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 5 11:52:37 2015 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:52:37 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/cassandra/config/Config.java | 12 +++
.../cassandra/config/DatabaseDescriptor.java | 12 ++-
.../cql3/statements/UpdateStatement.java | 41 ++++++--
src/java/org/apache/cassandra/db/Keyspace.java | 8 +-
.../cassandra/io/sstable/CQLSSTableWriter.java | 56 ++++++----
.../io/sstable/CQLSSTableWriterClientTest.java | 102 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 23 +++--
8 files changed, 215 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/config/Config.java
index 7e1e9ee,c683d7b..a174a0f
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@@ -205,9 -200,10 +205,11 @@@ public class Confi
public volatile int counter_cache_save_period = 7200;
public volatile int counter_cache_keys_to_save = Integer.MAX_VALUE;
- public String memory_allocator = NativeAllocator.class.getSimpleName();
+ @Deprecated
+ public String memory_allocator;
+ private static boolean isClientMode = false;
+
public Integer file_cache_size_in_mb;
public boolean inter_dc_tcp_nodelay = true;
@@@ -239,6 -231,16 +241,16 @@@
outboundBindAny = value;
}
+ public static boolean isClientMode()
+ {
- return isClientMode;
++ return isClientMode;
+ }
+
+ public static void setClientMode(boolean clientMode)
+ {
+ isClientMode = clientMode;
+ }
+
public void configHintedHandoff() throws ConfigurationException
{
if (hinted_handoff_enabled != null && !hinted_handoff_enabled.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index b1a2684,924ab3c..af21f74
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@@ -92,16 -115,36 +92,26 @@@ public class DatabaseDescripto
private static String localDC;
private static Comparator<InetAddress> localComparator;
+ public static void forceStaticInitialization() {}
static
{
+ // In client mode, we use a default configuration. Note that the fields of this class will be
+ // left unconfigured however (the partitioner or localDC will be null for instance) so this
+ // should be used with care.
try
{
- applyConfig(loadConfig());
+ if (Config.isClientMode())
+ {
+ conf = new Config();
- // at least we have to set memoryAllocator to open SSTable in client mode
- memoryAllocator = FBUtilities.newOffHeapAllocator(conf.memory_allocator);
+ }
+ else
+ {
+ applyConfig(loadConfig());
+ }
}
- catch (ConfigurationException e)
- {
- logger.error("Fatal configuration error", e);
- System.err.println(e.getMessage() + "\nFatal configuration error; unable to start. See log for stacktrace.");
- System.exit(1);
- }
catch (Exception e)
{
- logger.error("Fatal error during configuration loading", e);
- System.err.println(e.getMessage() + "\nFatal error during configuration loading; unable to start. See log for stacktrace.");
- JVMStabilityInspector.inspectThrowable(e);
- System.exit(1);
+ throw new ExceptionInInitializerError(e);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index b7156a0,63f87c0..c783d48
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@@ -30,6 -30,6 +30,8 @@@ import org.apache.cassandra.exceptions.
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
++import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
++
/**
* An <code>UPDATE</code> statement parsed from a CQL query statement.
*
@@@ -98,6 -109,20 +111,20 @@@ public class UpdateStatement extends Mo
update.execute(key, cf, prefix, params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
- * Checks that the value of the indexed columns is valid.
++ * Checks if the values of the indexed columns are valid.
+ *
+ * @param cf the column family
- * @throws InvalidRequestException if one of the values is invalid
++ * @throws InvalidRequestException if one of the values of the indexed columns is not valid
+ */
- private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
++ private void validateIndexedColumns(ColumnFamily cf)
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
@@@ -105,11 -130,11 +132,11 @@@
{
// Indexed values must be validated by any applicable index. See CASSANDRA-3057/4240/8081 for more details
if (!indexManager.validate(cell))
-- throw new InvalidRequestException(String.format("Can't index column value of size %d for index %s on %s.%s",
-- cell.value().remaining(),
-- cfm.getColumnDefinition(cell.name()).getIndexName(),
-- cfm.ksName,
-- cfm.cfName));
++ throw invalidRequest("Can't index column value of size %d for index %s on %s.%s",
++ cell.value().remaining(),
++ cfm.getColumnDefinition(cell.name()).getIndexName(),
++ cfm.ksName,
++ cfm.cfName);
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/src/java/org/apache/cassandra/db/Keyspace.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/Keyspace.java
index bffddc6,d92eea4..e3301b1
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@@ -67,7 -65,8 +64,8 @@@ public class Keyspac
// proper directories here as well as in CassandraDaemon.
static
{
- DatabaseDescriptor.createAllDirectories();
- if (!(Config.isClientMode() || StorageService.instance.isClientMode()))
++ if (!Config.isClientMode())
+ DatabaseDescriptor.createAllDirectories();
}
public final KSMetaData metadata;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index d137fe2,b4bef7d..53cb094
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@@ -21,22 -21,19 +21,21 @@@ import java.io.Closeable
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Iterables;
-import org.apache.cassandra.cql3.statements.*;
+import org.apache.cassandra.config.CFMetaData;
++import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.config.*;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.cql3.statements.CreateTableStatement;
+import org.apache.cassandra.cql3.statements.ParsedStatement;
+import org.apache.cassandra.cql3.statements.UpdateStatement;
+import org.apache.cassandra.db.ArrayBackedSortedColumns;
+import org.apache.cassandra.db.Cell;
+import org.apache.cassandra.db.ColumnFamily;
- import org.apache.cassandra.db.Keyspace;
- import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.composites.Composite;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
@@@ -388,6 -367,35 +371,35 @@@ public class CQLSSTableWriter implement
}
/**
- * Adds the specified column family to the specified keyspace.
++ * Creates the keyspace with the specified table.
+ *
- * @param ksm the keyspace meta data
- * @param cfm the column family meta data
++ * @param the table the table that must be created.
+ */
- private static void addTableToKeyspace(KSMetaData ksm, CFMetaData cfm)
++ private static void createKeyspaceWithTable(CFMetaData table)
+ {
- ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
- Schema.instance.load(cfm);
- Schema.instance.setKeyspaceDefinition(ksm);
++ KSMetaData ksm;
++ ksm = KSMetaData.newKeyspace(table.ksName,
++ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
++ ImmutableMap.of("replication_factor", "1"),
++ true,
++ Collections.singleton(table));
++ Schema.instance.load(ksm);
+ }
+
+ /**
- * Creates a keyspace for the specified column family.
++ * Adds the table to the to the specified keyspace.
+ *
- * @param cfm the column family
- * @throws ConfigurationException if a problem occurs while creating the keyspace.
++ * @param keyspace the keyspace to add to
++ * @param table the table to add
+ */
- private static void createKeyspaceWithTable(CFMetaData cfm) throws ConfigurationException
++ private static void addTableToKeyspace(KSMetaData keyspace, CFMetaData table)
+ {
- KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
- AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
- ImmutableMap.of("replication_factor", "1"),
- true,
- Collections.singleton(cfm));
- Schema.instance.load(ksm);
++ KSMetaData clone = keyspace.cloneWithTableAdded(table);
++ Schema.instance.load(table);
++ Schema.instance.setKeyspaceDefinition(clone);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
index 0000000,d10c9fb..ad2d876
mode 000000,100644..100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
@@@ -1,0 -1,116 +1,102 @@@
+ /*
+ * 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;
+
+ import java.io.File;
+ import java.io.FilenameFilter;
+ import java.io.IOException;
+ import java.util.Arrays;
+
+ import com.google.common.io.Files;
+
+ import org.junit.*;
+
+ import org.apache.cassandra.config.Config;
+ import org.apache.cassandra.exceptions.InvalidRequestException;
+ import org.apache.cassandra.io.util.FileUtils;
+
+ import static org.junit.Assert.assertEquals;
+
+ import static org.junit.Assert.assertTrue;
+
+ public class CQLSSTableWriterClientTest
+ {
+ private File testDirectory;
+
+ @Before
+ public void setUp()
+ {
+ this.testDirectory = Files.createTempDir();
+ }
+
+ @After
+ public void tearDown()
+ {
+ FileUtils.deleteRecursive(this.testDirectory);
+ }
+
+ @AfterClass
+ public static void cleanup() throws Exception
+ {
+ Config.setClientMode(false);
+ }
+
+ @Test
+ public void testWriterInClientMode() throws IOException, InvalidRequestException
+ {
+ final String TABLE1 = "table1";
+ final String TABLE2 = "table2";
+
+ String schema = "CREATE TABLE client_test.%s ("
+ + " k int PRIMARY KEY,"
+ + " v1 text,"
+ + " v2 int"
+ + ")";
+ String insert = "INSERT INTO client_test.%s (k, v1, v2) VALUES (?, ?, ?)";
+
+ CQLSSTableWriter writer = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE1))
+ .using(String.format(insert, TABLE1)).build();
+
+ CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE2))
+ .using(String.format(insert, TABLE2)).build();
+
+ writer.addRow(0, "A", 0);
+ writer2.addRow(0, "A", 0);
+ writer.addRow(1, "B", 1);
+ writer2.addRow(1, "B", 1);
+ writer.close();
+ writer2.close();
+
- assertContainsDataFiles(this.testDirectory, "client_test-table1", "client_test-table2");
- }
-
- /**
- * Checks that the specified directory contains the files with the specified prefixes.
- *
- * @param directory the directory containing the data files
- * @param prefixes the file prefixes
- */
- private static void assertContainsDataFiles(File directory, String... prefixes)
- {
+ FilenameFilter filter = new FilenameFilter()
+ {
+ @Override
+ public boolean accept(File dir, String name)
+ {
+ return name.endsWith("-Data.db");
+ }
+ };
+
- File[] dataFiles = directory.listFiles(filter);
- Arrays.sort(dataFiles);
++ File[] dataFiles = this.testDirectory.listFiles(filter);
++ assertEquals(2, dataFiles.length);
+
- assertEquals(dataFiles.length, prefixes.length);
- for (int i = 0; i < dataFiles.length; i++)
- assertTrue(dataFiles[i].toString().contains(prefixes[i]));
+ }
+ }
http://git-wip-us.apache.org/repos/asf/cassandra/blob/258e59fc/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
index 8f879d9,a2f1bcc..6278b45
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@@ -51,6 -56,12 +56,12 @@@ public class CQLSSTableWriterTes
StorageService.instance.initServer();
}
+ @AfterClass
- public static void tearDown()
++ public static void tearDown() throws Exception
+ {
+ Config.setClientMode(false);
+ }
+
@Test
public void testUnsortedWriter() throws Exception
{
@@@ -211,8 -217,8 +222,8 @@@
@Test
public void testConcurrentWriters() throws Exception
{
- String KS = "cql_keyspace";
- String KS = "cql_keyspace2";
-- String TABLE = "table2";
++ final String KS = "cql_keyspace2";
++ final String TABLE = "table2";
File tempdir = Files.createTempDir();
File dataDir = new File(tempdir.getAbsolutePath() + File.separator + KS + File.separator + TABLE);
@@@ -240,7 -246,7 +251,7 @@@
{
public void init(String keyspace)
{
- for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
- for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace2"))
++ for (Range<Token> range : StorageService.instance.getLocalRanges(KS))
addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
setPartitioner(StorageService.getPartitioner());
}
[2/6] cassandra git commit: Fix CQLSSTableWriter throwing exception
and spawning threads
Posted by yu...@apache.org.
Fix CQLSSTableWriter throwing exception and spawning threads
patch by Benjamin Lerer; reviewed by yukim for CASSANDRA-8808
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/fb67c41a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fb67c41a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fb67c41a
Branch: refs/heads/cassandra-2.1
Commit: fb67c41ad7faeff7a5f33e9e0bca6493a3febe89
Parents: e56d9ef
Author: blerer <be...@datastax.com>
Authored: Tue Mar 3 12:00:57 2015 +0100
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:41:50 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cql3/statements/UpdateStatement.java | 23 ++++
.../cassandra/io/sstable/CQLSSTableWriter.java | 50 ++++++--
.../io/sstable/CQLSSTableWriterClientTest.java | 116 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 31 +++--
5 files changed, 198 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4e34c9e..faa14d5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -27,6 +27,8 @@
* 'nodetool info' prints exception against older node (CASSANDRA-8796)
* Ensure SSTableSimpleUnsortedWriter.close() terminates if
disk writer has crashed (CASSANDRA-8807)
+ * Fix CQLSSTableWriter throwing exception and spawning threads
+ (CASSANDRA-8808)
2.0.12:
* Use more efficient slice size for querying internal secondary
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 594b5db..f34edaf 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -50,6 +50,15 @@ public class UpdateStatement extends ModificationStatement
public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
throws InvalidRequestException
{
+ addUpdateForKey(cf, key, builder, params, true);
+ }
+
+ public void addUpdateForKey(ColumnFamily cf,
+ ByteBuffer key,
+ ColumnNameBuilder builder,
+ UpdateParameters params,
+ boolean validateIndexedColumns) throws InvalidRequestException
+ {
CFDefinition cfDef = cfm.getCfDef();
if (builder.getLength() > FBUtilities.MAX_UNSIGNED_SHORT)
@@ -106,6 +115,20 @@ public class UpdateStatement extends ModificationStatement
update.execute(key, cf, builder.copy(), params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
+ * Checks that the value of the indexed columns is valid.
+ *
+ * @param cf the column family
+ * @throws InvalidRequestException if one of the values is invalid
+ */
+ private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 8006112..fb4c186 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -37,9 +37,9 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.RequestValidationException;
-import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.Allocator;
@@ -77,6 +77,11 @@ import org.apache.cassandra.utils.Pair;
*/
public class CQLSSTableWriter implements Closeable
{
+ static
+ {
+ Config.setClientMode(true);
+ }
+
private final AbstractSSTableSimpleWriter writer;
private final UpdateStatement insert;
private final List<ColumnSpecification> boundNames;
@@ -215,7 +220,7 @@ public class CQLSSTableWriter implements Closeable
{
if (writer.currentKey() == null || !key.equals(writer.currentKey().key))
writer.newRow(key);
- insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params);
+ insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params, false);
}
return this;
}
@@ -345,19 +350,11 @@ public class CQLSSTableWriter implements Closeable
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);
+ createKeyspaceWithColumnFamily(this.schema);
}
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);
+ addColumnFamilyToKeyspace(ksm, this.schema);
}
return this;
}
@@ -369,6 +366,35 @@ public class CQLSSTableWriter implements Closeable
}
/**
+ * Adds the specified column family to the specified keyspace.
+ *
+ * @param ksm the keyspace meta data
+ * @param cfm the column family meta data
+ */
+ private static void addColumnFamilyToKeyspace(KSMetaData ksm, CFMetaData cfm)
+ {
+ ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
+ Schema.instance.load(cfm);
+ Schema.instance.setKeyspaceDefinition(ksm);
+ }
+
+ /**
+ * Creates a keyspace for the specified column family.
+ *
+ * @param cfm the column family
+ * @throws ConfigurationException if a problem occurs while creating the keyspace.
+ */
+ private static void createKeyspaceWithColumnFamily(CFMetaData cfm) throws ConfigurationException
+ {
+ KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
+ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+ ImmutableMap.of("replication_factor", "1"),
+ true,
+ Collections.singleton(cfm));
+ Schema.instance.load(ksm);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
new file mode 100644
index 0000000..d10c9fb
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.util.Arrays;
+
+import com.google.common.io.Files;
+
+import org.junit.*;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.util.FileUtils;
+
+import static org.junit.Assert.assertEquals;
+
+import static org.junit.Assert.assertTrue;
+
+public class CQLSSTableWriterClientTest
+{
+ private File testDirectory;
+
+ @Before
+ public void setUp()
+ {
+ this.testDirectory = Files.createTempDir();
+ }
+
+ @After
+ public void tearDown()
+ {
+ FileUtils.deleteRecursive(this.testDirectory);
+ }
+
+ @AfterClass
+ public static void cleanup() throws Exception
+ {
+ Config.setClientMode(false);
+ }
+
+ @Test
+ public void testWriterInClientMode() throws IOException, InvalidRequestException
+ {
+ final String TABLE1 = "table1";
+ final String TABLE2 = "table2";
+
+ String schema = "CREATE TABLE client_test.%s ("
+ + " k int PRIMARY KEY,"
+ + " v1 text,"
+ + " v2 int"
+ + ")";
+ String insert = "INSERT INTO client_test.%s (k, v1, v2) VALUES (?, ?, ?)";
+
+ CQLSSTableWriter writer = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE1))
+ .using(String.format(insert, TABLE1)).build();
+
+ CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE2))
+ .using(String.format(insert, TABLE2)).build();
+
+ writer.addRow(0, "A", 0);
+ writer2.addRow(0, "A", 0);
+ writer.addRow(1, "B", 1);
+ writer2.addRow(1, "B", 1);
+ writer.close();
+ writer2.close();
+
+ assertContainsDataFiles(this.testDirectory, "client_test-table1", "client_test-table2");
+ }
+
+ /**
+ * Checks that the specified directory contains the files with the specified prefixes.
+ *
+ * @param directory the directory containing the data files
+ * @param prefixes the file prefixes
+ */
+ private static void assertContainsDataFiles(File directory, String... prefixes)
+ {
+ FilenameFilter filter = new FilenameFilter()
+ {
+ @Override
+ public boolean accept(File dir, String name)
+ {
+ return name.endsWith("-Data.db");
+ }
+ };
+
+ File[] dataFiles = directory.listFiles(filter);
+ Arrays.sort(dataFiles);
+
+ assertEquals(dataFiles.length, prefixes.length);
+ for (int i = 0; i < dataFiles.length; i++)
+ assertTrue(dataFiles[i].toString().contains(prefixes[i]));
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 0f123a4..0922502 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -25,24 +25,25 @@ import java.util.Iterator;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Files;
+
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import org.apache.cassandra.SchemaLoader;
-import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.OutputHandler;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
public class CQLSSTableWriterTest
{
@BeforeClass
@@ -51,6 +52,12 @@ public class CQLSSTableWriterTest
StorageService.instance.initServer();
}
+ @AfterClass
+ public static void tearDown()
+ {
+ Config.setClientMode(false);
+ }
+
@Test
public void testUnsortedWriter() throws Exception
{
@@ -176,12 +183,12 @@ public class CQLSSTableWriterTest
@Override
public void run()
{
- String schema = "CREATE TABLE cql_keyspace.table2 ("
+ String schema = "CREATE TABLE cql_keyspace2.table2 ("
+ " k int,"
+ " v int,"
+ " PRIMARY KEY (k, v)"
+ ")";
- String insert = "INSERT INTO cql_keyspace.table2 (k, v) VALUES (?, ?)";
+ String insert = "INSERT INTO cql_keyspace2.table2 (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
@@ -206,7 +213,7 @@ public class CQLSSTableWriterTest
@Test
public void testConcurrentWriters() throws Exception
{
- String KS = "cql_keyspace";
+ String KS = "cql_keyspace2";
String TABLE = "table2";
File tempdir = Files.createTempDir();
@@ -235,7 +242,7 @@ public class CQLSSTableWriterTest
{
public void init(String keyspace)
{
- for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
+ for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace2"))
addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
setPartitioner(StorageService.getPartitioner());
}
@@ -248,7 +255,7 @@ public class CQLSSTableWriterTest
loader.stream().get();
- UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace.table2;");
+ UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace2.table2;");
assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
}
}
[4/6] cassandra git commit: Merge branch 'cassandra-2.0' into
cassandra-2.1
Posted by yu...@apache.org.
Merge branch 'cassandra-2.0' into cassandra-2.1
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ecf48dd7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ecf48dd7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ecf48dd7
Branch: refs/heads/trunk
Commit: ecf48dd79d29ba80008afa26fd40c3fc6d9509ac
Parents: b1c2536 fb67c41
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 5 11:48:22 2015 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:48:22 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cql3/statements/UpdateStatement.java | 29 ++++-
.../cassandra/io/sstable/CQLSSTableWriter.java | 56 +++++----
.../io/sstable/CQLSSTableWriterClientTest.java | 116 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 21 +++-
5 files changed, 195 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index aa2e1f9,faa14d5..ea79e22
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -42,105 -10,10 +42,107 @@@ Merged from 2.0
table with ASC ordering and paging (CASSANDRA-8767)
* AssertionError: "Memory was freed" when running cleanup (CASSANDRA-8716)
* Make it possible to set max_sstable_age to fractional days (CASSANDRA-8406)
- * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate()
- (CASSANDRA-8748)
* Fix some multi-column relations with indexes on some clustering
columns (CASSANDRA-8275)
+ * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate()
+ (CASSANDRA-8748)
+ * Throw OOM if allocating memory fails to return a valid pointer (CASSANDRA-8726)
+ * Fix SSTableSimpleUnsortedWriter ConcurrentModificationException (CASSANDRA-8619)
+ * 'nodetool info' prints exception against older node (CASSANDRA-8796)
+ * Ensure SSTableSimpleUnsortedWriter.close() terminates if
+ disk writer has crashed (CASSANDRA-8807)
++ * Fix CQLSSTableWriter throwing exception and spawning threads
++ (CASSANDRA-8808)
+
+
+2.1.3
+ * Fix HSHA/offheap_objects corruption (CASSANDRA-8719)
+ * Upgrade libthrift to 0.9.2 (CASSANDRA-8685)
+ * Don't use the shared ref in sstableloader (CASSANDRA-8704)
+ * Purge internal prepared statements if related tables or
+ keyspaces are dropped (CASSANDRA-8693)
+ * (cqlsh) Handle unicode BOM at start of files (CASSANDRA-8638)
+ * Stop compactions before exiting offline tools (CASSANDRA-8623)
+ * Update tools/stress/README.txt to match current behaviour (CASSANDRA-7933)
+ * Fix schema from Thrift conversion with empty metadata (CASSANDRA-8695)
+ * Safer Resource Management (CASSANDRA-7705)
+ * Make sure we compact highly overlapping cold sstables with
+ STCS (CASSANDRA-8635)
+ * rpc_interface and listen_interface generate NPE on startup when specified
+ interface doesn't exist (CASSANDRA-8677)
+ * Fix ArrayIndexOutOfBoundsException in nodetool cfhistograms (CASSANDRA-8514)
+ * Switch from yammer metrics for nodetool cf/proxy histograms (CASSANDRA-8662)
+ * Make sure we don't add tmplink files to the compaction
+ strategy (CASSANDRA-8580)
+ * (cqlsh) Handle maps with blob keys (CASSANDRA-8372)
+ * (cqlsh) Handle DynamicCompositeType schemas correctly (CASSANDRA-8563)
+ * Duplicate rows returned when in clause has repeated values (CASSANDRA-6706)
+ * Add tooling to detect hot partitions (CASSANDRA-7974)
+ * Fix cassandra-stress user-mode truncation of partition generation (CASSANDRA-8608)
+ * Only stream from unrepaired sstables during inc repair (CASSANDRA-8267)
+ * Don't allow starting multiple inc repairs on the same sstables (CASSANDRA-8316)
+ * Invalidate prepared BATCH statements when related tables
+ or keyspaces are dropped (CASSANDRA-8652)
+ * Fix missing results in secondary index queries on collections
+ with ALLOW FILTERING (CASSANDRA-8421)
+ * Expose EstimatedHistogram metrics for range slices (CASSANDRA-8627)
+ * (cqlsh) Escape clqshrc passwords properly (CASSANDRA-8618)
+ * Fix NPE when passing wrong argument in ALTER TABLE statement (CASSANDRA-8355)
+ * Pig: Refactor and deprecate CqlStorage (CASSANDRA-8599)
+ * Don't reuse the same cleanup strategy for all sstables (CASSANDRA-8537)
+ * Fix case-sensitivity of index name on CREATE and DROP INDEX
+ statements (CASSANDRA-8365)
+ * Better detection/logging for corruption in compressed sstables (CASSANDRA-8192)
+ * Use the correct repairedAt value when closing writer (CASSANDRA-8570)
+ * (cqlsh) Handle a schema mismatch being detected on startup (CASSANDRA-8512)
+ * Properly calculate expected write size during compaction (CASSANDRA-8532)
+ * Invalidate affected prepared statements when a table's columns
+ are altered (CASSANDRA-7910)
+ * Stress - user defined writes should populate sequentally (CASSANDRA-8524)
+ * Fix regression in SSTableRewriter causing some rows to become unreadable
+ during compaction (CASSANDRA-8429)
+ * Run major compactions for repaired/unrepaired in parallel (CASSANDRA-8510)
+ * (cqlsh) Fix compression options in DESCRIBE TABLE output when compression
+ is disabled (CASSANDRA-8288)
+ * (cqlsh) Fix DESCRIBE output after keyspaces are altered (CASSANDRA-7623)
+ * Make sure we set lastCompactedKey correctly (CASSANDRA-8463)
+ * (cqlsh) Fix output of CONSISTENCY command (CASSANDRA-8507)
+ * (cqlsh) Fixed the handling of LIST statements (CASSANDRA-8370)
+ * Make sstablescrub check leveled manifest again (CASSANDRA-8432)
+ * Check first/last keys in sstable when giving out positions (CASSANDRA-8458)
+ * Disable mmap on Windows (CASSANDRA-6993)
+ * Add missing ConsistencyLevels to cassandra-stress (CASSANDRA-8253)
+ * Add auth support to cassandra-stress (CASSANDRA-7985)
+ * Fix ArrayIndexOutOfBoundsException when generating error message
+ for some CQL syntax errors (CASSANDRA-8455)
+ * Scale memtable slab allocation logarithmically (CASSANDRA-7882)
+ * cassandra-stress simultaneous inserts over same seed (CASSANDRA-7964)
+ * Reduce cassandra-stress sampling memory requirements (CASSANDRA-7926)
+ * Ensure memtable flush cannot expire commit log entries from its future (CASSANDRA-8383)
+ * Make read "defrag" async to reclaim memtables (CASSANDRA-8459)
+ * Remove tmplink files for offline compactions (CASSANDRA-8321)
+ * Reduce maxHintsInProgress (CASSANDRA-8415)
+ * BTree updates may call provided update function twice (CASSANDRA-8018)
+ * Release sstable references after anticompaction (CASSANDRA-8386)
+ * Handle abort() in SSTableRewriter properly (CASSANDRA-8320)
+ * Fix high size calculations for prepared statements (CASSANDRA-8231)
+ * Centralize shared executors (CASSANDRA-8055)
+ * Fix filtering for CONTAINS (KEY) relations on frozen collection
+ clustering columns when the query is restricted to a single
+ partition (CASSANDRA-8203)
+ * Do more aggressive entire-sstable TTL expiry checks (CASSANDRA-8243)
+ * Add more log info if readMeter is null (CASSANDRA-8238)
+ * add check of the system wall clock time at startup (CASSANDRA-8305)
+ * Support for frozen collections (CASSANDRA-7859)
+ * Fix overflow on histogram computation (CASSANDRA-8028)
+ * Have paxos reuse the timestamp generation of normal queries (CASSANDRA-7801)
+ * Fix incremental repair not remove parent session on remote (CASSANDRA-8291)
+ * Improve JBOD disk utilization (CASSANDRA-7386)
+ * Log failed host when preparing incremental repair (CASSANDRA-8228)
+ * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
+ * Fix sstableupgrade throws exception (CASSANDRA-8688)
+ * Fix hang when repairing empty keyspace (CASSANDRA-8694)
+Merged from 2.0:
* Fix IllegalArgumentException in dynamic snitch (CASSANDRA-8448)
* Add support for UPDATE ... IF EXISTS (CASSANDRA-8610)
* Fix reversal of list prepends (CASSANDRA-8733)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index c35f985,f34edaf..63f87c0
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@@ -48,9 -47,25 +48,20 @@@ public class UpdateStatement extends Mo
return true;
}
- public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, Composite prefix, UpdateParameters params)
- public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
-- throws InvalidRequestException
++ public void addUpdateForKey(ColumnFamily cf,
++ ByteBuffer key,
++ Composite prefix,
++ UpdateParameters params) throws InvalidRequestException
+ {
- addUpdateForKey(cf, key, builder, params, true);
++ addUpdateForKey(cf, key, prefix, params, true);
+ }
+
+ public void addUpdateForKey(ColumnFamily cf,
+ ByteBuffer key,
- ColumnNameBuilder builder,
++ Composite prefix,
+ UpdateParameters params,
+ boolean validateIndexedColumns) throws InvalidRequestException
{
- CFDefinition cfDef = cfm.getCfDef();
-
- if (builder.getLength() > FBUtilities.MAX_UNSIGNED_SHORT)
- throw new InvalidRequestException(String.format("The sum of all clustering columns is too long (%s > %s)",
- builder.getLength(),
- FBUtilities.MAX_UNSIGNED_SHORT));
-
// Inserting the CQL row marker (see #4361)
// We always need to insert a marker for INSERT, because of the following situation:
// CREATE TABLE t ( k int PRIMARY KEY, c text );
@@@ -95,9 -112,23 +106,23 @@@
else
{
for (Operation update : updates)
- update.execute(key, cf, builder.copy(), params);
+ update.execute(key, cf, prefix, params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
+ * Checks that the value of the indexed columns is valid.
+ *
+ * @param cf the column family
+ * @throws InvalidRequestException if one of the values is invalid
+ */
+ private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index 3e27ff9,fb4c186..b4bef7d
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@@ -34,8 -34,6 +34,7 @@@ import org.apache.cassandra.cql3.statem
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.*;
- import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.composites.Composite;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
@@@ -218,11 -216,11 +217,11 @@@ public class CQLSSTableWriter implement
try
{
- for (ByteBuffer key: keys)
+ for (ByteBuffer key : keys)
{
- if (writer.currentKey() == null || !key.equals(writer.currentKey().key))
+ if (writer.currentKey() == null || !key.equals(writer.currentKey().getKey()))
writer.newRow(key);
- insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params);
+ insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params, false);
}
return this;
}
@@@ -360,19 -350,11 +351,11 @@@
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);
- createKeyspaceWithColumnFamily(this.schema);
++ createKeyspaceWithTable(this.schema);
}
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);
- addColumnFamilyToKeyspace(ksm, this.schema);
++ addTableToKeyspace(ksm, this.schema);
}
return this;
}
@@@ -384,6 -366,35 +367,35 @@@
}
/**
+ * Adds the specified column family to the specified keyspace.
+ *
+ * @param ksm the keyspace meta data
+ * @param cfm the column family meta data
+ */
- private static void addColumnFamilyToKeyspace(KSMetaData ksm, CFMetaData cfm)
++ private static void addTableToKeyspace(KSMetaData ksm, CFMetaData cfm)
+ {
+ ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
+ Schema.instance.load(cfm);
+ Schema.instance.setKeyspaceDefinition(ksm);
+ }
+
+ /**
+ * Creates a keyspace for the specified column family.
+ *
+ * @param cfm the column family
+ * @throws ConfigurationException if a problem occurs while creating the keyspace.
+ */
- private static void createKeyspaceWithColumnFamily(CFMetaData cfm) throws ConfigurationException
++ private static void createKeyspaceWithTable(CFMetaData cfm) throws ConfigurationException
+ {
+ KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
+ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+ ImmutableMap.of("replication_factor", "1"),
+ true,
+ Collections.singleton(cfm));
+ Schema.instance.load(ksm);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
index 61b8b83,0922502..a2f1bcc
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@@ -28,7 -30,8 +30,9 @@@ import org.junit.AfterClass
import org.junit.BeforeClass;
import org.junit.Test;
++import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.CFMetaData;
+ import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
@@@ -47,7 -49,6 +51,8 @@@ public class CQLSSTableWriterTes
@BeforeClass
public static void setup() throws Exception
{
++ SchemaLoader.cleanupAndLeaveDirs();
+ Keyspace.setInitialized();
StorageService.instance.initServer();
}
@@@ -248,7 -255,7 +259,7 @@@
loader.stream().get();
- UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM cql_keyspace.table2;");
- UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace2.table2;");
++ UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM cql_keyspace2.table2;");
assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
}
}
[5/6] cassandra git commit: Merge branch 'cassandra-2.0' into
cassandra-2.1
Posted by yu...@apache.org.
Merge branch 'cassandra-2.0' into cassandra-2.1
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ecf48dd7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ecf48dd7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ecf48dd7
Branch: refs/heads/cassandra-2.1
Commit: ecf48dd79d29ba80008afa26fd40c3fc6d9509ac
Parents: b1c2536 fb67c41
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 5 11:48:22 2015 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:48:22 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cql3/statements/UpdateStatement.java | 29 ++++-
.../cassandra/io/sstable/CQLSSTableWriter.java | 56 +++++----
.../io/sstable/CQLSSTableWriterClientTest.java | 116 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 21 +++-
5 files changed, 195 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index aa2e1f9,faa14d5..ea79e22
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -42,105 -10,10 +42,107 @@@ Merged from 2.0
table with ASC ordering and paging (CASSANDRA-8767)
* AssertionError: "Memory was freed" when running cleanup (CASSANDRA-8716)
* Make it possible to set max_sstable_age to fractional days (CASSANDRA-8406)
- * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate()
- (CASSANDRA-8748)
* Fix some multi-column relations with indexes on some clustering
columns (CASSANDRA-8275)
+ * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate()
+ (CASSANDRA-8748)
+ * Throw OOM if allocating memory fails to return a valid pointer (CASSANDRA-8726)
+ * Fix SSTableSimpleUnsortedWriter ConcurrentModificationException (CASSANDRA-8619)
+ * 'nodetool info' prints exception against older node (CASSANDRA-8796)
+ * Ensure SSTableSimpleUnsortedWriter.close() terminates if
+ disk writer has crashed (CASSANDRA-8807)
++ * Fix CQLSSTableWriter throwing exception and spawning threads
++ (CASSANDRA-8808)
+
+
+2.1.3
+ * Fix HSHA/offheap_objects corruption (CASSANDRA-8719)
+ * Upgrade libthrift to 0.9.2 (CASSANDRA-8685)
+ * Don't use the shared ref in sstableloader (CASSANDRA-8704)
+ * Purge internal prepared statements if related tables or
+ keyspaces are dropped (CASSANDRA-8693)
+ * (cqlsh) Handle unicode BOM at start of files (CASSANDRA-8638)
+ * Stop compactions before exiting offline tools (CASSANDRA-8623)
+ * Update tools/stress/README.txt to match current behaviour (CASSANDRA-7933)
+ * Fix schema from Thrift conversion with empty metadata (CASSANDRA-8695)
+ * Safer Resource Management (CASSANDRA-7705)
+ * Make sure we compact highly overlapping cold sstables with
+ STCS (CASSANDRA-8635)
+ * rpc_interface and listen_interface generate NPE on startup when specified
+ interface doesn't exist (CASSANDRA-8677)
+ * Fix ArrayIndexOutOfBoundsException in nodetool cfhistograms (CASSANDRA-8514)
+ * Switch from yammer metrics for nodetool cf/proxy histograms (CASSANDRA-8662)
+ * Make sure we don't add tmplink files to the compaction
+ strategy (CASSANDRA-8580)
+ * (cqlsh) Handle maps with blob keys (CASSANDRA-8372)
+ * (cqlsh) Handle DynamicCompositeType schemas correctly (CASSANDRA-8563)
+ * Duplicate rows returned when in clause has repeated values (CASSANDRA-6706)
+ * Add tooling to detect hot partitions (CASSANDRA-7974)
+ * Fix cassandra-stress user-mode truncation of partition generation (CASSANDRA-8608)
+ * Only stream from unrepaired sstables during inc repair (CASSANDRA-8267)
+ * Don't allow starting multiple inc repairs on the same sstables (CASSANDRA-8316)
+ * Invalidate prepared BATCH statements when related tables
+ or keyspaces are dropped (CASSANDRA-8652)
+ * Fix missing results in secondary index queries on collections
+ with ALLOW FILTERING (CASSANDRA-8421)
+ * Expose EstimatedHistogram metrics for range slices (CASSANDRA-8627)
+ * (cqlsh) Escape clqshrc passwords properly (CASSANDRA-8618)
+ * Fix NPE when passing wrong argument in ALTER TABLE statement (CASSANDRA-8355)
+ * Pig: Refactor and deprecate CqlStorage (CASSANDRA-8599)
+ * Don't reuse the same cleanup strategy for all sstables (CASSANDRA-8537)
+ * Fix case-sensitivity of index name on CREATE and DROP INDEX
+ statements (CASSANDRA-8365)
+ * Better detection/logging for corruption in compressed sstables (CASSANDRA-8192)
+ * Use the correct repairedAt value when closing writer (CASSANDRA-8570)
+ * (cqlsh) Handle a schema mismatch being detected on startup (CASSANDRA-8512)
+ * Properly calculate expected write size during compaction (CASSANDRA-8532)
+ * Invalidate affected prepared statements when a table's columns
+ are altered (CASSANDRA-7910)
+ * Stress - user defined writes should populate sequentally (CASSANDRA-8524)
+ * Fix regression in SSTableRewriter causing some rows to become unreadable
+ during compaction (CASSANDRA-8429)
+ * Run major compactions for repaired/unrepaired in parallel (CASSANDRA-8510)
+ * (cqlsh) Fix compression options in DESCRIBE TABLE output when compression
+ is disabled (CASSANDRA-8288)
+ * (cqlsh) Fix DESCRIBE output after keyspaces are altered (CASSANDRA-7623)
+ * Make sure we set lastCompactedKey correctly (CASSANDRA-8463)
+ * (cqlsh) Fix output of CONSISTENCY command (CASSANDRA-8507)
+ * (cqlsh) Fixed the handling of LIST statements (CASSANDRA-8370)
+ * Make sstablescrub check leveled manifest again (CASSANDRA-8432)
+ * Check first/last keys in sstable when giving out positions (CASSANDRA-8458)
+ * Disable mmap on Windows (CASSANDRA-6993)
+ * Add missing ConsistencyLevels to cassandra-stress (CASSANDRA-8253)
+ * Add auth support to cassandra-stress (CASSANDRA-7985)
+ * Fix ArrayIndexOutOfBoundsException when generating error message
+ for some CQL syntax errors (CASSANDRA-8455)
+ * Scale memtable slab allocation logarithmically (CASSANDRA-7882)
+ * cassandra-stress simultaneous inserts over same seed (CASSANDRA-7964)
+ * Reduce cassandra-stress sampling memory requirements (CASSANDRA-7926)
+ * Ensure memtable flush cannot expire commit log entries from its future (CASSANDRA-8383)
+ * Make read "defrag" async to reclaim memtables (CASSANDRA-8459)
+ * Remove tmplink files for offline compactions (CASSANDRA-8321)
+ * Reduce maxHintsInProgress (CASSANDRA-8415)
+ * BTree updates may call provided update function twice (CASSANDRA-8018)
+ * Release sstable references after anticompaction (CASSANDRA-8386)
+ * Handle abort() in SSTableRewriter properly (CASSANDRA-8320)
+ * Fix high size calculations for prepared statements (CASSANDRA-8231)
+ * Centralize shared executors (CASSANDRA-8055)
+ * Fix filtering for CONTAINS (KEY) relations on frozen collection
+ clustering columns when the query is restricted to a single
+ partition (CASSANDRA-8203)
+ * Do more aggressive entire-sstable TTL expiry checks (CASSANDRA-8243)
+ * Add more log info if readMeter is null (CASSANDRA-8238)
+ * add check of the system wall clock time at startup (CASSANDRA-8305)
+ * Support for frozen collections (CASSANDRA-7859)
+ * Fix overflow on histogram computation (CASSANDRA-8028)
+ * Have paxos reuse the timestamp generation of normal queries (CASSANDRA-7801)
+ * Fix incremental repair not remove parent session on remote (CASSANDRA-8291)
+ * Improve JBOD disk utilization (CASSANDRA-7386)
+ * Log failed host when preparing incremental repair (CASSANDRA-8228)
+ * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
+ * Fix sstableupgrade throws exception (CASSANDRA-8688)
+ * Fix hang when repairing empty keyspace (CASSANDRA-8694)
+Merged from 2.0:
* Fix IllegalArgumentException in dynamic snitch (CASSANDRA-8448)
* Add support for UPDATE ... IF EXISTS (CASSANDRA-8610)
* Fix reversal of list prepends (CASSANDRA-8733)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index c35f985,f34edaf..63f87c0
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@@ -48,9 -47,25 +48,20 @@@ public class UpdateStatement extends Mo
return true;
}
- public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, Composite prefix, UpdateParameters params)
- public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
-- throws InvalidRequestException
++ public void addUpdateForKey(ColumnFamily cf,
++ ByteBuffer key,
++ Composite prefix,
++ UpdateParameters params) throws InvalidRequestException
+ {
- addUpdateForKey(cf, key, builder, params, true);
++ addUpdateForKey(cf, key, prefix, params, true);
+ }
+
+ public void addUpdateForKey(ColumnFamily cf,
+ ByteBuffer key,
- ColumnNameBuilder builder,
++ Composite prefix,
+ UpdateParameters params,
+ boolean validateIndexedColumns) throws InvalidRequestException
{
- CFDefinition cfDef = cfm.getCfDef();
-
- if (builder.getLength() > FBUtilities.MAX_UNSIGNED_SHORT)
- throw new InvalidRequestException(String.format("The sum of all clustering columns is too long (%s > %s)",
- builder.getLength(),
- FBUtilities.MAX_UNSIGNED_SHORT));
-
// Inserting the CQL row marker (see #4361)
// We always need to insert a marker for INSERT, because of the following situation:
// CREATE TABLE t ( k int PRIMARY KEY, c text );
@@@ -95,9 -112,23 +106,23 @@@
else
{
for (Operation update : updates)
- update.execute(key, cf, builder.copy(), params);
+ update.execute(key, cf, prefix, params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
+ * Checks that the value of the indexed columns is valid.
+ *
+ * @param cf the column family
+ * @throws InvalidRequestException if one of the values is invalid
+ */
+ private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index 3e27ff9,fb4c186..b4bef7d
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@@ -34,8 -34,6 +34,7 @@@ import org.apache.cassandra.cql3.statem
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.*;
- import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.composites.Composite;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
@@@ -218,11 -216,11 +217,11 @@@ public class CQLSSTableWriter implement
try
{
- for (ByteBuffer key: keys)
+ for (ByteBuffer key : keys)
{
- if (writer.currentKey() == null || !key.equals(writer.currentKey().key))
+ if (writer.currentKey() == null || !key.equals(writer.currentKey().getKey()))
writer.newRow(key);
- insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params);
+ insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params, false);
}
return this;
}
@@@ -360,19 -350,11 +351,11 @@@
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);
- createKeyspaceWithColumnFamily(this.schema);
++ createKeyspaceWithTable(this.schema);
}
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);
- addColumnFamilyToKeyspace(ksm, this.schema);
++ addTableToKeyspace(ksm, this.schema);
}
return this;
}
@@@ -384,6 -366,35 +367,35 @@@
}
/**
+ * Adds the specified column family to the specified keyspace.
+ *
+ * @param ksm the keyspace meta data
+ * @param cfm the column family meta data
+ */
- private static void addColumnFamilyToKeyspace(KSMetaData ksm, CFMetaData cfm)
++ private static void addTableToKeyspace(KSMetaData ksm, CFMetaData cfm)
+ {
+ ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
+ Schema.instance.load(cfm);
+ Schema.instance.setKeyspaceDefinition(ksm);
+ }
+
+ /**
+ * Creates a keyspace for the specified column family.
+ *
+ * @param cfm the column family
+ * @throws ConfigurationException if a problem occurs while creating the keyspace.
+ */
- private static void createKeyspaceWithColumnFamily(CFMetaData cfm) throws ConfigurationException
++ private static void createKeyspaceWithTable(CFMetaData cfm) throws ConfigurationException
+ {
+ KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
+ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+ ImmutableMap.of("replication_factor", "1"),
+ true,
+ Collections.singleton(cfm));
+ Schema.instance.load(ksm);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecf48dd7/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
index 61b8b83,0922502..a2f1bcc
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@@ -28,7 -30,8 +30,9 @@@ import org.junit.AfterClass
import org.junit.BeforeClass;
import org.junit.Test;
++import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.CFMetaData;
+ import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
@@@ -47,7 -49,6 +51,8 @@@ public class CQLSSTableWriterTes
@BeforeClass
public static void setup() throws Exception
{
++ SchemaLoader.cleanupAndLeaveDirs();
+ Keyspace.setInitialized();
StorageService.instance.initServer();
}
@@@ -248,7 -255,7 +259,7 @@@
loader.stream().get();
- UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM cql_keyspace.table2;");
- UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace2.table2;");
++ UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM cql_keyspace2.table2;");
assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
}
}
[3/6] cassandra git commit: Fix CQLSSTableWriter throwing exception
and spawning threads
Posted by yu...@apache.org.
Fix CQLSSTableWriter throwing exception and spawning threads
patch by Benjamin Lerer; reviewed by yukim for CASSANDRA-8808
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/fb67c41a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fb67c41a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fb67c41a
Branch: refs/heads/trunk
Commit: fb67c41ad7faeff7a5f33e9e0bca6493a3febe89
Parents: e56d9ef
Author: blerer <be...@datastax.com>
Authored: Tue Mar 3 12:00:57 2015 +0100
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 5 11:41:50 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cql3/statements/UpdateStatement.java | 23 ++++
.../cassandra/io/sstable/CQLSSTableWriter.java | 50 ++++++--
.../io/sstable/CQLSSTableWriterClientTest.java | 116 +++++++++++++++++++
.../io/sstable/CQLSSTableWriterTest.java | 31 +++--
5 files changed, 198 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4e34c9e..faa14d5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -27,6 +27,8 @@
* 'nodetool info' prints exception against older node (CASSANDRA-8796)
* Ensure SSTableSimpleUnsortedWriter.close() terminates if
disk writer has crashed (CASSANDRA-8807)
+ * Fix CQLSSTableWriter throwing exception and spawning threads
+ (CASSANDRA-8808)
2.0.12:
* Use more efficient slice size for querying internal secondary
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 594b5db..f34edaf 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -50,6 +50,15 @@ public class UpdateStatement extends ModificationStatement
public void addUpdateForKey(ColumnFamily cf, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params)
throws InvalidRequestException
{
+ addUpdateForKey(cf, key, builder, params, true);
+ }
+
+ public void addUpdateForKey(ColumnFamily cf,
+ ByteBuffer key,
+ ColumnNameBuilder builder,
+ UpdateParameters params,
+ boolean validateIndexedColumns) throws InvalidRequestException
+ {
CFDefinition cfDef = cfm.getCfDef();
if (builder.getLength() > FBUtilities.MAX_UNSIGNED_SHORT)
@@ -106,6 +115,20 @@ public class UpdateStatement extends ModificationStatement
update.execute(key, cf, builder.copy(), params);
}
+ // validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
+ //(e.g. when using CQLSSTableWriter)
+ if (validateIndexedColumns)
+ validateIndexedColumns(cf);
+ }
+
+ /**
+ * Checks that the value of the indexed columns is valid.
+ *
+ * @param cf the column family
+ * @throws InvalidRequestException if one of the values is invalid
+ */
+ private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
+ {
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 8006112..fb4c186 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -37,9 +37,9 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.RequestValidationException;
-import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.Allocator;
@@ -77,6 +77,11 @@ import org.apache.cassandra.utils.Pair;
*/
public class CQLSSTableWriter implements Closeable
{
+ static
+ {
+ Config.setClientMode(true);
+ }
+
private final AbstractSSTableSimpleWriter writer;
private final UpdateStatement insert;
private final List<ColumnSpecification> boundNames;
@@ -215,7 +220,7 @@ public class CQLSSTableWriter implements Closeable
{
if (writer.currentKey() == null || !key.equals(writer.currentKey().key))
writer.newRow(key);
- insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params);
+ insert.addUpdateForKey(writer.currentColumnFamily(), key, clusteringPrefix, params, false);
}
return this;
}
@@ -345,19 +350,11 @@ public class CQLSSTableWriter implements Closeable
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);
+ createKeyspaceWithColumnFamily(this.schema);
}
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);
+ addColumnFamilyToKeyspace(ksm, this.schema);
}
return this;
}
@@ -369,6 +366,35 @@ public class CQLSSTableWriter implements Closeable
}
/**
+ * Adds the specified column family to the specified keyspace.
+ *
+ * @param ksm the keyspace meta data
+ * @param cfm the column family meta data
+ */
+ private static void addColumnFamilyToKeyspace(KSMetaData ksm, CFMetaData cfm)
+ {
+ ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
+ Schema.instance.load(cfm);
+ Schema.instance.setKeyspaceDefinition(ksm);
+ }
+
+ /**
+ * Creates a keyspace for the specified column family.
+ *
+ * @param cfm the column family
+ * @throws ConfigurationException if a problem occurs while creating the keyspace.
+ */
+ private static void createKeyspaceWithColumnFamily(CFMetaData cfm) throws ConfigurationException
+ {
+ KSMetaData ksm = KSMetaData.newKeyspace(cfm.ksName,
+ AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+ ImmutableMap.of("replication_factor", "1"),
+ true,
+ Collections.singleton(cfm));
+ Schema.instance.load(ksm);
+ }
+
+ /**
* The partitioner to use.
* <p>
* By default, {@code Murmur3Partitioner} will be used. If this is not the partitioner used
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
new file mode 100644
index 0000000..d10c9fb
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterClientTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.util.Arrays;
+
+import com.google.common.io.Files;
+
+import org.junit.*;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.util.FileUtils;
+
+import static org.junit.Assert.assertEquals;
+
+import static org.junit.Assert.assertTrue;
+
+public class CQLSSTableWriterClientTest
+{
+ private File testDirectory;
+
+ @Before
+ public void setUp()
+ {
+ this.testDirectory = Files.createTempDir();
+ }
+
+ @After
+ public void tearDown()
+ {
+ FileUtils.deleteRecursive(this.testDirectory);
+ }
+
+ @AfterClass
+ public static void cleanup() throws Exception
+ {
+ Config.setClientMode(false);
+ }
+
+ @Test
+ public void testWriterInClientMode() throws IOException, InvalidRequestException
+ {
+ final String TABLE1 = "table1";
+ final String TABLE2 = "table2";
+
+ String schema = "CREATE TABLE client_test.%s ("
+ + " k int PRIMARY KEY,"
+ + " v1 text,"
+ + " v2 int"
+ + ")";
+ String insert = "INSERT INTO client_test.%s (k, v1, v2) VALUES (?, ?, ?)";
+
+ CQLSSTableWriter writer = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE1))
+ .using(String.format(insert, TABLE1)).build();
+
+ CQLSSTableWriter writer2 = CQLSSTableWriter.builder()
+ .inDirectory(this.testDirectory)
+ .forTable(String.format(schema, TABLE2))
+ .using(String.format(insert, TABLE2)).build();
+
+ writer.addRow(0, "A", 0);
+ writer2.addRow(0, "A", 0);
+ writer.addRow(1, "B", 1);
+ writer2.addRow(1, "B", 1);
+ writer.close();
+ writer2.close();
+
+ assertContainsDataFiles(this.testDirectory, "client_test-table1", "client_test-table2");
+ }
+
+ /**
+ * Checks that the specified directory contains the files with the specified prefixes.
+ *
+ * @param directory the directory containing the data files
+ * @param prefixes the file prefixes
+ */
+ private static void assertContainsDataFiles(File directory, String... prefixes)
+ {
+ FilenameFilter filter = new FilenameFilter()
+ {
+ @Override
+ public boolean accept(File dir, String name)
+ {
+ return name.endsWith("-Data.db");
+ }
+ };
+
+ File[] dataFiles = directory.listFiles(filter);
+ Arrays.sort(dataFiles);
+
+ assertEquals(dataFiles.length, prefixes.length);
+ for (int i = 0; i < dataFiles.length; i++)
+ assertTrue(dataFiles[i].toString().contains(prefixes[i]));
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fb67c41a/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 0f123a4..0922502 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -25,24 +25,25 @@ import java.util.Iterator;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Files;
+
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import org.apache.cassandra.SchemaLoader;
-import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.OutputHandler;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
public class CQLSSTableWriterTest
{
@BeforeClass
@@ -51,6 +52,12 @@ public class CQLSSTableWriterTest
StorageService.instance.initServer();
}
+ @AfterClass
+ public static void tearDown()
+ {
+ Config.setClientMode(false);
+ }
+
@Test
public void testUnsortedWriter() throws Exception
{
@@ -176,12 +183,12 @@ public class CQLSSTableWriterTest
@Override
public void run()
{
- String schema = "CREATE TABLE cql_keyspace.table2 ("
+ String schema = "CREATE TABLE cql_keyspace2.table2 ("
+ " k int,"
+ " v int,"
+ " PRIMARY KEY (k, v)"
+ ")";
- String insert = "INSERT INTO cql_keyspace.table2 (k, v) VALUES (?, ?)";
+ String insert = "INSERT INTO cql_keyspace2.table2 (k, v) VALUES (?, ?)";
CQLSSTableWriter writer = CQLSSTableWriter.builder()
.inDirectory(dataDir)
.forTable(schema)
@@ -206,7 +213,7 @@ public class CQLSSTableWriterTest
@Test
public void testConcurrentWriters() throws Exception
{
- String KS = "cql_keyspace";
+ String KS = "cql_keyspace2";
String TABLE = "table2";
File tempdir = Files.createTempDir();
@@ -235,7 +242,7 @@ public class CQLSSTableWriterTest
{
public void init(String keyspace)
{
- for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
+ for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace2"))
addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
setPartitioner(StorageService.getPartitioner());
}
@@ -248,7 +255,7 @@ public class CQLSSTableWriterTest
loader.stream().get();
- UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace.table2;");
+ UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace2.table2;");
assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
}
}