Allow specification of the order_column to remove need for a primary key

Instead of relying on a table to have a primary key, we now accept the
order_column option and put that everywhere. This kind of uncomfortable
state about the migration now needs to be given to everyone and I think
an appropriate place is the migration, since most objects in the system
have access to it and they all need to agree on the value. This required
passing the options down a couple more objects for the Migrator creating
the Migration to have access, but I think thats acceptable for this
increase in functionality.

lib/lhm.rb

@@ -40,12 +40,14 @@ module Lhm
   #   Use atomic switch to rename tables (defaults to: true)
   #   If using a version of mysql affected by atomic switch bug, LHM forces user
   #   to set this option (see SqlHelper#supports_atomic_switch?)
+  # @option options [String] :order_column
+  #   Column name to order records by. This column must be unique for every record (defaults to: "id")
   # @yield [Migrator] Yielded Migrator object records the changes
   # @return [Boolean] Returns true if the migration finishes
   # @raise [Error] Raises Lhm::Error in case of a error and aborts the migration
   def change_table(table_name, options = {}, &block)
     origin = Table.parse(table_name, connection)
-    invoker = Invoker.new(origin, connection)
+    invoker = Invoker.new(origin, connection, options)
     block.call(invoker.migrator)
     invoker.run(options)
     true
@@ -59,6 +61,7 @@ def change_table(table_name, options = {}, &block)
   #   Filter to only remove tables up to specified time (defaults to: nil)
   def cleanup(run = false, options = {})
     lhm_tables = connection.select_values("show tables").select { |name| name =~ /^lhm(a|n)_/ }
+
     if options[:until]
       lhm_tables.select!{ |table|
         table_date_time = Time.strptime(table, "lhma_%Y_%m_%d_%H_%M_%S")

lib/lhm/chunker.rb

@@ -52,16 +52,16 @@ def top(stride)
     def copy(lowest, highest)
       "insert ignore into `#{ destination_name }` (#{ columns }) " +
       "select #{ select_columns } from `#{ origin_name }` " +
-      "#{ conditions } `#{ origin_name }`.`id` between #{ lowest } and #{ highest }"
+      "#{ conditions } `#{ origin_name }`.`#{ @migration.order_column }` between #{ lowest } and #{ highest }"
     end
 
     def select_start
-      start = connection.select_value("select min(id) from `#{ origin_name }`")
+      start = connection.select_value("select min(#{ @migration.order_column }) from `#{ origin_name }`")
       start ? start.to_i : nil
     end
 
     def select_limit
-      limit = connection.select_value("select max(id) from `#{ origin_name }`")
+      limit = connection.select_value("select max(#{ @migration.order_column }) from `#{ origin_name }`")
       limit ? limit.to_i : nil
     end
 

lib/lhm/entangler.rb

@@ -17,6 +17,7 @@ def initialize(migration, connection = nil)
       @common = migration.intersection
       @origin = migration.origin
       @destination = migration.destination
+      @order_column = migration.order_column
       @connection = connection
     end
 
@@ -59,7 +60,7 @@ def create_delete_trigger
         create trigger `#{ trigger(:del) }`
         after delete on `#{ @origin.name }` for each row
         delete ignore from `#{ @destination.name }` #{ SqlHelper.annotation }
-        where `#{ @destination.name }`.`id` = OLD.`id`
+        where `#{ @destination.name }`.`#{ @order_column }` = OLD.`#{ @order_column }`
       }
     end
 

lib/lhm/invoker.rb

@@ -18,9 +18,9 @@ class Invoker
 
     attr_reader :migrator, :connection
 
-    def initialize(origin, connection)
+    def initialize(origin, connection, options)
       @connection = connection
-      @migrator = Migrator.new(origin, connection)
+      @migrator = Migrator.new(origin, connection, options)
     end
 
     def run(options = {})

lib/lhm/migration.rb

@@ -5,12 +5,13 @@
 
 module Lhm
   class Migration
-    attr_reader :origin, :destination, :conditions
+    attr_reader :origin, :destination, :conditions, :order_column
 
-    def initialize(origin, destination, conditions = nil, time = Time.now)
+    def initialize(origin, destination, order_column, conditions = nil, time = Time.now)
       @origin = origin
       @destination = destination
       @conditions = conditions
+      @order_column = order_column
       @start = time
     end
 

lib/lhm/migrator.rb

@@ -15,10 +15,11 @@ class Migrator
 
     attr_reader :name, :statements, :connection, :conditions
 
-    def initialize(table, connection = nil)
+    def initialize(table, connection = nil, options = nil)
       @connection = connection
       @origin = table
       @name = table.destination_name
+      @options = options
       @statements = []
     end
 
@@ -166,7 +167,11 @@ def validate
       end
 
       unless @origin.satisfies_primary_key?
-        error("origin does not satisfy primary key requirements")
+        if @options[:order_column]
+          error("order column needs to be specified because no satisfactory primary key exists") unless @origin.can_use_order_column?(@options[:order_column])
+        else
+          error("origin does not satisfy primary key requirements")
+        end
       end
 
       dest = @origin.destination_name
@@ -179,7 +184,7 @@ def validate
     def execute
       destination_create
       @connection.sql(@statements)
-      Migration.new(@origin, destination_read, conditions)
+      Migration.new(@origin, destination_read, order_column, conditions)
     end
 
     def destination_create
@@ -196,5 +201,9 @@ def index_ddl(cols, unique = nil, index_name = nil)
       parts = [type, index_name, @name, idx_spec(cols)]
       "create %s `%s` on `%s` (%s)" % parts
     end
+
+    def order_column
+      @options[:order_column] || @origin.pk
+    end
   end
 end

lib/lhm/table.rb

@@ -16,7 +16,11 @@ def initialize(name, pk = "id", ddl = nil)
     end
 
     def satisfies_primary_key?
-      @pk == "id"
+      @pk.is_a?(String) && numeric_type?(@columns[@pk][:type])
+    end
+
+    def can_use_order_column?(column)
+      numeric_type?(@columns[column][:type])
     end
 
     def destination_name
@@ -104,8 +108,16 @@ def extract_primary_key(schema)
           defn[column_name]
         end
 
-        keys.length == 1 ? keys.first : keys
+        case keys.length
+        when 0 then nil
+        when 1 then keys.first
+        else keys
+        end
       end
     end
+
+    def numeric_type?(type)
+      !!(type.match(/int\(\d+\)|decimal|numeric|float|double|integer/))
+    end
   end
 end

spec/fixtures/no_primary_key.ddl

@@ -0,0 +1,4 @@
+CREATE TABLE `no_primary_key` (
+  `foreign_id` int(11) NOT NULL,
+  `value` int(11) NOT NULL DEFAULT '0'
+) ENGINE=InnoDB DEFAULT CHARSET=utf8

spec/fixtures/primary_keys.ddl

@@ -0,0 +1,6 @@
+CREATE TABLE `primary_keys` (
+  `weird_id` int(11) NOT NULL AUTO_INCREMENT,
+  `origin` int(11) DEFAULT NULL,
+  `common` varchar(255) DEFAULT NULL,
+  PRIMARY KEY (`weird_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8

spec/integration/atomic_switcher_spec.rb

@@ -16,7 +16,7 @@
     before(:each) do
       @origin      = table_create("origin")
       @destination = table_create("destination")
-      @migration   = Lhm::Migration.new(@origin, @destination)
+      @migration   = Lhm::Migration.new(@origin, @destination, "id")
     end
 
     it "rename origin to archive" do

spec/integration/chunker_spec.rb

@@ -14,7 +14,7 @@
     before(:each) do
       @origin = table_create(:origin)
       @destination = table_create(:destination)
-      @migration = Lhm::Migration.new(@origin, @destination)
+      @migration = Lhm::Migration.new(@origin, @destination, "id")
     end
 
     it "should copy 23 rows from origin to destination" do

spec/integration/cleanup_spec.rb

@@ -32,7 +32,7 @@
 
     it "should show temporary tables within range" do
       table = OpenStruct.new(:name => 'users')
-      table_name = Lhm::Migration.new(table, nil, nil, Time.now - 172800).archive_name
+      table_name = Lhm::Migration.new(table, nil, nil, nil, Time.now - 172800).archive_name
       table_rename(:users, table_name)
 
       output = capture_stdout do
@@ -44,7 +44,7 @@
 
     it "should exclude temporary tables outside range" do
       table = OpenStruct.new(:name => 'users')
-      table_name = Lhm::Migration.new(table, nil, nil, Time.now).archive_name
+      table_name = Lhm::Migration.new(table, nil, nil, nil, Time.now).archive_name
       table_rename(:users, table_name)
 
       output = capture_stdout do

spec/integration/entangler_spec.rb

@@ -16,7 +16,7 @@
     before(:each) do
       @origin = table_create("origin")
       @destination = table_create("destination")
-      @migration = Lhm::Migration.new(@origin, @destination)
+      @migration = Lhm::Migration.new(@origin, @destination, "id")
       @entangler = Lhm::Entangler.new(@migration, connection)
     end
 

spec/integration/lhm_spec.rb

@@ -225,6 +225,38 @@
       end
     end
 
+    it "should change a table with a primary key other than id" do
+      table_create(:primary_keys)
+
+      Lhm.change_table(:primary_keys, :atomic_switch => false) do |t|
+        t.change_column(:weird_id, "int(5)")
+      end
+
+      slave do
+        table_read(:primary_keys).columns["weird_id"].must_equal({
+          :type => "int(5)",
+          :is_nullable => "NO",
+          :column_default => "0"
+        })
+      end
+    end
+
+    it "should change a table with a no primary key" do
+      table_create(:no_primary_key)
+
+      Lhm.change_table(:no_primary_key, :atomic_switch => false, :order_column => 'foreign_id') do |t|
+        t.change_column(:value, "text")
+      end
+
+      slave do
+        table_read(:no_primary_key).columns["value"].must_equal({
+          :type => "text",
+          :is_nullable => "YES",
+          :column_default => nil
+        })
+      end
+    end
+
     describe "parallel" do
       it "should perserve inserts during migration" do
         50.times { |n| execute("insert into users set reference = '#{ n }'") }

spec/integration/locked_switcher_spec.rb

@@ -16,7 +16,7 @@
     before(:each) do
       @origin = table_create("origin")
       @destination = table_create("destination")
-      @migration = Lhm::Migration.new(@origin, @destination)
+      @migration = Lhm::Migration.new(@origin, @destination, "id")
     end
 
     it "rename origin to archive" do

spec/unit/chunker_spec.rb

@@ -14,7 +14,7 @@
   before(:each) do
     @origin = Lhm::Table.new("foo")
     @destination = Lhm::Table.new("bar")
-    @migration = Lhm::Migration.new(@origin, @destination)
+    @migration = Lhm::Migration.new(@origin, @destination, "id")
     @connection = MiniTest::Mock.new
     # This is a poor man's stub
     @throttler = Object.new
@@ -117,4 +117,26 @@ def @migration.conditions
       @connection.verify
     end
   end
+
+  describe "copy into with a different column to order by" do
+    before(:each) do
+      @migration   = Lhm::Migration.new(@origin, @destination, "weird_id")
+      @origin.columns["secret"] = { :metadata => "VARCHAR(255)"}
+      @destination.columns["secret"] = { :metadata => "VARCHAR(255)"}
+      @chunker = Lhm::Chunker.new(@migration, @connection, :throttler => @throttler,
+                                  :start     => 1,
+                                  :limit     => 2)
+    end
+
+    it "should copy the correct range and column" do
+      @connection.expect(:update, 1) do |stmt|
+        stmt.first == "insert ignore into `bar` (`secret`) " +
+                      "select `foo`.`secret` from `foo` " +
+                      "where `foo`.`weird_id` between 1 and 1"
+      end
+
+      @chunker.run
+      @connection.verify
+    end
+  end
 end

spec/unit/entangler_spec.rb

@@ -13,7 +13,7 @@
   before(:each) do
     @origin = Lhm::Table.new("origin")
     @destination = Lhm::Table.new("destination")
-    @migration = Lhm::Migration.new(@origin, @destination)
+    @migration = Lhm::Migration.new(@origin, @destination, "id")
     @entangler = Lhm::Entangler.new(@migration)
   end
 

spec/unit/migration_spec.rb

@@ -13,7 +13,7 @@
     @start = Time.now
     @origin = Lhm::Table.new("origin")
     @destination = Lhm::Table.new("destination")
-    @migration = Lhm::Migration.new(@origin, @destination, nil, @start)
+    @migration = Lhm::Migration.new(@origin, @destination, "id", nil, @start)
   end
 
   it "should name archive" do
@@ -22,7 +22,7 @@
   end
 
   it "should limit table name to 64 characters" do
-    migration = Lhm::Migration.new(OpenStruct.new(:name => "a_very_very_long_table_name_that_should_make_the_LHMA_table_go_over_64_chars"), nil)
+    migration = Lhm::Migration.new(OpenStruct.new(:name => "a_very_very_long_table_name_that_should_make_the_LHMA_table_go_over_64_chars"), nil, "id")
     migration.archive_name.size == 64
   end
 end

spec/unit/table_spec.rb

@@ -18,11 +18,19 @@
   describe "constraints" do
     it "should be satisfied with a single column primary key called id" do
       @table = Lhm::Table.new("table", "id")
+      @table.columns["id"] = {:type => "int(11)"}
       @table.satisfies_primary_key?.must_equal true
     end
 
-    it "should not be satisfied with a primary key unless called id" do
-      @table = Lhm::Table.new("table", "uuid")
+    it "should be satisfied with a primary key called something other than id" do
+      @table = Lhm::Table.new("table", "weird_id")
+      @table.columns["weird_id"] = {:type => "int(11)"}
+      @table.satisfies_primary_key?.must_equal true
+    end
+
+    it "should not be satisfied with a non numeric primary key" do
+      @table = Lhm::Table.new("table", "id")
+      @table.columns["id"] = {:type => "varchar(255)"}
       @table.satisfies_primary_key?.must_equal false
     end