Class: Exwiw::BatchedExtraction

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/exwiw/batched_extraction.rb

Overview

Extracts a table configured with batch_scope as one query per slice of the scope's id set, so each query stays index-driven instead of degrading into a full scan. Rows are the unbatched query's, in batch order; the slices partition the id set, so none is dropped or repeated. See README.md.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(adapter:, table:, dump_target:, table_by_name:, logger:) ⇒ BatchedExtraction

Returns a new instance of BatchedExtraction.



25
26
27
28
29
30
31
32
33
34
 
# File 'lib/exwiw/batched_extraction.rb', line 25

def initialize(adapter:, table:, dump_target:, table_by_name:, logger:)
  @adapter = adapter
  @table = table
  @dump_target = dump_target
  @table_by_name = table_by_name
  @logger = logger
  @terminus = QueryAstBuilder
    .new(table.name, table_by_name, dump_target, logger)
    .batch_scope_terminus!
end

Instance Attribute Details

#terminus ⇒ Object (readonly)

Returns the value of attribute terminus.



11
12
13
 
# File 'lib/exwiw/batched_extraction.rb', line 11

def terminus
  @terminus
end

Class Method Details

.build(adapter:, table:, dump_target:, table_by_name:, logger:) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
 
# File 'lib/exwiw/batched_extraction.rb', line 13

def self.build(adapter:, table:, dump_target:, table_by_name:, logger:)
  return nil unless table.respond_to?(:batch_scope) && table.batch_scope

  new(
    adapter: adapter,
    table: table,
    dump_target: dump_target,
    table_by_name: table_by_name,
    logger: logger,
  )
end

Instance Method Details

#batch_count ⇒ Object 



81
82
83
 
# File 'lib/exwiw/batched_extraction.rb', line 81

def batch_count
  (key_ids.size + batch_size - 1) / batch_size
end

#batch_query_ast(ids) ⇒ Object 



56
57
58
 
# File 'lib/exwiw/batched_extraction.rb', line 56

def batch_query_ast(ids)
  QueryAstBuilder.run(@table.name, @table_by_name, @dump_target, @logger, batch_ids: ids)
end

#batch_size ⇒ Object 



36
37
38
 
# File 'lib/exwiw/batched_extraction.rb', line 36

def batch_size
  @table.batch_scope.batch_size
end

#describe_plan ⇒ Object 



109
110
111
112
113
114
 
# File 'lib/exwiw/batched_extraction.rb', line 109

def describe_plan
  "-- batch_scope: extracted in batches of up to #{batch_size} #{@terminus.name}." \
    "#{@terminus.primary_key} value(s). Each batch runs the query above with " \
    "`#{@terminus.name}.#{@terminus.primary_key} IN (<batch ids>)` in place of the scope filter, " \
    "over the ids of:"
end

#each ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/exwiw/batched_extraction.rb', line 85

def each
  return enum_for(:each) unless block_given?

  extracted = 0
  key_ids.each_slice(batch_size).with_index do |ids, idx|
    rows = 0
    @adapter.execute(batch_query_ast(ids)).each do |row|
      rows += 1
      yield row
    end
    extracted += rows
    @logger.info("  Batch #{idx + 1}/#{batch_count}: #{rows} record(s), #{extracted} so far.")
  end

  self
end

#key_ids ⇒ Object

Drained in full (the connection must be free for the batch queries) and sorted here rather than via ORDER BY, which would push a sort onto the source DB. Any total order makes the batches reproducible run to run.



77
78
79
 
# File 'lib/exwiw/batched_extraction.rb', line 77

def key_ids
  @key_ids ||= @adapter.execute(key_query_ast).map(&:first).sort
end

#key_query_ast ⇒ Object

The batch table's own extraction query projected to its primary key, so the ids are narrowed by exactly the filter the unbatched query would carry. The key is a plain column so masking configured on it cannot corrupt the ids.



43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/exwiw/batched_extraction.rb', line 43

def key_query_ast
  @key_query_ast ||= begin
    scoped = QueryAstBuilder.run(@terminus.name, @table_by_name, @dump_target, @logger)

    QueryAst::Select.new.tap do |ast|
      ast.from(scoped.from_table_name)
      ast.select([TableColumn.from_symbol_keys(name: @terminus.primary_key)])
      scoped.join_clauses.each { |join_clause| ast.join(join_clause) }
      scoped.where_clauses.each { |where_clause| ast.where(where_clause) }
    end
  end
end

#prepare! ⇒ Object

Resolve the id set and log the plan before extraction starts, so its cost (and an empty id set) is reported where it happens rather than mid-stream.



62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/exwiw/batched_extraction.rb', line 62

def prepare!
  if key_ids.empty?
    @logger.info("  No in-scope #{@terminus.name} ids to batch by; extracting nothing.")
  else
    @logger.info(
      "  Extracting in #{batch_count} batch(es) of up to #{batch_size} " \
      "#{@terminus.name}.#{@terminus.primary_key} value(s) (#{key_ids.size} in scope)."
    )
  end
  self
end

#size ⇒ Object Also known as: length

Only the COPY output format needs the count up front, and each batch answers it with its own count query, so this stays lazy.



104
105
106
 
# File 'lib/exwiw/batched_extraction.rb', line 104

def size
  @size ||= key_ids.each_slice(batch_size).sum { |ids| @adapter.execute(batch_query_ast(ids)).size }
end