From cd1b798fa852b9a038ae234c5639c643a3c98d0a Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 31 Mar 2011 22:37:04 +0000 Subject: [PATCH 01/15] add file path to seek-past-eof exception patch by jbellis git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1087469 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/cassandra/io/util/BufferedRandomAccessFile.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java index ec4cc93e32..a36fd1ace7 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java @@ -338,7 +338,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa throw new IllegalArgumentException("new position should not be negative"); if (isReadOnly() && newPosition > fileLength) - throw new EOFException("unable to seek past the end of the file in read-only mode."); + throw new EOFException("unable to seek past the end of " + filePath + " in read-only mode."); current = newPosition; From f02b2ed0c16c74c91ccbad4135decc90340ef2c7 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 31 Mar 2011 22:40:52 +0000 Subject: [PATCH 02/15] add seek position, fileLength too git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1087470 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/cassandra/io/util/BufferedRandomAccessFile.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java index a36fd1ace7..7317d1d1af 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java @@ -338,7 +338,8 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa throw new IllegalArgumentException("new position should not be negative"); if (isReadOnly() && newPosition > fileLength) - throw new EOFException("unable to seek past the end of " + filePath + " in read-only mode."); + throw new EOFException(String.format("unable to seek to position %d in %s (%d bytes) in read-only mode", + newPosition, filePath, fileLength)); current = newPosition; From 2953e788bc3a1df918c7ced590fecd9be779ca90 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 1 Apr 2011 14:52:42 +0000 Subject: [PATCH 03/15] upgrade to whirr 0.4 patch by Pavel Yaskevich for CASSANDRA-2265 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1087728 13f79535-47bb-0310-9956-ffa450edef68 --- build.xml | 6 +- test/distributed/README.txt | 5 +- .../cassandra/CassandraServiceController.java | 46 ++++++--------- .../org/apache/cassandra/utils/BlobUtils.java | 59 ++++++++++++++++--- test/resources/whirr-default.properties | 3 +- 5 files changed, 76 insertions(+), 43 deletions(-) diff --git a/build.xml b/build.xml index 466c418088..f3c45ea45f 100644 --- a/build.xml +++ b/build.xml @@ -207,8 +207,10 @@ sourcesFilesetId="test-dependency-sources" cacheDependencyRefs="true" dependencyRefsBuildFile="${build.dir}/test-dependencies.xml"> - - + + + + diff --git a/test/distributed/README.txt b/test/distributed/README.txt index 44e1ab9107..651038b77c 100644 --- a/test/distributed/README.txt +++ b/test/distributed/README.txt @@ -26,7 +26,8 @@ configuration is located at: An example EC2/S3 whirr configuration would be: ############################################### -whirr.provider=ec2 +whirr.cluster-user=[username] +whirr.provider=aws-ec2 whirr.location-id=us-west-1 whirr.image-id=us-west-1/ami-16f3a253 whirr.hardware-id=m1.large @@ -35,7 +36,7 @@ whirr.credential=[EC2 Secret Access Key] whirr.private-key-file=${sys:user.home}/.ssh/id_rsa whirr.public-key-file=${sys:user.home}/.ssh/id_rsa.pub whirr.run-url-base=http://hoodidge.net/scripts/ -whirr.blobstore.provider=s3 +whirr.blobstore.provider=aws-s3 whirr.blobstore.container=cassandratests ############################################### diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java index f597b3ee3a..b4f34854d7 100644 --- a/test/distributed/org/apache/cassandra/CassandraServiceController.java +++ b/test/distributed/org/apache/cassandra/CassandraServiceController.java @@ -18,52 +18,41 @@ package org.apache.cassandra; -import java.io.IOException; import java.net.InetAddress; import java.net.URI; import java.util.*; -import org.apache.cassandra.thrift.Cassandra; -import org.apache.cassandra.thrift.TokenRange; -import org.apache.cassandra.utils.KeyPair; -import org.apache.cassandra.utils.BlobUtils; -import org.apache.cassandra.utils.Pair; +import com.google.common.base.Predicate; -import org.apache.commons.configuration.Configuration; +import org.apache.cassandra.thrift.Cassandra; +import org.apache.cassandra.utils.BlobUtils; +import org.apache.cassandra.utils.KeyPair; +import org.apache.cassandra.utils.Pair; import org.apache.commons.configuration.CompositeConfiguration; import org.apache.commons.configuration.PropertiesConfiguration; - import org.apache.thrift.TException; -import org.apache.thrift.protocol.*; -import org.apache.thrift.transport.*; - -import org.apache.whirr.service.Cluster; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TFramedTransport; +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; +import org.apache.whirr.service.*; import org.apache.whirr.service.Cluster.Instance; -import org.apache.whirr.service.ClusterSpec; -import org.apache.whirr.service.ComputeServiceContextBuilder; -import org.apache.whirr.service.Service; -import org.apache.whirr.service.ServiceFactory; -import org.apache.whirr.service.cassandra.CassandraService; import org.apache.whirr.service.cassandra.CassandraClusterActionHandler; import org.apache.whirr.service.jclouds.RunUrlStatement; import org.jclouds.blobstore.domain.BlobMetadata; - import org.jclouds.compute.ComputeService; +import org.jclouds.compute.domain.ExecResponse; import org.jclouds.compute.domain.NodeMetadata; import org.jclouds.compute.options.RunScriptOptions; import org.jclouds.domain.Credentials; -import org.jclouds.io.Payload; import org.jclouds.scriptbuilder.domain.OsFamily; -import org.jclouds.ssh.ExecResponse; -import static org.jclouds.io.Payloads.newStringPayload; - -import com.google.common.base.Predicate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.junit.Assert.assertThat; +import static org.jclouds.io.Payloads.newStringPayload; public class CassandraServiceController { @@ -84,7 +73,7 @@ public class CassandraServiceController private boolean running; private ClusterSpec clusterSpec; - private CassandraService service; + private Service service; private Cluster cluster; private ComputeService computeService; private Credentials credentials; @@ -96,8 +85,7 @@ public class CassandraServiceController { } - public Cassandra.Client createClient(InetAddress addr) - throws TTransportException, TException + public Cassandra.Client createClient(InetAddress addr) throws TException { TTransport transport = new TSocket( addr.getHostAddress(), @@ -170,10 +158,10 @@ public class CassandraServiceController tarball = blob.left; config.setProperty(CassandraClusterActionHandler.BIN_TARBALL, blob.right.toURL().toString()); // TODO: parse the CassandraVersion property file instead - config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.7"); + config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.8"); } - service = (CassandraService)new ServiceFactory().create(clusterSpec.getServiceName()); + service = new ServiceFactory().create(clusterSpec.getServiceName()); cluster = service.launchCluster(clusterSpec); computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService(); hosts = new ArrayList(); diff --git a/test/distributed/org/apache/cassandra/utils/BlobUtils.java b/test/distributed/org/apache/cassandra/utils/BlobUtils.java index 94474cf086..ddcbfa5fc1 100644 --- a/test/distributed/org/apache/cassandra/utils/BlobUtils.java +++ b/test/distributed/org/apache/cassandra/utils/BlobUtils.java @@ -19,24 +19,26 @@ package org.apache.cassandra.utils; import java.io.File; +import java.io.FileWriter; +import java.io.IOException; import java.net.URI; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import com.google.common.io.Files; import org.apache.commons.configuration.Configuration; import org.apache.whirr.service.ClusterSpec; -import org.jclouds.blobstore.BlobStore; import org.jclouds.blobstore.BlobStoreContext; import org.jclouds.blobstore.BlobStoreContextFactory; -import org.jclouds.blobstore.domain.BlobMetadata; import org.jclouds.blobstore.InputStreamMap; +import org.jclouds.blobstore.domain.BlobMetadata; -import org.jclouds.aws.s3.S3Client; -import org.jclouds.aws.s3.S3AsyncClient; -import org.jclouds.aws.s3.domain.AccessControlList; -import org.jclouds.aws.s3.domain.CannedAccessPolicy; - -import org.jclouds.rest.RestContext; +import org.jclouds.s3.S3AsyncClient; +import org.jclouds.s3.S3Client; +import org.jclouds.s3.domain.AccessControlList; +import org.jclouds.s3.domain.CannedAccessPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,24 +79,63 @@ public final class BlobUtils File file = new File(filename); String container = getContainer(config); String provider = getProvider(config); + + // blob name and checksum of the file String blobName = System.nanoTime() + "/" + file.getName(); + String blobNameChecksum = blobName + ".md5"; + BlobStoreContext context = getContext(config, spec); + + File checksumFile; + + try + { + checksumFile = File.createTempFile("dtchecksum", "md5"); + checksumFile.deleteOnExit(); + + FileWriter checksumWriter = new FileWriter(checksumFile); + + String checksum = FBUtilities.bytesToHex(Files.getDigest(file, MessageDigest.getInstance("MD5"))); + + checksumWriter.write(String.format("%s %s", checksum, file.getName())); + checksumWriter.close(); + } + catch (IOException e) + { + throw new RuntimeException("Can't create a checksum of the file: " + filename); + } + catch (NoSuchAlgorithmException e) + { + throw new RuntimeException(e.getMessage()); + } + try { InputStreamMap map = context.createInputStreamMap(container); + map.putFile(blobName, file); + map.putFile(blobNameChecksum, checksumFile); + // TODO: magic! in order to expose the blob as public, we need to dive into provider specific APIs // the hope is that permissions are encapsulated in jclouds in the future - if (provider.equals("s3")) + if (provider.contains("s3")) { S3Client sss = context.getProviderSpecificContext().getApi(); String ownerId = sss.getObjectACL(container, blobName).getOwner().getId(); + sss.putObjectACL(container, blobName, AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); + + sss.putObjectACL(container, + blobNameChecksum, + AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); } else + { LOG.warn(provider + " may not be properly supported for tarball transfer."); + } + // resolve the full URI of the blob (see http://code.google.com/p/jclouds/issues/detail?id=431) BlobMetadata blob = context.getBlobStore().blobMetadata(container, blobName); URI uri = context.getProviderSpecificContext().getEndpoint().resolve("/" + container + "/" + blob.getName()); diff --git a/test/resources/whirr-default.properties b/test/resources/whirr-default.properties index fc33c48122..5f2066080e 100644 --- a/test/resources/whirr-default.properties +++ b/test/resources/whirr-default.properties @@ -17,5 +17,6 @@ # whirr.service-name=cassandra whirr.cluster-name=cassandra_test +whirr.cluster-user=app whirr.instance-templates=4 cassandra -whirr.version=0.3.0-incubating-SNAPSHOT +whirr.version=0.4.0-incubating From 8c925375a32d833ea846449287a8956352e12870 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 1 Apr 2011 14:57:44 +0000 Subject: [PATCH 04/15] revert whirr change, it's for 0.8 only git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1087732 13f79535-47bb-0310-9956-ffa450edef68 --- build.xml | 6 +- test/distributed/README.txt | 5 +- .../cassandra/CassandraServiceController.java | 42 ++++++++----- .../org/apache/cassandra/utils/BlobUtils.java | 59 +++---------------- test/resources/whirr-default.properties | 3 +- 5 files changed, 41 insertions(+), 74 deletions(-) diff --git a/build.xml b/build.xml index f3c45ea45f..466c418088 100644 --- a/build.xml +++ b/build.xml @@ -207,10 +207,8 @@ sourcesFilesetId="test-dependency-sources" cacheDependencyRefs="true" dependencyRefsBuildFile="${build.dir}/test-dependencies.xml"> - - - - + + diff --git a/test/distributed/README.txt b/test/distributed/README.txt index 651038b77c..44e1ab9107 100644 --- a/test/distributed/README.txt +++ b/test/distributed/README.txt @@ -26,8 +26,7 @@ configuration is located at: An example EC2/S3 whirr configuration would be: ############################################### -whirr.cluster-user=[username] -whirr.provider=aws-ec2 +whirr.provider=ec2 whirr.location-id=us-west-1 whirr.image-id=us-west-1/ami-16f3a253 whirr.hardware-id=m1.large @@ -36,7 +35,7 @@ whirr.credential=[EC2 Secret Access Key] whirr.private-key-file=${sys:user.home}/.ssh/id_rsa whirr.public-key-file=${sys:user.home}/.ssh/id_rsa.pub whirr.run-url-base=http://hoodidge.net/scripts/ -whirr.blobstore.provider=aws-s3 +whirr.blobstore.provider=s3 whirr.blobstore.container=cassandratests ############################################### diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java index b4f34854d7..f597b3ee3a 100644 --- a/test/distributed/org/apache/cassandra/CassandraServiceController.java +++ b/test/distributed/org/apache/cassandra/CassandraServiceController.java @@ -18,41 +18,52 @@ package org.apache.cassandra; +import java.io.IOException; import java.net.InetAddress; import java.net.URI; import java.util.*; -import com.google.common.base.Predicate; - import org.apache.cassandra.thrift.Cassandra; -import org.apache.cassandra.utils.BlobUtils; +import org.apache.cassandra.thrift.TokenRange; import org.apache.cassandra.utils.KeyPair; +import org.apache.cassandra.utils.BlobUtils; import org.apache.cassandra.utils.Pair; + +import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.CompositeConfiguration; import org.apache.commons.configuration.PropertiesConfiguration; + import org.apache.thrift.TException; -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.protocol.TProtocol; -import org.apache.thrift.transport.TFramedTransport; -import org.apache.thrift.transport.TSocket; -import org.apache.thrift.transport.TTransport; -import org.apache.whirr.service.*; +import org.apache.thrift.protocol.*; +import org.apache.thrift.transport.*; + +import org.apache.whirr.service.Cluster; import org.apache.whirr.service.Cluster.Instance; +import org.apache.whirr.service.ClusterSpec; +import org.apache.whirr.service.ComputeServiceContextBuilder; +import org.apache.whirr.service.Service; +import org.apache.whirr.service.ServiceFactory; +import org.apache.whirr.service.cassandra.CassandraService; import org.apache.whirr.service.cassandra.CassandraClusterActionHandler; import org.apache.whirr.service.jclouds.RunUrlStatement; import org.jclouds.blobstore.domain.BlobMetadata; + import org.jclouds.compute.ComputeService; -import org.jclouds.compute.domain.ExecResponse; import org.jclouds.compute.domain.NodeMetadata; import org.jclouds.compute.options.RunScriptOptions; import org.jclouds.domain.Credentials; +import org.jclouds.io.Payload; import org.jclouds.scriptbuilder.domain.OsFamily; +import org.jclouds.ssh.ExecResponse; +import static org.jclouds.io.Payloads.newStringPayload; + +import com.google.common.base.Predicate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.jclouds.io.Payloads.newStringPayload; +import static org.junit.Assert.assertThat; public class CassandraServiceController { @@ -73,7 +84,7 @@ public class CassandraServiceController private boolean running; private ClusterSpec clusterSpec; - private Service service; + private CassandraService service; private Cluster cluster; private ComputeService computeService; private Credentials credentials; @@ -85,7 +96,8 @@ public class CassandraServiceController { } - public Cassandra.Client createClient(InetAddress addr) throws TException + public Cassandra.Client createClient(InetAddress addr) + throws TTransportException, TException { TTransport transport = new TSocket( addr.getHostAddress(), @@ -158,10 +170,10 @@ public class CassandraServiceController tarball = blob.left; config.setProperty(CassandraClusterActionHandler.BIN_TARBALL, blob.right.toURL().toString()); // TODO: parse the CassandraVersion property file instead - config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.8"); + config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.7"); } - service = new ServiceFactory().create(clusterSpec.getServiceName()); + service = (CassandraService)new ServiceFactory().create(clusterSpec.getServiceName()); cluster = service.launchCluster(clusterSpec); computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService(); hosts = new ArrayList(); diff --git a/test/distributed/org/apache/cassandra/utils/BlobUtils.java b/test/distributed/org/apache/cassandra/utils/BlobUtils.java index ddcbfa5fc1..94474cf086 100644 --- a/test/distributed/org/apache/cassandra/utils/BlobUtils.java +++ b/test/distributed/org/apache/cassandra/utils/BlobUtils.java @@ -19,26 +19,24 @@ package org.apache.cassandra.utils; import java.io.File; -import java.io.FileWriter; -import java.io.IOException; import java.net.URI; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import com.google.common.io.Files; import org.apache.commons.configuration.Configuration; import org.apache.whirr.service.ClusterSpec; +import org.jclouds.blobstore.BlobStore; import org.jclouds.blobstore.BlobStoreContext; import org.jclouds.blobstore.BlobStoreContextFactory; -import org.jclouds.blobstore.InputStreamMap; import org.jclouds.blobstore.domain.BlobMetadata; +import org.jclouds.blobstore.InputStreamMap; -import org.jclouds.s3.S3AsyncClient; -import org.jclouds.s3.S3Client; -import org.jclouds.s3.domain.AccessControlList; -import org.jclouds.s3.domain.CannedAccessPolicy; +import org.jclouds.aws.s3.S3Client; +import org.jclouds.aws.s3.S3AsyncClient; +import org.jclouds.aws.s3.domain.AccessControlList; +import org.jclouds.aws.s3.domain.CannedAccessPolicy; + +import org.jclouds.rest.RestContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -79,63 +77,24 @@ public final class BlobUtils File file = new File(filename); String container = getContainer(config); String provider = getProvider(config); - - // blob name and checksum of the file String blobName = System.nanoTime() + "/" + file.getName(); - String blobNameChecksum = blobName + ".md5"; - BlobStoreContext context = getContext(config, spec); - - File checksumFile; - - try - { - checksumFile = File.createTempFile("dtchecksum", "md5"); - checksumFile.deleteOnExit(); - - FileWriter checksumWriter = new FileWriter(checksumFile); - - String checksum = FBUtilities.bytesToHex(Files.getDigest(file, MessageDigest.getInstance("MD5"))); - - checksumWriter.write(String.format("%s %s", checksum, file.getName())); - checksumWriter.close(); - } - catch (IOException e) - { - throw new RuntimeException("Can't create a checksum of the file: " + filename); - } - catch (NoSuchAlgorithmException e) - { - throw new RuntimeException(e.getMessage()); - } - try { InputStreamMap map = context.createInputStreamMap(container); - map.putFile(blobName, file); - map.putFile(blobNameChecksum, checksumFile); - // TODO: magic! in order to expose the blob as public, we need to dive into provider specific APIs // the hope is that permissions are encapsulated in jclouds in the future - if (provider.contains("s3")) + if (provider.equals("s3")) { S3Client sss = context.getProviderSpecificContext().getApi(); String ownerId = sss.getObjectACL(container, blobName).getOwner().getId(); - sss.putObjectACL(container, blobName, AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); - - sss.putObjectACL(container, - blobNameChecksum, - AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); } else - { LOG.warn(provider + " may not be properly supported for tarball transfer."); - } - // resolve the full URI of the blob (see http://code.google.com/p/jclouds/issues/detail?id=431) BlobMetadata blob = context.getBlobStore().blobMetadata(container, blobName); URI uri = context.getProviderSpecificContext().getEndpoint().resolve("/" + container + "/" + blob.getName()); diff --git a/test/resources/whirr-default.properties b/test/resources/whirr-default.properties index 5f2066080e..fc33c48122 100644 --- a/test/resources/whirr-default.properties +++ b/test/resources/whirr-default.properties @@ -17,6 +17,5 @@ # whirr.service-name=cassandra whirr.cluster-name=cassandra_test -whirr.cluster-user=app whirr.instance-templates=4 cassandra -whirr.version=0.4.0-incubating +whirr.version=0.3.0-incubating-SNAPSHOT From f674bcc07220ea7accc086406f1d99cfa6578a5f Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 4 Apr 2011 13:21:28 +0000 Subject: [PATCH 05/15] add IntegerType to CliUserHelp patch by Joaquin Casares; reviewed by Pavel Yaskevich for CASSANDRA-2414 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1088612 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/cli/CliUserHelp.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/cli/CliUserHelp.java b/src/java/org/apache/cassandra/cli/CliUserHelp.java index ad2c8c577f..3833e22c62 100644 --- a/src/java/org/apache/cassandra/cli/CliUserHelp.java +++ b/src/java/org/apache/cassandra/cli/CliUserHelp.java @@ -55,7 +55,7 @@ public class CliUserHelp { {{ put(ColumnFamilyArgument.COLUMN_TYPE, "Super or Standard"); put(ColumnFamilyArgument.COMMENT, "Human-readable column family description. Any string is acceptable"); - put(ColumnFamilyArgument.COMPARATOR, "The class used as a comparator when sorting column names.\n Valid options include: AsciiType, BytesType, LexicalUUIDType,\n LongType, TimeUUIDType, and UTF8Type"); + put(ColumnFamilyArgument.COMPARATOR, "The class used as a comparator when sorting column names.\n Valid options include: AsciiType, BytesType, LexicalUUIDType,\n LongType, IntegerType, TimeUUIDType, and UTF8Type"); put(ColumnFamilyArgument.SUBCOMPARATOR, "Comparator for sorting subcolumn names, for Super columns only"); put(ColumnFamilyArgument.MEMTABLE_OPERATIONS, "Flush memtables after this many operations (in millions)"); put(ColumnFamilyArgument.MEMTABLE_THROUGHPUT, "... or after this many MB have been written"); From 5e8361bbcb142e06b46bd72263b0826acc5b390f Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Mon, 4 Apr 2011 21:53:09 +0000 Subject: [PATCH 06/15] Pig uses schema information to cast to/from native types. Patch by Jeremy Hanna and brandonwilliams, reviewed by brandonwilliams for CASSANDRA-2387 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1088800 13f79535-47bb-0310-9956-ffa450edef68 --- .../hadoop/pig/CassandraStorage.java | 206 ++++++++++++++++-- .../cassandra/db/marshal/AbstractType.java | 6 +- .../cassandra/db/marshal/AsciiType.java | 24 +- .../cassandra/db/marshal/BytesType.java | 26 ++- .../cassandra/db/marshal/IntegerType.java | 21 +- .../cassandra/db/marshal/LexicalUUIDType.java | 19 +- .../db/marshal/LocalByPartionerType.java | 12 +- .../apache/cassandra/db/marshal/LongType.java | 19 +- .../cassandra/db/marshal/TimeUUIDType.java | 19 +- .../apache/cassandra/db/marshal/UTF8Type.java | 27 ++- .../cassandra/utils/ByteBufferUtil.java | 26 +++ .../apache/cassandra/utils/FBUtilities.java | 25 +++ 12 files changed, 393 insertions(+), 37 deletions(-) diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index cb1cc7f91a..4eea86446a 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -20,6 +20,9 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.FBUtilities; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -27,9 +30,8 @@ import org.apache.commons.logging.LogFactory; import org.apache.cassandra.db.Column; import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.SuperColumn; +import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.hadoop.*; -import org.apache.cassandra.thrift.SlicePredicate; -import org.apache.cassandra.thrift.SliceRange; import org.apache.cassandra.avro.Mutation; import org.apache.cassandra.avro.Deletion; import org.apache.cassandra.avro.ColumnOrSuperColumn; @@ -44,6 +46,14 @@ import org.apache.pig.backend.executionengine.ExecException; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit; import org.apache.pig.data.*; import org.apache.pig.impl.logicalLayer.FrontendException; +import org.apache.pig.impl.util.UDFContext; +import org.apache.thrift.TDeserializer; +import org.apache.thrift.TException; +import org.apache.thrift.TSerializer; +import org.apache.thrift.transport.TFramedTransport; +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; /** * A LoadFunc wrapping ColumnFamilyInputFormat. @@ -58,6 +68,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo public final static String PIG_INITIAL_ADDRESS = "PIG_INITIAL_ADDRESS"; public final static String PIG_PARTITIONER = "PIG_PARTITIONER"; + private static String UDFCONTEXT_SCHEMA_KEY = "schema"; + private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER; private static final Log logger = LogFactory.getLog(CassandraStorage.class); @@ -72,8 +84,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo private RecordWriter writer; private int limit; - public CassandraStorage() - { + public CassandraStorage() + { this(1024); } @@ -100,19 +112,20 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo if (!reader.nextKeyValue()) return null; + CfDef cfDef = getCfDef(); ByteBuffer key = (ByteBuffer)reader.getCurrentKey(); SortedMap cf = (SortedMap)reader.getCurrentValue(); assert key != null && cf != null; // and wrap it in a tuple - Tuple tuple = TupleFactory.getInstance().newTuple(2); + Tuple tuple = TupleFactory.getInstance().newTuple(2); ArrayList columns = new ArrayList(); tuple.set(0, new DataByteArray(key.array(), key.position()+key.arrayOffset(), key.limit()+key.arrayOffset())); for (Map.Entry entry : cf.entrySet()) - { - columns.add(columnToTuple(entry.getKey(), entry.getValue())); + { + columns.add(columnToTuple(entry.getKey(), entry.getValue(), cfDef)); } - + tuple.set(1, new DefaultDataBag(columns)); return tuple; } @@ -122,28 +135,85 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo } } - private Tuple columnToTuple(ByteBuffer name, IColumn col) throws IOException + private Tuple columnToTuple(ByteBuffer name, IColumn col, CfDef cfDef) throws IOException { Tuple pair = TupleFactory.getInstance().newTuple(2); - pair.set(0, new DataByteArray(name.array(), name.position()+name.arrayOffset(), name.limit()+name.arrayOffset())); + List marshallers = getDefaultMarshallers(cfDef); + Map validators = getValidatorMap(cfDef); + if (col instanceof Column) { // standard - pair.set(1, new DataByteArray(col.value().array(), - col.value().position()+col.value().arrayOffset(), - col.value().limit()+col.value().arrayOffset())); + pair.set(0, marshallers.get(0).compose(name)); + if (validators.get(name) == null) + // Have to special case BytesType because compose returns a ByteBuffer + if (marshallers.get(1) instanceof BytesType) + pair.set(1, new DataByteArray(ByteBufferUtil.getArray(col.value()))); + else + pair.set(1, marshallers.get(1).compose(col.value())); + else + pair.set(1, validators.get(name).compose(col.value())); return pair; } // super ArrayList subcols = new ArrayList(); for (IColumn subcol : ((SuperColumn)col).getSubColumns()) - subcols.add(columnToTuple(subcol.name(), subcol)); + subcols.add(columnToTuple(subcol.name(), subcol, cfDef)); pair.set(1, new DefaultDataBag(subcols)); return pair; } + private CfDef getCfDef() + { + UDFContext context = UDFContext.getUDFContext(); + Properties property = context.getUDFProperties(ResourceSchema.class); + return cfdefFromString(property.getProperty(UDFCONTEXT_SCHEMA_KEY)); + } + + private List getDefaultMarshallers(CfDef cfDef) throws IOException + { + ArrayList marshallers = new ArrayList(); + AbstractType comparator = null; + AbstractType default_validator = null; + try + { + comparator = FBUtilities.getInstance(cfDef.comparator_type, "comparator"); + default_validator = FBUtilities.getInstance(cfDef.default_validation_class, "validator"); + } + catch (ConfigurationException e) + { + throw new IOException(e); + } + + marshallers.add(comparator); + marshallers.add(default_validator); + return marshallers; + } + + private Map getValidatorMap(CfDef cfDef) throws IOException + { + Map validators = new HashMap(); + for (ColumnDef cd : cfDef.column_metadata) + { + if (cd.getValidation_class() != null && !cd.getValidation_class().isEmpty()) + { + AbstractType validator = null; + try + { + validator = FBUtilities.getInstance(cd.getValidation_class(), "validator"); + validators.put(cd.name, validator); + } + catch (ConfigurationException e) + { + throw new IOException(e); + } + } + } + return validators; + } + @Override public InputFormat getInputFormat() { @@ -156,7 +226,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo this.reader = reader; } - private void setLocationFromUri(String location) throws IOException + private void setLocationFromUri(String location) throws IOException { // parse uri into keyspace and columnfamily String names[]; @@ -219,6 +289,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo } ConfigHelper.setInputColumnFamily(conf, keyspace, column_family); setConnectionInformation(); + initSchema(); } @Override @@ -274,7 +345,9 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo ByteBuffer key = objToBB(t.get(0)); DefaultDataBag pairs = (DefaultDataBag) t.get(1); ArrayList mutationList = new ArrayList(); - + CfDef cfDef = getCfDef(); + List marshallers = getDefaultMarshallers(cfDef); + Map validators = getValidatorMap(cfDef); try { for (Tuple pair : pairs) @@ -306,7 +379,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo mutation.column_or_supercolumn.super_column = sc; } } - else // assume column since it could be anything else + else // assume column since it couldn't be anything else { if (pair.get(1) == null) { @@ -318,8 +391,15 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo else { org.apache.cassandra.avro.Column column = new org.apache.cassandra.avro.Column(); - column.name = objToBB(pair.get(0)); - column.value = objToBB(pair.get(1)); + column.name = marshallers.get(0).decompose((pair.get(0))); + if (validators.get(column.name) == null) + // Have to special case BytesType to convert DataByteArray into ByteBuffer + if (marshallers.get(1) instanceof BytesType) + column.value = ByteBuffer.wrap(((DataByteArray) pair.get(1)).get()); + else + column.value = marshallers.get(1).decompose(pair.get(1)); + else + column.value = validators.get(column.name).decompose(pair.get(1)); column.timestamp = System.currentTimeMillis() * 1000; mutation.column_or_supercolumn = new ColumnOrSuperColumn(); mutation.column_or_supercolumn.column = column; @@ -358,4 +438,92 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo return new RequiredFieldResponse(true); } + + /* Methods to get the column family schema from Cassandra */ + + private void initSchema() + { + Cassandra.Client client = null; + try + { + client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); + CfDef cfDef = null; + client.set_keyspace(keyspace); + KsDef ksDef = client.describe_keyspace(keyspace); + List defs = ksDef.getCf_defs(); + for (CfDef def : defs) + { + if (column_family.equalsIgnoreCase(def.getName())) + { + cfDef = def; + break; + } + } + UDFContext context = UDFContext.getUDFContext(); + Properties property = context.getUDFProperties(ResourceSchema.class); + property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef)); + } + catch (TException e) + { + throw new RuntimeException(e); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (NotFoundException e) + { + throw new RuntimeException(e); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static Cassandra.Client createConnection(String host, Integer port, boolean framed) throws IOException + { + TSocket socket = new TSocket(host, port); + TTransport trans = framed ? new TFramedTransport(socket) : socket; + try + { + trans.open(); + } + catch (TTransportException e) + { + throw new IOException("unable to connect to server", e); + } + return new Cassandra.Client(new TBinaryProtocol(trans)); + } + + private static String cfdefToString(CfDef cfDef) + { + assert cfDef != null; + // this is so awful it's kind of cool! + TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory()); + try + { + return FBUtilities.bytesToHex(serializer.serialize(cfDef)); + } + catch (TException e) + { + throw new RuntimeException(e); + } + } + + private static CfDef cfdefFromString(String st) + { + assert st != null; + TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory()); + CfDef cfDef = new CfDef(); + try + { + deserializer.deserialize(cfDef, FBUtilities.hexToBytes(st)); + } + catch (TException e) + { + throw new RuntimeException(e); + } + return cfDef; + } } diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java index 4fc244e19a..7a38e8e006 100644 --- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java +++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java @@ -36,7 +36,7 @@ import static org.apache.cassandra.io.sstable.IndexHelper.IndexInfo; * should always handle those values even if they normally do not * represent a valid ByteBuffer for the type being compared. */ -public abstract class AbstractType implements Comparator +public abstract class AbstractType implements Comparator { public final Comparator indexComparator; public final Comparator indexReverseComparator; @@ -108,6 +108,10 @@ public abstract class AbstractType implements Comparator throw new UnsupportedOperationException(); } + public abstract T compose(ByteBuffer bytes); + + public abstract ByteBuffer decompose(T value); + /* validate that the byte array is a valid sequence for the type we are supposed to be comparing */ public abstract void validate(ByteBuffer bytes) throws MarshalException; diff --git a/src/java/org/apache/cassandra/db/marshal/AsciiType.java b/src/java/org/apache/cassandra/db/marshal/AsciiType.java index d797429620..1d0f24d2fe 100644 --- a/src/java/org/apache/cassandra/db/marshal/AsciiType.java +++ b/src/java/org/apache/cassandra/db/marshal/AsciiType.java @@ -28,12 +28,32 @@ import com.google.common.base.Charsets; import org.apache.cassandra.utils.ByteBufferUtil; -public class AsciiType extends BytesType +public class AsciiType extends AbstractType { public static final AsciiType instance = new AsciiType(); + public static AsciiType getInstance() + { + return instance; + } + AsciiType() {} // singleton + public String compose(ByteBuffer bytes) + { + return getString(bytes); + } + + public ByteBuffer decompose(String value) + { + return ByteBufferUtil.bytes(value, Charsets.US_ASCII); + } + + public int compare(ByteBuffer o1, ByteBuffer o2) + { + return BytesType.bytesCompare(o1, o2); + } + @Override public String getString(ByteBuffer bytes) { @@ -49,7 +69,7 @@ public class AsciiType extends BytesType public ByteBuffer fromString(String source) { - return ByteBufferUtil.bytes(source, Charsets.US_ASCII); + return decompose(source); } public void validate(ByteBuffer bytes) throws MarshalException diff --git a/src/java/org/apache/cassandra/db/marshal/BytesType.java b/src/java/org/apache/cassandra/db/marshal/BytesType.java index 94ee7bdd60..d6f9bfafdd 100644 --- a/src/java/org/apache/cassandra/db/marshal/BytesType.java +++ b/src/java/org/apache/cassandra/db/marshal/BytesType.java @@ -25,19 +25,39 @@ import java.nio.ByteBuffer; import org.apache.cassandra.utils.ByteBufferUtil; -public class BytesType extends AbstractType +public class BytesType extends AbstractType { public static final BytesType instance = new BytesType(); + public static BytesType getInstance() + { + return instance; + } + BytesType() {} // singleton - + + public ByteBuffer compose(ByteBuffer bytes) + { + return bytes.duplicate(); + } + + public ByteBuffer decompose(ByteBuffer value) + { + return value; + } + public int compare(ByteBuffer o1, ByteBuffer o2) + { + return BytesType.bytesCompare(o1, o2); + } + + public static int bytesCompare(ByteBuffer o1, ByteBuffer o2) { if(null == o1){ if(null == o2) return 0; else return -1; } - + return ByteBufferUtil.compareUnsigned(o1, o2); } diff --git a/src/java/org/apache/cassandra/db/marshal/IntegerType.java b/src/java/org/apache/cassandra/db/marshal/IntegerType.java index fda5be8f05..4c14ab66dc 100644 --- a/src/java/org/apache/cassandra/db/marshal/IntegerType.java +++ b/src/java/org/apache/cassandra/db/marshal/IntegerType.java @@ -22,12 +22,19 @@ package org.apache.cassandra.db.marshal; import java.math.BigInteger; import java.nio.ByteBuffer; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.thrift.TBaseHelper; -public final class IntegerType extends AbstractType +public final class IntegerType extends AbstractType { public static final IntegerType instance = new IntegerType(); + public static IntegerType getInstance() + { + return instance; + } + + private static int findMostSignificantByte(ByteBuffer bytes) { int len = bytes.remaining() - 1; @@ -56,6 +63,16 @@ public final class IntegerType extends AbstractType IntegerType() {/* singleton */} + public BigInteger compose(ByteBuffer bytes) + { + return new BigInteger(ByteBufferUtil.getArray(bytes)); + } + + public ByteBuffer decompose(BigInteger value) + { + return ByteBuffer.wrap(value.toByteArray()); + } + public int compare(ByteBuffer lhs, ByteBuffer rhs) { int lhsLen = lhs.remaining(); @@ -138,7 +155,7 @@ public final class IntegerType extends AbstractType throw new RuntimeException("'" + source + "' could not be translated into an IntegerType."); } - return ByteBuffer.wrap(integerType.toByteArray()); + return decompose(integerType); } public void validate(ByteBuffer bytes) throws MarshalException diff --git a/src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java b/src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java index 6aff641132..d4e27e3da1 100644 --- a/src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java +++ b/src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java @@ -26,12 +26,27 @@ import java.util.UUID; import org.apache.cassandra.utils.UUIDGen; -public class LexicalUUIDType extends AbstractType +public class LexicalUUIDType extends AbstractType { public static final LexicalUUIDType instance = new LexicalUUIDType(); + public static LexicalUUIDType getInstance() + { + return instance; + } + LexicalUUIDType() {} // singleton + public UUID compose(ByteBuffer bytes) + { + return UUIDGen.getUUID(bytes); + } + + public ByteBuffer decompose(UUID value) + { + return ByteBuffer.wrap(UUIDGen.decompose(value)); + } + public int compare(ByteBuffer o1, ByteBuffer o2) { if (o1.remaining() == 0) @@ -61,7 +76,7 @@ public class LexicalUUIDType extends AbstractType public ByteBuffer fromString(String source) { - return ByteBuffer.wrap(UUIDGen.decompose(UUID.fromString(source))); + return decompose(UUID.fromString(source)); } public void validate(ByteBuffer bytes) throws MarshalException diff --git a/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java b/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java index ebc3ed3a8a..f064acb856 100644 --- a/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java +++ b/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java @@ -29,7 +29,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; /** for sorting columns representing row keys in the row ordering as determined by a partitioner. * Not intended for user-defined CFs, and will in fact error out if used with such. */ -public class LocalByPartionerType extends AbstractType +public class LocalByPartionerType extends AbstractType { private final IPartitioner partitioner; @@ -38,6 +38,16 @@ public class LocalByPartionerType extends AbstractType this.partitioner = partitioner; } + public ByteBuffer compose(ByteBuffer bytes) + { + throw new UnsupportedOperationException("You can't do this with a local partitioner."); + } + + public ByteBuffer decompose(ByteBuffer bytes) + { + throw new UnsupportedOperationException("You can't do this with a local partitioner."); + } + public String getString(ByteBuffer bytes) { return ByteBufferUtil.bytesToHex(bytes); diff --git a/src/java/org/apache/cassandra/db/marshal/LongType.java b/src/java/org/apache/cassandra/db/marshal/LongType.java index 7d715cd9f6..dc0ddb2e79 100644 --- a/src/java/org/apache/cassandra/db/marshal/LongType.java +++ b/src/java/org/apache/cassandra/db/marshal/LongType.java @@ -25,12 +25,27 @@ import java.nio.ByteBuffer; import org.apache.cassandra.utils.ByteBufferUtil; -public class LongType extends AbstractType +public class LongType extends AbstractType { public static final LongType instance = new LongType(); + public static LongType getInstance() + { + return instance; + } + LongType() {} // singleton + public Long compose(ByteBuffer bytes) + { + return ByteBufferUtil.toLong(bytes); + } + + public ByteBuffer decompose(Long value) + { + return ByteBufferUtil.bytes(value); + } + public int compare(ByteBuffer o1, ByteBuffer o2) { if (o1.remaining() == 0) @@ -78,7 +93,7 @@ public class LongType extends AbstractType throw new RuntimeException("'" + source + "' could not be translated into a LongType."); } - return ByteBufferUtil.bytes(longType); + return decompose(longType); } public void validate(ByteBuffer bytes) throws MarshalException diff --git a/src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java b/src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java index 521010e9f1..47d59b2cc0 100644 --- a/src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java +++ b/src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java @@ -26,12 +26,27 @@ import java.util.UUID; import org.apache.cassandra.utils.UUIDGen; -public class TimeUUIDType extends AbstractType +public class TimeUUIDType extends AbstractType { public static final TimeUUIDType instance = new TimeUUIDType(); + public static TimeUUIDType getInstance() + { + return instance; + } + TimeUUIDType() {} // singleton + public UUID compose(ByteBuffer bytes) + { + return UUIDGen.getUUID(bytes); + } + + public ByteBuffer decompose(UUID value) + { + return ByteBuffer.wrap(UUIDGen.decompose(value)); + } + public int compare(ByteBuffer o1, ByteBuffer o2) { if (o1.remaining() == 0) @@ -102,7 +117,7 @@ public class TimeUUIDType extends AbstractType if (uuid.version() != 1) throw new IllegalArgumentException("TimeUUID supports only version 1 UUIDs"); - return ByteBuffer.wrap(UUIDGen.decompose(uuid)); + return decompose(uuid); } public void validate(ByteBuffer bytes) throws MarshalException diff --git a/src/java/org/apache/cassandra/db/marshal/UTF8Type.java b/src/java/org/apache/cassandra/db/marshal/UTF8Type.java index 3f94486d96..87c8937828 100644 --- a/src/java/org/apache/cassandra/db/marshal/UTF8Type.java +++ b/src/java/org/apache/cassandra/db/marshal/UTF8Type.java @@ -23,14 +23,35 @@ package org.apache.cassandra.db.marshal; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; +import com.google.common.base.Charsets; import org.apache.cassandra.utils.ByteBufferUtil; -public class UTF8Type extends BytesType +public class UTF8Type extends AbstractType { public static final UTF8Type instance = new UTF8Type(); + public static UTF8Type getInstance() + { + return instance; + } + UTF8Type() {} // singleton + public String compose(ByteBuffer bytes) + { + return getString(bytes); + } + + public ByteBuffer decompose(String value) + { + return ByteBufferUtil.bytes(value, Charsets.UTF_8); + } + + public int compare(ByteBuffer o1, ByteBuffer o2) + { + return BytesType.bytesCompare(o1, o2); + } + public String getString(ByteBuffer bytes) { try @@ -45,9 +66,9 @@ public class UTF8Type extends BytesType public ByteBuffer fromString(String source) { - return ByteBufferUtil.bytes(source); + return decompose(source); } - + public void validate(ByteBuffer bytes) throws MarshalException { if (!UTF8Validator.validate(bytes.slice())) diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 2c0ce47185..1d841a6369 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -377,6 +377,21 @@ public class ByteBufferUtil return bytes.getInt(bytes.position()); } + public static long toLong(ByteBuffer bytes) + { + return bytes.getLong(bytes.position()); + } + + public static float toFloat(ByteBuffer bytes) + { + return bytes.getFloat(bytes.position()); + } + + public static double toDouble(ByteBuffer bytes) + { + return bytes.getDouble(bytes.position()); + } + public static ByteBuffer bytes(int i) { return ByteBuffer.allocate(4).putInt(0, i); @@ -387,6 +402,17 @@ public class ByteBufferUtil return ByteBuffer.allocate(8).putLong(0, n); } + public static ByteBuffer bytes(float f) + { + return ByteBuffer.allocate(4).putFloat(0, f); + } + + public static ByteBuffer bytes(double d) + { + return ByteBuffer.allocate(8).putDouble(0, d); + } + + public static InputStream inputStream(ByteBuffer bytes) { final ByteBuffer copy = bytes.duplicate(); diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 9567ec65a4..3a1f58bc10 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -543,6 +543,31 @@ public class FBUtilities } } + public static T getInstance(String classname, String readable) throws ConfigurationException + { + Class cls = classForName(classname, readable); + T rval = null; + try + { + rval = (T) cls.getDeclaredMethod("getInstance").invoke(null, (Object) null); + + } + catch (NoSuchMethodException e) + { + throw new ConfigurationException("Class does not have the getInstance method with no arguments"); + } + catch (InvocationTargetException e) + { + throw new ConfigurationException(String.format("Could not call method getInstance on %s class %s", readable, classname)); + } + catch (IllegalAccessException e) + { + throw new ConfigurationException(String.format("Could not call method getInstance on %s class %s", readable, classname)); + } + + return rval; + } + public static SortedSet singleton(T column) { return new TreeSet(Arrays.asList(column)); From 6d777b46f49e3f8af9b96911a4455bbeaac86ab2 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Mon, 4 Apr 2011 22:15:39 +0000 Subject: [PATCH 07/15] Uncomment tests that were commented for no apparent reason git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1088805 13f79535-47bb-0310-9956-ffa450edef68 --- test/unit/org/apache/cassandra/dht/PartitionerTestCase.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java b/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java index c398e56656..e7ea6bd788 100644 --- a/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java +++ b/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java @@ -97,8 +97,8 @@ public abstract class PartitionerTestCase @Test public void testMidpointWrapping() { - //assertMidpoint(tok("b"), tok("a"), 16); - //assertMidpoint(tok("bbb"), tok("a"), 16); + assertMidpoint(tok("b"), tok("a"), 16); + assertMidpoint(tok("bbb"), tok("a"), 16); } @Test From adbff64e2d345a5653d5b46cf8a08f8c35d2f5e7 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 5 Apr 2011 00:34:56 +0000 Subject: [PATCH 08/15] Namepsace the udf context key to avoid accidental overwrites in the future git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1088835 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/cassandra/hadoop/pig/CassandraStorage.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 4eea86446a..348cc4a11f 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -68,7 +68,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo public final static String PIG_INITIAL_ADDRESS = "PIG_INITIAL_ADDRESS"; public final static String PIG_PARTITIONER = "PIG_PARTITIONER"; - private static String UDFCONTEXT_SCHEMA_KEY = "schema"; + private static String UDFCONTEXT_SCHEMA_KEY = "cassandra.schema"; private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER; private static final Log logger = LogFactory.getLog(CassandraStorage.class); From ee1bb5aa95b1975981b31c13a08df069d520894f Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 5 Apr 2011 15:30:31 +0000 Subject: [PATCH 09/15] halve default memtable thresholds patch by jbellis; reviewed by bcoverston for CASSANDRA-2413 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089079 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/config/CFMetaData.java | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index 123dc0073d..7f7b65e4c9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -23,6 +23,7 @@ index (CASSANDRA-2376) * fix race condition that could leave orphaned data files when dropping CF or KS (CASSANDRA-2381) + * halve default memtable thresholds (CASSANDRA-2413) 0.7.4 diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 2c20f5b0d8..705b104cfb 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -101,7 +101,7 @@ public final class CFMetaData */ public static int sizeMemtableThroughput() { - return (int) (Runtime.getRuntime().maxMemory() / (1048576 * 16)); + return (int) (Runtime.getRuntime().maxMemory() / (1048576 * 32)); } /** From bdba69ff8d59c975bdfb5ba60d3435c757633209 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Tue, 5 Apr 2011 17:27:01 +0000 Subject: [PATCH 10/15] Convert mmap assertion to if/throw patch by jbellis; reviewed by slebresne for CASSANDRA-2417 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089139 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/cassandra/io/util/MappedFileDataInput.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java b/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java index a5e16cc800..5f410b6cbc 100644 --- a/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java +++ b/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java @@ -114,9 +114,9 @@ public class MappedFileDataInput extends AbstractDataInput implements FileDataIn public synchronized ByteBuffer readBytes(int length) throws IOException { int remaining = buffer.remaining() - position; - - assert length <= remaining - : String.format("mmap segment underflow; remaining is %d but %d requested", remaining, length); + if (length > remaining) + throw new IOException(String.format("mmap segment underflow; remaining is %d but %d requested", + remaining, length)); ByteBuffer bytes = buffer.duplicate(); bytes.position(buffer.position() + position).limit(buffer.position() + position + length); From 7e1e4b95195251bd6f56d30f504ade09ef6a7b3f Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 5 Apr 2011 22:00:00 +0000 Subject: [PATCH 11/15] Optimize schema fetch/store. Patch by Jeremy Hanna, reviewed by brandonwilliams for CASSANDRA-2421 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089261 13f79535-47bb-0310-9956-ffa450edef68 --- .../hadoop/pig/CassandraStorage.java | 68 ++++++++++--------- 1 file changed, 37 insertions(+), 31 deletions(-) diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 348cc4a11f..529ce3d05c 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -168,7 +168,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo private CfDef getCfDef() { UDFContext context = UDFContext.getUDFContext(); - Properties property = context.getUDFProperties(ResourceSchema.class); + Properties property = context.getUDFProperties(CassandraStorage.class); return cfdefFromString(property.getProperty(UDFCONTEXT_SCHEMA_KEY)); } @@ -314,6 +314,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo setLocationFromUri(location); ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family); setConnectionInformation(); + initSchema(); } public OutputFormat getOutputFormat() @@ -443,41 +444,46 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo private void initSchema() { - Cassandra.Client client = null; - try + UDFContext context = UDFContext.getUDFContext(); + Properties property = context.getUDFProperties(CassandraStorage.class); + + // Only get the schema if we haven't already gotten it + if (!property.containsKey(UDFCONTEXT_SCHEMA_KEY)) { - client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); - CfDef cfDef = null; - client.set_keyspace(keyspace); - KsDef ksDef = client.describe_keyspace(keyspace); - List defs = ksDef.getCf_defs(); - for (CfDef def : defs) + Cassandra.Client client = null; + try { - if (column_family.equalsIgnoreCase(def.getName())) + client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); + CfDef cfDef = null; + client.set_keyspace(keyspace); + KsDef ksDef = client.describe_keyspace(keyspace); + List defs = ksDef.getCf_defs(); + for (CfDef def : defs) { - cfDef = def; - break; + if (column_family.equalsIgnoreCase(def.getName())) + { + cfDef = def; + break; + } } + property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef)); + } + catch (TException e) + { + throw new RuntimeException(e); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (NotFoundException e) + { + throw new RuntimeException(e); + } + catch (IOException e) + { + throw new RuntimeException(e); } - UDFContext context = UDFContext.getUDFContext(); - Properties property = context.getUDFProperties(ResourceSchema.class); - property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef)); - } - catch (TException e) - { - throw new RuntimeException(e); - } - catch (InvalidRequestException e) - { - throw new RuntimeException(e); - } - catch (NotFoundException e) - { - throw new RuntimeException(e); - } - catch (IOException e) - { - throw new RuntimeException(e); } } From cc0322fef8cfa526932d72784e261f0dea519516 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 6 Apr 2011 17:35:20 +0000 Subject: [PATCH 12/15] fix FBU.getInstance git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089542 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/utils/FBUtilities.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 3a1f58bc10..732d143cb1 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -549,7 +549,7 @@ public class FBUtilities T rval = null; try { - rval = (T) cls.getDeclaredMethod("getInstance").invoke(null, (Object) null); + rval = (T) cls.getDeclaredMethod("getInstance").invoke(new Object[] {null, null}); } catch (NoSuchMethodException e) From 381401af7095c9244abe81ff2c13617c1fdf975a Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 7 Apr 2011 15:53:08 +0000 Subject: [PATCH 13/15] remove unnecessary copy during range/index scans patch by jbellis; reviewed by slebresne for CASSANDRA-2425 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089920 13f79535-47bb-0310-9956-ffa450edef68 --- .../service/RangeSliceResponseResolver.java | 15 ++++----- .../cassandra/service/StorageProxy.java | 32 ++++++++----------- 2 files changed, 19 insertions(+), 28 deletions(-) diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index 4118dfe485..28923895d6 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java @@ -24,6 +24,8 @@ import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import com.google.common.collect.AbstractIterator; +import com.google.common.collect.Iterables; +import com.google.common.collect.Iterators; import org.apache.commons.collections.iterators.CollatingIterator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,7 +42,7 @@ import org.apache.cassandra.utils.ReducingIterator; * Turns RangeSliceReply objects into row (string -> CF) maps, resolving * to the most recent ColumnFamily and setting up read repairs as necessary. */ -public class RangeSliceResponseResolver implements IResponseResolver> +public class RangeSliceResponseResolver implements IResponseResolver> { private static final Logger logger_ = LoggerFactory.getLogger(RangeSliceResponseResolver.class); private final String table; @@ -62,7 +64,7 @@ public class RangeSliceResponseResolver implements IResponseResolver> // Note: this deserializes the response a 2nd time if getData was called first // (this is not currently an issue since we don't do read repair for range queries.) - public List resolve() throws IOException + public Iterable resolve() throws IOException { CollatingIterator collator = new CollatingIterator(new Comparator>() { @@ -81,7 +83,8 @@ public class RangeSliceResponseResolver implements IResponseResolver> } // for each row, compute the combination of all different versions seen, and repair incomplete versions - ReducingIterator, Row> iter = new ReducingIterator, Row>(collator) + + return new ReducingIterator, Row>(collator) { List versions = new ArrayList(sources.size()); List versionSources = new ArrayList(sources.size()); @@ -109,12 +112,6 @@ public class RangeSliceResponseResolver implements IResponseResolver> return new Row(key, resolved); } }; - - List resolvedRows = new ArrayList(n); - while (iter.hasNext()) - resolvedRows.add(iter.next()); - - return resolvedRows; } public void preprocess(Message message) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 0a01c1c522..96f853b332 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -512,7 +512,7 @@ public class StorageProxy implements StorageProxyMBean // collect replies and resolve according to consistency level RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); - ReadCallback> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints); + ReadCallback> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints); handler.assureSufficientLiveNodes(); for (InetAddress endpoint : handler.endpoints) { @@ -521,24 +521,21 @@ public class StorageProxy implements StorageProxyMBean logger.debug("reading " + c2 + " from " + endpoint); } - // if we're done, great, otherwise, move to the next range - try + try { - if (logger.isDebugEnabled()) + for (Row row : handler.get()) { - for (Row row : handler.get()) - { - logger.debug("range slices read " + row.key); - } + rows.add(row); + logger.debug("range slices read {}", row.key); } - rows.addAll(handler.get()); - } + } catch (DigestMismatchException e) { throw new AssertionError(e); // no digests in range slices yet } } + // if we're done, great, otherwise, move to the next range if (rows.size() >= command.max_keys) break; } @@ -771,7 +768,7 @@ public class StorageProxy implements StorageProxyMBean return keyspace; } }; - ReadCallback> handler = getReadCallback(resolver, iCommand, consistency_level, liveEndpoints); + ReadCallback> handler = getReadCallback(resolver, iCommand, consistency_level, liveEndpoints); handler.assureSufficientLiveNodes(); IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range); @@ -783,21 +780,18 @@ public class StorageProxy implements StorageProxyMBean logger.debug("reading " + command + " from " + endpoint); } - List theseRows; try { - theseRows = handler.get(); + for (Row row : handler.get()) + { + rows.add(row); + logger.debug("read {}", row); + } } catch (DigestMismatchException e) { throw new RuntimeException(e); } - rows.addAll(theseRows); - if (logger.isDebugEnabled()) - { - for (Row row : theseRows) - logger.debug("read " + row); - } if (rows.size() >= index_clause.count) return rows.subList(0, index_clause.count); } From 705cd1e8bbfe06246ca31894395eb101eb95d4a2 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 7 Apr 2011 15:54:24 +0000 Subject: [PATCH 14/15] add date in seconds-since-epoch to default gc log filename patch by Chris Burroughs and jbellis for CASSANDRA-2418 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089924 13f79535-47bb-0310-9956-ffa450edef68 --- conf/cassandra-env.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conf/cassandra-env.sh b/conf/cassandra-env.sh index 9e98be2b82..dc105d8838 100644 --- a/conf/cassandra-env.sh +++ b/conf/cassandra-env.sh @@ -136,7 +136,7 @@ JVM_OPTS="$JVM_OPTS -XX:+UseCMSInitiatingOccupancyOnly" # JVM_OPTS="$JVM_OPTS -XX:+PrintClassHistogram" # JVM_OPTS="$JVM_OPTS -XX:+PrintTenuringDistribution" # JVM_OPTS="$JVM_OPTS -XX:+PrintGCApplicationStoppedTime" -# JVM_OPTS="$JVM_OPTS -Xloggc:/var/log/cassandra/gc.log" +# JVM_OPTS="$JVM_OPTS -Xloggc:/var/log/cassandra/gc-`date +%s`.log" # uncomment to have Cassandra JVM listen for remote debuggers/profilers on port 1414 # JVM_OPTS="$JVM_OPTS -Xdebug -Xnoagent -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=1414" From 51a07442885360168fdad33b6da792911e07dd83 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Thu, 7 Apr 2011 20:12:22 +0000 Subject: [PATCH 15/15] Try harder to close files after compaction patch by stuhood; reviewed by slebresne for CASSANDRA-2431 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089976 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../cassandra/io/CompactionIterator.java | 6 ++---- .../apache/cassandra/io/util/FileUtils.java | 21 +++++++++++++++++++ 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 7f7b65e4c9..0a11e6c56b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -24,6 +24,7 @@ * fix race condition that could leave orphaned data files when dropping CF or KS (CASSANDRA-2381) * halve default memtable thresholds (CASSANDRA-2413) + * Try harder to close files after compaction (CASSANDRA-2431) 0.7.4 diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java index 128162dc22..6bfadb34bd 100644 --- a/src/java/org/apache/cassandra/io/CompactionIterator.java +++ b/src/java/org/apache/cassandra/io/CompactionIterator.java @@ -36,6 +36,7 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.io.sstable.SSTableIdentityIterator; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.sstable.SSTableScanner; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.ReducingIterator; @@ -149,10 +150,7 @@ implements Closeable, ICompactionInfo public void close() throws IOException { - for (SSTableScanner scanner : getScanners()) - { - scanner.close(); - } + FileUtils.close(getScanners()); } protected Iterable getScanners() diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index 265768c5f5..4ffd3c799a 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -20,6 +20,7 @@ package org.apache.cassandra.io.util; import java.io.*; import java.text.DecimalFormat; +import java.util.Collection; import java.util.Comparator; import java.util.List; @@ -95,6 +96,26 @@ public class FileUtils } } + public static void close(Iterable cs) throws IOException + { + IOException e = null; + for (Closeable c : cs) + { + try + { + if (c != null) + c.close(); + } + catch (IOException ex) + { + e = ex; + logger_.warn("Failed closing stream " + c, ex); + } + } + if (e != null) + throw e; + } + public static class FileComparator implements Comparator { public int compare(File f, File f2)