lib/kwery/executor/reshard.rb
# target is the rs (replica set) number
target is the rs (replica set) number
target is the rs ( replica set ) number
def call(context)
        context.shards.lock(@table, @shard)
        sharding_key = context.shards.key(@table)
        num_shards   = context.shards.count(@table)
        source       = context.shards.primary_for_shard(@table, @shard)
        client = Kwery::Client::Batch.new
        result = client.query(
          source => "SELECT * FROM #{@table} WHERE hashmod(#{sharding_key}, #{num_shards}) = #{@shard}",
        )
        tups = result.first[:data]
        format = Kwery::Format::Json.new
        stdin = format.encode(tups).join
        target = context.shards.primary_for_rs(@target)
        client.query(
          target => {
            query: "COPY #{@table} FROM STDIN",
            data:  stdin,
          },
        )
        client.query(
          source => "DELETE FROM #{@table} WHERE hashmod(#{sharding_key}, #{num_shards}) = #{@shard}",
        )
        context.shards.reassign(@table, @shard, @target)
        context.shards.unlock(@table, @shard)
        [{ success: true }]
      end
def call ( context ) context . shards . lock ( @table , @shard ) sharding_key = context . shards . key ( @table ) num_shards = context . shards . count ( @table ) source = context . shards . primary_for_shard ( @table , @shard ) client = Kwery :: Client :: Batch . new result = client . query ( source => "SELECT * FROM #{@table} WHERE hashmod(#{sharding_key}, #{num_shards}) = #{@shard}" , ) tups = result . first [ :data ] format = Kwery :: Format :: Json . new stdin = format . encode ( tups ) . join target = context . shards . primary_for_rs ( @target ) client . query ( target => { query : "COPY #{@table} FROM STDIN" , data : stdin , } , ) client . query ( source => "DELETE FROM #{@table} WHERE hashmod(#{sharding_key}, #{num_shards}) = #{@shard}" , ) context . shards . reassign ( @table , @shard , @target ) context . shards . unlock ( @table , @shard ) [ { success : true } ] end
target is the rs (replica set) number
target is the rs ( replica set ) number
# TODO: split large stream into smaller requests
context