Support multiple versions of Cassandra

Rakefile

@@ -6,12 +6,12 @@ require 'rspec/core'
 require 'rspec/core/rake_task'
 
 CassandraBinaries = {
-  '0.8' => 'http://archive.apache.org/dist/cassandra/0.8.4/apache-cassandra-0.8.4-bin.tar.gz',
-  '1.0' => 'http://archive.apache.org/dist/cassandra/1.0.1/apache-cassandra-1.0.1-bin.tar.gz',
+  '0.8' => 'http://archive.apache.org/dist/cassandra/0.8.8/apache-cassandra-0.8.8-bin.tar.gz',
+  '1.0' => 'http://archive.apache.org/dist/cassandra/1.0.5/apache-cassandra-1.0.5-bin.tar.gz',
 }
 
 CASSANDRA_VERSION = ENV['CASSANDRA_VERSION'] || '1.0'
-CASSANDRA_HOME = File.dirname(__FILE__) + '/tmp'
+CASSANDRA_HOME = ENV['CASSANDRA_HOME'] || File.dirname(__FILE__) + '/tmp'
 CASSANDRA_PIDFILE = ENV['CASSANDRA_PIDFILE'] || "#{CASSANDRA_HOME}/cassandra.pid"
 
 RSpec::Core::RakeTask.new(:spec) do |spec|
@@ -111,12 +111,12 @@ namespace :cassandra do
     env = setup_environment
     sh("kill $(cat #{CASSANDRA_PIDFILE})")
   end
-  
+
   desc "Delete all data files in #{CASSANDRA_HOME}"
   task :clean do
     sh("rm -rf #{File.join(CASSANDRA_HOME, "cassandra-#{CASSANDRA_VERSION}", 'data')}")
   end
-  
+
 end
 
 desc "Start Cassandra"

lib/cassandra-cql.rb

@@ -14,8 +14,13 @@
 limitations under the License.
 =end
 
-here = File.dirname(__FILE__)
-require "#{here}/../vendor/gen-rb/cassandra"
+module CassandraCQL; end;
+unless CassandraCQL.respond_to?(:CASSANDRA_VERSION)
+  require "cassandra-cql/1.0"
+end
+
+here = File.expand_path(File.dirname(__FILE__))
+require "#{here}/../vendor/#{CassandraCQL.CASSANDRA_VERSION}/gen-rb/cassandra"
 
 require 'bigdecimal'
 require 'date'
@@ -39,3 +44,6 @@
 require 'cassandra-cql/statement'
 require 'cassandra-cql/result'
 require 'cassandra-cql/row'
+
+require "cassandra-cql/#{CassandraCQL.CASSANDRA_VERSION}/result"
+require "cassandra-cql/#{CassandraCQL.CASSANDRA_VERSION}/statement"

lib/cassandra-cql/0.8.rb

@@ -0,0 +1,7 @@
+module CassandraCQL
+  def self.CASSANDRA_VERSION
+    "0.8"
+  end
+end
+
+require "#{File.expand_path(File.dirname(__FILE__))}/../cassandra-cql"

lib/cassandra-cql/0.8/result.rb

@@ -0,0 +1,23 @@
+module CassandraCQL
+  module V08
+    class ResultSchema < CassandraCQL::ResultSchema
+      def initialize(column_family)
+        type_slice = lambda {|type| type[type.rindex('.')+1..-1] }
+
+        @names = Hash.new(type_slice.call(column_family.comparator_type))
+        @values = Hash.new(type_slice.call(column_family.default_validation_class))
+        column_family.columns.each_pair do |name, type|
+          @values[name] = type_slice.call(type)
+        end
+      end
+    end
+
+    class Result < CassandraCQL::Result
+      def initialize(result, column_family)
+        @result = result
+        @schema = ResultSchema.new(column_family) if rows?
+        @cursor = 0
+      end
+    end
+  end
+end

lib/cassandra-cql/0.8/statement.rb

@@ -0,0 +1,38 @@
+module CassandraCQL
+  module V08
+    class Statement < CassandraCQL::Statement
+      SCHEMA_CHANGE_RE = /\s*(create|drop|alter)\s+(\w+)/i
+      COLFAM_RE = /\s*select.*from\s+'?(\w+)/i
+
+      def execute(bind_vars=[], options={})
+        column_family = nil
+        if @statement =~ COLFAM_RE
+          column_family = @handle.schema.column_families[$1].dup
+        end
+
+        sanitized_query = self.class.sanitize(@statement, bind_vars)
+        compression_type = CassandraCQL::Thrift::Compression::NONE
+        if options[:compression]
+          compression_type = CassandraCQL::Thrift::Compression::GZIP
+          sanitized_query = Utility.compress(sanitized_query)
+        end
+
+        res = V08::Result.new(@handle.execute_cql_query(sanitized_query, compression_type), column_family)
+
+        # Change our keyspace if required
+        if @statement =~ KS_CHANGE_RE
+          @handle.keyspace = $1
+        elsif @statement =~ KS_DROP_RE
+          @handle.keyspace = nil
+        end
+
+        # We let ints be fetched for now because they'll probably be deprecated later
+        if res.void?
+          nil
+        else
+          res
+        end
+      end
+    end
+  end
+end

lib/cassandra-cql/1.0.rb

@@ -0,0 +1,7 @@
+module CassandraCQL
+  def self.CASSANDRA_VERSION
+    "1.0"
+  end
+end
+
+require "#{File.expand_path(File.dirname(__FILE__))}/../cassandra-cql"

lib/cassandra-cql/1.0/result.rb

@@ -0,0 +1,6 @@
+module CassandraCQL
+  module V10
+    class Result < CassandraCQL::Result
+    end
+  end
+end

lib/cassandra-cql/1.0/statement.rb

@@ -0,0 +1,6 @@
+module CassandraCQL
+  module V10
+    class Statement < CassandraCQL::Statement
+    end
+  end
+end

lib/cassandra-cql/database.rb

@@ -64,8 +64,15 @@ def reset!
     end
     alias_method :reconnect!, :reset!
 
+    def statement_class
+      return @statement_class if @statement_class
+
+      version_module = 'V' + CassandraCQL.CASSANDRA_VERSION.gsub('.', '')
+      return @statement_class = CassandraCQL.const_get(version_module).const_get(:Statement)
+    end
+
     def prepare(statement, options={}, &block)
-      stmt = Statement.new(self, statement)
+      stmt = statement_class.new(self, statement)
       if block_given?
         yield stmt
       else
@@ -74,7 +81,7 @@ def prepare(statement, options={}, &block)
     end
 
     def execute(statement, *bind_vars)
-      result = Statement.new(self, statement).execute(bind_vars)
+      result = statement_class.new(self, statement).execute(bind_vars)
       if block_given?
         yield result
       else

lib/cassandra-cql/statement.rb

@@ -31,18 +31,21 @@ def initialize(handle, statement)
       @handle = handle
       prepare(statement)
     end
-  
+
     def prepare(statement)
       @statement = statement
     end
-  
+
     def execute(bind_vars=[], options={})
+      sanitized_query = self.class.sanitize(@statement, bind_vars)
+      compression_type = CassandraCQL::Thrift::Compression::NONE
       if options[:compression]
-        res = Result.new(@handle.execute_cql_query(Utility.compress(self.class.sanitize(@statement, bind_vars)), CassandraCQL::Thrift::Compression::GZIP))
-      else
-        res = Result.new(@handle.execute_cql_query(self.class.sanitize(@statement, bind_vars), CassandraCQL::Thrift::Compression::NONE))
+        compression_type = CassandraCQL::Thrift::Compression::GZIP
+        sanitized_query = Utility.compress(sanitized_query)
       end
 
+      res = Result.new(@handle.execute_cql_query(sanitized_query, compression_type))
+
       # Change our keyspace if required
       if @statement =~ KS_CHANGE_RE
         @handle.keyspace = $1

spec/comparator_spec.rb

@@ -97,23 +97,24 @@ def test_for_value(value)
       end
       res.class.should eq(Time)
     end
-
   end
 
-  context "with decimal comparator" do
-    let(:cf_name) { "comparator_cf_decimal" }
-    before(:each) { create_column_family(cf_name, 'DecimalType') }
+  if CASSANDRA_VERSION.to_f >= 1.0
+    context "with decimal comparator" do
+      let(:cf_name) { "comparator_cf_decimal" }
+      before(:each) { create_column_family(cf_name, 'DecimalType') }
 
-    def test_for_value(value)
-      create_and_fetch_column(cf_name, value).should eq(value)
-      create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
-    end
-
-    it "should return a small decimal" do
-      test_for_value(15.333)
-    end
-    it "should return a huge decimal" do
-      test_for_value(BigDecimal.new('129182739481237481341234123411.1029348102934810293481039'))
+      def test_for_value(value)
+        create_and_fetch_column(cf_name, value).should eq(value)
+        create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
+      end
+
+      it "should return a small decimal" do
+        test_for_value(15.333)
+      end
+      it "should return a huge decimal" do
+        test_for_value(BigDecimal.new('129182739481237481341234123411.1029348102934810293481039'))
+      end
     end
   end
 
@@ -151,26 +152,29 @@ def test_for_value(value)
     end
   end
 
-  context "with int comparator" do
-    let(:cf_name) { "comparator_cf_int" }
-    before(:each) { create_column_family(cf_name, 'Int32Type') }
+  if CASSANDRA_VERSION.to_f >= 1.0
+    #Int32Type was added in 1.0 (CASSANDRA-3031)
+    context "with int comparator" do
+      let(:cf_name) { "comparator_cf_int" }
+      before(:each) { create_column_family(cf_name, 'Int32Type') }
 
-    def test_for_value(value)
-      create_and_fetch_column(cf_name, value).should eq(value)
-      create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
-    end
-
-    it "should properly convert integer values that fit into 1 byte" do
-      test_for_value(1)
-    end
-    it "should properly convert integer values that fit into 2 bytes" do
-      test_for_value(2**8 + 80)
-    end
-    it "should properly convert integer values that fit into 3 bytes" do
-      test_for_value(2**16 + 622)
-    end
-    it "should properly convert integer values that fit into 4 bytes" do
-      test_for_value(2**24 + 45820)
+      def test_for_value(value)
+        create_and_fetch_column(cf_name, value).should eq(value)
+        create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
+      end
+
+      it "should properly convert integer values that fit into 1 byte" do
+        test_for_value(1)
+      end
+      it "should properly convert integer values that fit into 2 bytes" do
+        test_for_value(2**8 + 80)
+      end
+      it "should properly convert integer values that fit into 3 bytes" do
+        test_for_value(2**16 + 622)
+      end
+      it "should properly convert integer values that fit into 4 bytes" do
+        test_for_value(2**24 + 45820)
+      end
     end
   end
 

spec/rowkey_spec.rb

@@ -56,7 +56,11 @@ def test_for_value(value)
 
   context "with blob row_key_validation" do
     let(:cf_name) { "row_key_validation_cf_blob" }
-    before(:each) { create_column_family(cf_name, 'blob') }
+    if CASSANDRA_VERSION.to_f == 0.8
+      before(:each) { create_column_family(cf_name, 'bytea') }
+    else
+      before(:each) { create_column_family(cf_name, 'blob') }
+    end
 
     it "should return a blob" do
       bytes = "binary\x00"
@@ -78,20 +82,22 @@ def test_for_value(value)
     end
   end
 
-  context "with decimal row_key_validation" do
-    let(:cf_name) { "row_key_validation_cf_decimal" }
-    before(:each) { create_column_family(cf_name, 'decimal') }
+  if CASSANDRA_VERSION.to_f >= 1.0
+    context "with decimal row_key_validation" do
+      let(:cf_name) { "row_key_validation_cf_decimal" }
+      before(:each) { create_column_family(cf_name, 'decimal') }
 
-    def test_for_value(value)
-      create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
-      create_and_fetch_column(cf_name, value).should eq(value)
-    end
-
-    it "should return a small decimal" do
-      test_for_value(15.333)
-    end
-    it "should return a huge decimal" do
-      test_for_value(BigDecimal.new('129182739481237481341234123411.1029348102934810293481039'))
+      def test_for_value(value)
+        create_and_fetch_column(cf_name, value*-1).should eq(value*-1)
+        create_and_fetch_column(cf_name, value).should eq(value)
+      end
+
+      it "should return a small decimal" do
+        test_for_value(15.333)
+      end
+      it "should return a huge decimal" do
+        test_for_value(BigDecimal.new('129182739481237481341234123411.1029348102934810293481039'))
+      end
     end
   end
 
@@ -167,7 +173,11 @@ def test_for_value(value)
 
   context "with timestamp row_key_validation" do
     let(:cf_name) { "row_key_validation_cf_timestamp" }
-    before(:each) { create_column_family(cf_name, 'timestamp') }
+    if CASSANDRA_VERSION.to_f == 0.8
+      before(:each) { create_column_family(cf_name, 'date') }
+    else
+      before(:each) { create_column_family(cf_name, 'timestamp') }
+    end
 
     it "should return a timestamp" do
       uuid = UUID.new

spec/spec_helper.rb

@@ -1,7 +1,10 @@
 require 'yaml'
 require 'rspec'
+
+CASSANDRA_VERSION = ENV['CASSANDRA_VERSION'] || '1.0' unless defined?(CASSANDRA_VERSION)
+
 $LOAD_PATH << "#{File.expand_path(File.dirname(__FILE__))}/../lib"
-require 'cassandra-cql'
+require "cassandra-cql/#{CASSANDRA_VERSION}"
 
 def yaml_fixture(file)
   if file.kind_of?(Symbol)
@@ -13,11 +16,11 @@ def yaml_fixture(file)
 end
 
 def setup_cassandra_connection
-  connection = CassandraCQL::Database.new(["127.0.0.1:9160"], {}, :retries => 2, :timeout => 1) rescue false
+  connection = CassandraCQL::Database.new(["127.0.0.1:9160"], {}, :retries => 2, :timeout => 1)
   if !connection.keyspaces.map(&:name).include?("CassandraCQLTestKeyspace")
     connection.execute("CREATE KEYSPACE CassandraCQLTestKeyspace WITH strategy_class='org.apache.cassandra.locator.SimpleStrategy' AND strategy_options:replication_factor=1")
   end
   connection.execute("USE CassandraCQLTestKeyspace")
-  
+
   connection
-end
+end