Class: Tapsoob::DataStreamKeyed

Inherits:
DataStream show all
Defined in:
lib/tapsoob/data_stream.rb

Constant Summary

Constants inherited from DataStream

Tapsoob::DataStream::DEFAULT_CHUNKSIZE

Instance Attribute Summary collapse

Attributes inherited from DataStream

#db, #options, #state

Instance Method Summary collapse

Methods inherited from DataStream

#complete?, #encode_rows, #error, #error=, factory, #fetch, #fetch_chunksize, #fetch_data_in_database, #fetch_database, #fetch_file, #fetch_from_database, #fetch_rows, #import_rows, #log, #max_chunksize_training, #order_by, #parse_encoded_data, parse_json, #string_columns, #table, #table_name, #table_name_sql, #to_hash, #to_json, #update_chunksize_stats

Constructor Details

#initialize(db, state, opts = {}) ⇒ DataStreamKeyed

Returns a new instance of DataStreamKeyed.



293
294
295
296
297
298
 
# File 'lib/tapsoob/data_stream.rb', line 293

def initialize(db, state, opts = {})
  super(db, state, opts)
  @state = { :primary_key => order_by(state[:table_name]).first, :filter => 0 }.merge(@state)
  @state[:chunksize] ||= DEFAULT_CHUNKSIZE
  @buffer = []
end

Instance Attribute Details

#buffer ⇒ Object

Returns the value of attribute buffer.



291
292
293
 
# File 'lib/tapsoob/data_stream.rb', line 291

def buffer
  @buffer
end

Instance Method Details

#buffer_limit ⇒ Object 



304
305
306
307
308
309
310
 
# File 'lib/tapsoob/data_stream.rb', line 304

def buffer_limit
  if state[:last_fetched] and state[:last_fetched] < state[:filter] and self.buffer.size == 0
    state[:last_fetched]
  else
    state[:filter]
  end
end

#calc_limit(chunksize) ⇒ Object 



312
313
314
315
316
317
318
319
320
321
 
# File 'lib/tapsoob/data_stream.rb', line 312

def calc_limit(chunksize)
  # we want to not fetch more than is needed while we're
  # inside sinatra but locally we can select more than
  # is strictly needed
  if defined?(Sinatra)
    (chunksize * 1.1).ceil
  else
    (chunksize * 3).ceil
  end
end

#fetch_buffered(chunksize) ⇒ Object 



347
348
349
350
351
352
353
354
355
356
 
# File 'lib/tapsoob/data_stream.rb', line 347

def fetch_buffered(chunksize)
  load_buffer(chunksize) if self.buffer.size < chunksize
  rows = buffer.slice(0, chunksize)
  state[:last_fetched] = if rows.size > 0
    rows.last[ primary_key ]
  else
    nil
  end
  rows
end

#increment(row_count) ⇒ Object

def fetch_rows

chunksize = state[:chunksize]
Tapsoob::Utils.format_data(fetch_buffered(chunksize) || [],
  :string_columns => string_columns)

end



368
369
370
371
 
# File 'lib/tapsoob/data_stream.rb', line 368

def increment(row_count)
  # pop the rows we just successfully sent off the buffer
  @buffer.slice!(0, row_count)
end

#load_buffer(chunksize) ⇒ Object 



323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
 
# File 'lib/tapsoob/data_stream.rb', line 323

def load_buffer(chunksize)
  # make sure BasicObject is not polluted by subsequent requires
  Sequel::BasicObject.remove_methods!

  num = 0
  loop do
    limit = calc_limit(chunksize)
    # we have to use local variables in order for the virtual row filter to work correctly
    key = primary_key
    buf_limit = buffer_limit
    ds = table.order(*order_by).filter { key.sql_number > buf_limit }.limit(limit)
    log.debug "DataStreamKeyed#load_buffer SQL -> #{ds.sql}"
    data = ds.all
    self.buffer += data
    num += data.size
    if data.size > 0
      # keep a record of the last primary key value in the buffer
      state[:filter] = self.buffer.last[ primary_key ]
    end

    break if num >= chunksize or data.size == 0
  end
end

#primary_key ⇒ Object 



300
301
302
 
# File 'lib/tapsoob/data_stream.rb', line 300

def primary_key
  state[:primary_key].to_sym
end

#verify_stream ⇒ Object 



373
374
375
376
377
378
379
380
381
382
383
384
 
# File 'lib/tapsoob/data_stream.rb', line 373

def verify_stream
  key = primary_key
  ds = table.order(*order_by)
  current_filter = ds.max(key.sql_number)

  # set the current filter to the max of the primary key
  state[:filter] = current_filter
  # clear out the last_fetched value so it can restart from scratch
  state[:last_fetched] = nil

  log.debug "DataStreamKeyed#verify_stream -> state: #{state.inspect}"
end