Class: ActiveRecord::ConnectionAdapters::HypertableAdapter

Inherits:
AbstractAdapter
  • Object
show all
Defined in:
lib/active_record/connection_adapters/hypertable_adapter.rb

Constant Summary collapse

@@read_latency =

Following cattr_accessors are used to record and access query performance statistics.

0.0
@@write_latency = 
0.0
@@cells_read = 
0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, logger, config) ⇒ HypertableAdapter

Returns a new instance of HypertableAdapter.



79
80
81
82
83
84
85
86
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 79

def initialize(connection, logger, config)
  super(connection, logger)
  @config = config
  @hypertable_column_names = {}
  if connection
    @namespace = connection.open_namespace(@config[:namespace])
  end
end

Instance Attribute Details

#retry_on_failure ⇒ Object

Returns the value of attribute retry_on_failure.



77
78
79
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 77

def retry_on_failure
  @retry_on_failure
end

Class Method Details

.get_timing ⇒ Object

Return the current set of performance statistics. The application can retrieve (and reset) these statistics after every query or request for its own logging purposes.



99
100
101
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 99

def self.get_timing
  [@@read_latency, @@write_latency, @@cells_read]
end

.reset_timing ⇒ Object

Reset performance metrics.



104
105
106
107
108
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 104

def self.reset_timing
  @@read_latency = 0.0
  @@write_latency = 0.0
  @@cells_read = 0
end

Instance Method Details

#adapter_name ⇒ Object 



110
111
112
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 110

def adapter_name
  'Hypertable'
end

#add_column(table_name, column_name, type = :string, options = {}) ⇒ Object 



451
452
453
454
455
456
457
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 451

def add_column(table_name, column_name, type=:string, options = {})
  hql = [ "ALTER TABLE #{quote_table_name(table_name)} ADD (" ]
  hql << quote_column_name(column_name)
  hql << "MAX_VERSIONS=#{options[:max_versions]}" if !options[:max_versions].blank?
  hql << ")"
  execute(hql.join(' '))
end

#add_column_options!(hql, options) ⇒ Object 



459
460
461
462
463
464
465
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 459

def add_column_options!(hql, options)
  hql << " MAX_VERSIONS =1 #{quote(options[:default], options[:column])}" if options_include_default?(options)
  # must explicitly check for :null to allow change_column to work on migrations
  if options[:null] == false
    hql << " NOT NULL"
  end
end

#add_column_to_name_map(table_name, name) ⇒ Object 



361
362
363
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 361

def add_column_to_name_map(table_name, name)
  @hypertable_column_names[table_name][rubify_column_name(name)] = name
end

#add_qualified_column(table_name, column_family, qualifiers = [], default = '', sql_type = nil, null = true) ⇒ Object 



365
366
367
368
369
370
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 365

def add_qualified_column(table_name, column_family, qualifiers=[], default='', sql_type=nil, null=true)
  qc = QualifiedColumn.new(column_family, default, sql_type, null)
  qc.qualifiers = qualifiers
  qualifiers.each{|q| add_column_to_name_map(table_name, qualified_column_name(column_family, q))}
  qc
end

#cell_native_array(row_key, column_family, column_qualifier, value = nil, timestamp = nil) ⇒ Object

Create native array format for cell. Most HyperRecord operations deal with cells in native array format since operations on an array are much faster than operations on Hypertable::ThriftGen::Cell objects. [“row_key”, “column_family”, “column_qualifier”, “value”],



595
596
597
598
599
600
601
602
603
604
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 595

def cell_native_array(row_key, column_family, column_qualifier, value=nil, timestamp=nil)
  [
    row_key.to_s,
    column_family.to_s,
    column_qualifier.to_s,
    value.to_s
  ].map do |s| 
    s.respond_to?(:force_encoding) ? s.force_encoding('ascii-8bit') : s 
  end
end

#change_column(table_name, column_name, new_column_name) ⇒ Object 



399
400
401
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 399

def change_column(table_name, column_name, new_column_name)
  raise "change_column operation not supported by Hypertable."
end

#change_column_default(table_name, column_name, default) ⇒ Object 



443
444
445
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 443

def change_column_default(table_name, column_name, default)
  raise "change_column_default operation not supported by Hypertable."
end

#change_column_null(table_name, column_name, null, default = nil) ⇒ Object 



447
448
449
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 447

def change_column_null(table_name, column_name, null, default = nil)
  raise "change_column_null operation not supported by Hypertable."
end

#close_mutator(mutator, flush = 0) ⇒ Object

Flush is always called in a mutator’s destructor due to recent no_log_sync changes. Adding an explicit flush here just adds one round trip for an extra flush call, so change the default to flush=0. Consider removing this argument and always sending 0.



667
668
669
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 667

def close_mutator(mutator, flush=0)
  @connection.close_mutator(mutator, flush)
end

#close_scanner(scanner) ⇒ Object 



681
682
683
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 681

def close_scanner(scanner)
  @connection.close_scanner(scanner)
end

#columns(table_name, name = nil) ⇒ Object

Returns array of column objects for the table associated with this class. Hypertable allows columns to include dashes in the name.

This doesn’t play well with Ruby (can’t have dashes in method names), so we maintain a mapping of original column names to Ruby-safe names.



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 332

def columns(table_name, name = nil)
  # Each table always has a row key called 'ROW'
  columns = [ Column.new('ROW', '') ]

  schema = describe_table(table_name)
  doc = REXML::Document.new(schema)
  column_families = doc.each_element('Schema/AccessGroup/ColumnFamily') { |cf| cf }

  @hypertable_column_names[table_name] ||= {}
  for cf in column_families
    # Columns are lazily-deleted in Hypertable so still may show up
    # in describe table output.  Ignore.
    del = cf.elements['deleted']
    deleted = del ? del.text : 'false'
    next if deleted == 'true'

    column_name = cf.elements['Name'].text
    rubified_name = rubify_column_name(column_name)
    @hypertable_column_names[table_name][rubified_name] = column_name
    columns << new_column(rubified_name, '')
  end

  columns
end

#convert_options_to_scan_spec(options = {}) ⇒ Object

Convert an options hash to a scan spec. A scan spec is native representation of the query parameters that must be sent to Hypertable. hypertable.org/thrift-api-ref/Client.html#Struct_ScanSpec



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 182

def convert_options_to_scan_spec(options={})
  sanitize_conditions(options)

  # Rows can be specified using a number of different options:
  # :row_keys => [row_key_1, row_key_2, ...]
  # :start_row and :end_row
  # :row_intervals => [[start_1, end_1], [start_2, end_2]]
  row_intervals = []

  options[:start_inclusive] = options.has_key?(:start_inclusive) ? options[:start_inclusive] : true
  options[:end_inclusive] = options.has_key?(:end_inclusive) ? options[:end_inclusive] : true

  if options[:row_keys]
    options[:row_keys].flatten.each do |rk|
      row_intervals << [rk, rk]
    end
  elsif options[:row_intervals]
    options[:row_intervals].each do |ri|
      row_intervals << [ri.first, ri.last]
    end
  elsif options[:start_row]
    raise "missing :end_row" if !options[:end_row]
    row_intervals << [options[:start_row], options[:end_row]]
  end

  # Add each row interval to the scan spec
  options[:row_intervals] = row_intervals.map do |row_interval|
    ri = Hypertable::ThriftGen::RowInterval.new
    ri.start_row = row_interval.first
    ri.start_inclusive = options[:start_inclusive]
    ri.end_row = row_interval.last
    ri.end_inclusive = options[:end_inclusive]
    ri
  end

  scan_spec = Hypertable::ThriftGen::ScanSpec.new

  # Hypertable can store multiple revisions for each cell but this
  # feature does not map into an ORM very well.  By default, just 
  # retrieve the latest revision of each cell.  Since this is most 
  # common config when using HyperRecord, tables should be defined 
  # with MAX_VERSIONS=1 at creation time to save space and reduce 
  # query time.
  options[:revs] ||= 1

  # Most of the time, we're not interested in cells that have been 
  # marked deleted but have not actually been deleted yet.
  options[:return_deletes] ||= false

  for key in options.keys
    case key.to_sym
      when :row_intervals
        scan_spec.row_intervals = options[key]
      when :cell_intervals
        scan_spec.cell_intervals = options[key]
      when :start_time
        scan_spec.start_time = options[key]
      when :end_time
        scan_spec.end_time = options[key]
      when :limit
        scan_spec.row_limit = options[key]
      when :revs
        scan_spec.revs = options[key]
      when :return_deletes
        scan_spec.return_deletes = options[key]
      when :select
        # Columns listed here can only be column families (not
        # column qualifiers) at this time.
        requested_columns = if options[key].is_a?(String)
          requested_columns_from_string(options[key])
        elsif options[key].is_a?(Symbol)
          requested_columns_from_string(options[key].to_s)
        elsif options[key].is_a?(Array)
           options[key].map{|k| k.to_s}
        else
          options[key]
        end

        scan_spec.columns = requested_columns.map do |column|
          status, family, qualifier = is_qualified_column_name?(column)
          family
        end.uniq
      when :table_name, :start_row, :end_row, :start_inclusive, :end_inclusive, :select, :columns, :row_keys, :conditions, :include, :readonly, :scan_spec, :instantiate_only_requested_columns
        # ignore
      else
        raise "Unrecognized scan spec option: #{key}"
    end
  end

  scan_spec
end

#create_table(table_name, options = {}, &block) ⇒ Object 



428
429
430
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 428

def create_table(table_name, options={}, &block)
  execute(create_table_hql(table_name, options, &block))
end

#create_table_hql(table_name, options = {}) {|table_definition| ... } ⇒ Object

Translate “sexy” ActiveRecord::Migration syntax to an HQL CREATE TABLE statement.

Yields:

  • (table_definition) 


405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 405

def create_table_hql(table_name, options={}, &block)
  table_definition = HyperTableDefinition.new(self)

  yield table_definition

  if options[:force] && table_exists?(table_name)
    drop_table(table_name, options)
  end

  create_sql = [ "CREATE TABLE #{quote_table_name(table_name)} (" ]
  column_sql = []
  for col in table_definition.columns
    column_sql << [
      quote_table_name(col.name),
      col.max_versions ? "MAX_VERSIONS=#{col.max_versions}" : ''
    ].join(' ')
  end
  create_sql << column_sql.join(', ')

  create_sql << ") #{options[:options]}"
  create_sql.join(' ').strip
end

#delete_cells(table_name, cells) ⇒ Object

Delete cells from a table.



607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 607

def delete_cells(table_name, cells)
  t1 = Time.now

  retry_on_connection_error {
    @connection.with_mutator(@namespace, table_name) do |mutator|
      thrift_cells = cells.map{|c|
        cell = thrift_cell_from_native_array(c)
        cell.key.flag = Hypertable::ThriftGen::KeyFlag::DELETE_CELL
        cell
      }
      @connection.set_cells(mutator, thrift_cells)
    end
  }

  @@write_latency += Time.now - t1
end

#delete_rows(table_name, row_keys) ⇒ Object

Delete rows from a table.



625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 625

def delete_rows(table_name, row_keys)
  t1 = Time.now
  cells = row_keys.map do |row_key|
    cell = Hypertable::ThriftGen::Cell.new
    cell.key = Hypertable::ThriftGen::Key.new
    cell.key.row = row_key
    cell.key.flag = Hypertable::ThriftGen::KeyFlag::DELETE_ROW
    cell
  end

  retry_on_connection_error {
    @connection.with_mutator(@namespace, table_name) do |mutator|
      @connection.set_cells(mutator, cells)
    end
  }

  @@write_latency += Time.now - t1
end

#describe_table(table_name) ⇒ Object

Return an XML document describing the table named in the first argument. Output is equivalent to that returned by the DESCRIBE TABLE command available in the Hypertable CLI. <Schema generation=“2”>

<AccessGroup name="default">
  <ColumnFamily id="1">
    <Generation>1</Generation>
    <Name>date</Name>
    <deleted>false</deleted>
  </ColumnFamily>
</AccessGroup>

</Schema>



519
520
521
522
523
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 519

def describe_table(table_name)
  retry_on_connection_error {
    @connection.get_schema_str(@namespace, table_name)
  }
end

#drop_table(table_name, options = {}) ⇒ Object 



432
433
434
435
436
437
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 432

def drop_table(table_name, options = {})
  retry_on_connection_error {
    @connection.drop_table(@namespace, table_name, 
      options[:if_exists] || false)
  }
end

#each_cell(scanner, &block) ⇒ Object

Iterator methods



691
692
693
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 691

def each_cell(scanner, &block)
  @connection.each_cell(scanner, &block)
end

#each_cell_as_arrays(scanner, &block) ⇒ Object 



695
696
697
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 695

def each_cell_as_arrays(scanner, &block)
  @connection.each_cell_as_arrays(scanner, &block)
end

#each_row(scanner, &block) ⇒ Object 



699
700
701
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 699

def each_row(scanner, &block)
  @connection.each_row(scanner, &block)
end

#each_row_as_arrays(scanner, &block) ⇒ Object 



703
704
705
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 703

def each_row_as_arrays(scanner, &block)
  @connection.each_row_as_arrays(scanner, &block)
end

#execute(hql, options = {}) ⇒ Object

Execute an HQL query against Hypertable and return the native HqlResult object that comes back from the Thrift client API.



141
142
143
144
145
146
147
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 141

def execute(hql, options={})
  log(hql, options[:name]) {
    retry_on_connection_error { 
      @connection.hql_query(@namespace, hql) 
    }
  }
end

#execute_with_options(options) ⇒ Object

Execute a query against Hypertable and return the matching cells. The query parameters are denoted in an options hash, which is converted to a “scan spec” by convert_options_to_scan_spec. A “scan spec” is the mechanism used to specify query parameters (e.g., the columns to retrieve, the number of rows to retrieve, etc.) to Hypertable.



155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 155

def execute_with_options(options)
  scan_spec = convert_options_to_scan_spec(options)
  t1 = Time.now

  # Use native array method (get_cells_as_arrays) for cell retrieval - 
  # much faster than get_cells that returns Hypertable::ThriftGen::Cell
  # objects.
  # [
  #   ["page_1", "name", "", "LOLcats and more", "1237331693147619001"], 
  #   ["page_1", "url", "", "http://...", "1237331693147619002"]
  # ]
  cells = retry_on_connection_error {
    @connection.get_cells_as_arrays(@namespace, options[:table_name], 
      scan_spec)
  }

  # Capture performance metrics
  @@read_latency += Time.now - t1
  @@cells_read += cells.length

  cells
end

#flush_mutator(mutator) ⇒ Object 



671
672
673
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 671

def flush_mutator(mutator)
  @connection.flush_mutator(mutator)
end

#handle_thrift_exceptions_with_missing_message ⇒ Object

Exceptions generated by Thrift IDL do not set a message. This causes a lot of problems for Rails which expects a String value and throws exception when it encounters NilClass. Unfortunately, you cannot assign a message to exceptions so define a singleton to accomplish same goal.



287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 287

def handle_thrift_exceptions_with_missing_message
  begin
    yield
  rescue Exception => err
    if !err.message
      if err.respond_to?("message=")
        err.message = err.what || ''
      else
        def err.message
          self.what || ''
        end
      end
    end

    raise err
  end
end

#hypertable_column_name(name, table_name, declared_columns_only = false) ⇒ Object 



490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 490

def hypertable_column_name(name, table_name, declared_columns_only=false)
  begin
    columns(table_name) if @hypertable_column_names[table_name].blank?
    n = @hypertable_column_names[table_name][name]
    n ||= name if !declared_columns_only
    n
  rescue Exception => err
    raise [
      "hypertable_column_name exception",
      err.message,
      "table: #{table_name}",
      "column: #{name}",
      "@htcn: #{pp @hypertable_column_names}"
    ].join("\n")
  end
end

#insert_fixture(fixture, table_name) ⇒ Object

Insert a test fixture into a table.



645
646
647
648
649
650
651
652
653
654
655
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 645

def insert_fixture(fixture, table_name)
  fixture_hash = fixture.to_hash
  timestamp = fixture_hash.delete('timestamp')
  row_key = fixture_hash.delete('ROW')
  cells = []
  fixture_hash.keys.each do |k|
    column_name, column_family = k.split(':', 2)
    cells << cell_native_array(row_key, column_name, column_family, fixture_hash[k], timestamp)
  end
  write_cells(table_name, cells)
end

#is_qualified_column_name?(column_name) ⇒ Boolean

Returns:

  • (Boolean) 


384
385
386
387
388
389
390
391
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 384

def is_qualified_column_name?(column_name)
  column_family, qualifier = column_name.split(':', 2)
  if qualifier
    [true, column_family, qualifier]
  else
    [false, column_name, nil]
  end
end

#native_database_types ⇒ Object

Hypertable only supports string types at the moment, so treat all values as strings and leave it to the application to handle types.



121
122
123
124
125
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 121

def native_database_types
  {
    :string      => { :name => "varchar", :limit => 255 }
  }
end

#new_column(column_name, default_value = '') ⇒ Object 



372
373
374
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 372

def new_column(column_name, default_value='')
  Column.new(rubify_column_name(column_name), default_value)
end

#open_mutator(table_name, flags = 0, flush_interval = 0) ⇒ Object

Mutator methods



659
660
661
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 659

def open_mutator(table_name, flags=0, flush_interval=0)
  @connection.open_mutator(@namespace, table_name, flags, flush_interval)
end

#open_scanner(table_name, scan_spec) ⇒ Object

Scanner methods



677
678
679
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 677

def open_scanner(table_name, scan_spec)
  @connection.open_scanner(@namespace, table_name, scan_spec, true)
end

#qualified_column_name(column_family, qualifier = nil) ⇒ Object 



376
377
378
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 376

def qualified_column_name(column_family, qualifier=nil)
  [column_family, qualifier].compact.join(':')
end

#quote(value, column = nil) ⇒ Object 



474
475
476
477
478
479
480
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 474

def quote(value, column = nil)
  case value
    when NilClass then ''
    when String then value
    else super(value, column)
  end
end

#quote_column_name(name) ⇒ Object 



482
483
484
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 482

def quote_column_name(name)
  "'#{name}'"
end

#quote_column_name_for_table(name, table_name) ⇒ Object 



486
487
488
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 486

def quote_column_name_for_table(name, table_name)
  quote_column_name(hypertable_column_name(name, table_name))
end

#raw_thrift_client(&block) ⇒ Object 



88
89
90
91
92
93
94
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 88

def raw_thrift_client(&block)
  t1 = Time.now
  results = Hypertable.with_thrift_client(@config[:host], @config[:port], 
    @config[:timeout], &block)
  @@read_latency += Time.now - t1
  results 
end

#remove_column(table_name, *column_names) ⇒ Object Also known as: remove_columns 



467
468
469
470
471
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 467

def remove_column(table_name, *column_names)
  column_names.flatten.each do |column_name|
    execute "ALTER TABLE #{quote_table_name(table_name)} DROP(#{quote_column_name(column_name)})"
  end
end

#remove_column_from_name_map(table_name, name) ⇒ Object 



357
358
359
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 357

def remove_column_from_name_map(table_name, name)
  @hypertable_column_names[table_name].delete(rubify_column_name(name))
end

#rename_column(table_name, column_name, new_column_name) ⇒ Object

Schema alterations



395
396
397
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 395

def rename_column(table_name, column_name, new_column_name)
  raise "rename_column operation not supported by Hypertable."
end

#rename_table(table_name, options = {}) ⇒ Object 



439
440
441
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 439

def rename_table(table_name, options = {})
  raise "rename_table operation not supported by Hypertable."
end

#requested_columns_from_string(s) ⇒ Object 



274
275
276
277
278
279
280
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 274

def requested_columns_from_string(s)
  if s == '*'
    []
  else
    s.split(',').map{|s| s.strip}
  end
end

#retry_on_connection_error ⇒ Object

Attempt to reconnect to the Thrift Broker once before aborting. This ensures graceful recovery in the case that the Thrift Broker goes down and then comes back up.



308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 308

def retry_on_connection_error
  @retry_on_failure = true
  begin
    handle_thrift_exceptions_with_missing_message { yield }
  rescue Thrift::TransportException, IOError, Thrift::ApplicationException, Thrift::ProtocolException => err
    if @retry_on_failure
      @retry_on_failure = false
      @connection.close
      @connection.open
      @namespace = @connection.open_namespace(@config[:namespace])
      retry
    else
      raise err
    end
  end
end

#rubify_column_name(column_name) ⇒ Object 



380
381
382
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 380

def rubify_column_name(column_name)
  column_name.to_s.gsub(/-+/, '_')
end

#sanitize_conditions(options) ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 127

def sanitize_conditions(options)
  case options[:conditions]
    when Hash
      # requires Hypertable API to support query by arbitrary cell value
      raise "HyperRecord does not support specifying conditions by Hash"
    when NilClass
      # do nothing
    else
      raise "Only hash conditions are supported"
  end
end

#supports_migrations? ⇒ Boolean

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 114

def supports_migrations?
  true
end

#tables(name = nil) ⇒ Object

Returns an array of tables available in the current Hypertable instance.



527
528
529
530
531
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 527

def tables(name=nil)
  retry_on_connection_error {
    @connection.get_tables(@namespace)
  }
end

#thrift_cell_from_native_array(array) ⇒ Object

Return a Hypertable::ThriftGen::Cell object from a cell passed in as an array of format: [row_key, column_name, value] Hypertable::ThriftGen::Cell objects are required when setting a flag on write - used by special operations (e.g,. delete )



579
580
581
582
583
584
585
586
587
588
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 579

def thrift_cell_from_native_array(array)
  cell = Hypertable::ThriftGen::Cell.new
  cell.key = Hypertable::ThriftGen::Key.new
  cell.key.row = array[0]
  cell.key.column_family = array[1]
  cell.key.column_qualifier = array[2] if !array[2].blank?
  cell.value = array[3] if array[3]
  cell.key.timestamp = array[4] if array[4]
  cell
end

#with_scanner(table_name, scan_spec, &block) ⇒ Object 



685
686
687
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 685

def with_scanner(table_name, scan_spec, &block)
  @connection.with_scanner(@namespace, table_name, scan_spec, &block)
end

#write_cells(table_name, cells, options = {}) ⇒ Object

Write an array of cells to the named table. By default, write_cells will open and close a mutator for this operation. Closing the mutator flushes the data, which guarantees is it is stored in Hypertable before the call returns. This also slows down the operation, so if you’re doing lots of writes and want to manage mutator flushes at the application layer then you can pass in a mutator as argument. Mutators can be created with the open_mutator method. In the near future (Summer 2009), Hypertable will provide a periodic mutator that automatically flushes at specific intervals.



542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
 
# File 'lib/active_record/connection_adapters/hypertable_adapter.rb', line 542

def write_cells(table_name, cells, options={})
  return if cells.blank?

  mutator = options[:mutator]
  flags = options[:flags]
  flush_interval = options[:flush_interval]

  retry_on_connection_error {
    local_mutator_created = !mutator

    begin
      t1 = Time.now
      mutator ||= open_mutator(table_name, flags, flush_interval)
      if options[:asynchronous_write]
        mutate_spec = Hypertable::ThriftGen::MutateSpec.new
        mutate_spec.appname = 'hyper_record'
        mutate_spec.flush_interval = 1000
        mutate_spec.flags = 2
        @connection.offer_cells_as_arrays(@namespace, table_name, 
          mutate_spec, cells)
      else
        @connection.set_cells_as_arrays(mutator, cells)
      end
    ensure
      if local_mutator_created && mutator
        close_mutator(mutator)
        mutator = nil
      end
      @@write_latency += Time.now - t1
    end
  }
end