Class: ActiveSanction::Parsers::Join

Inherits:
Object
  • Object
show all
Extended by:
T::Sig
Defined in:
lib/active_sanction/parsers/join.rb

Overview

Joins one primary file to any number of child files on a shared column.

Sanctions publishers routinely split one logical record across several files. OFAC is the extreme case: an entity's name is in SDN.CSV, its aliases are in ALT.CSV and its addresses are in ADD.CSV, and all three are keyed on ent_num. None of the three means anything alone.

join = ActiveSanction::Parsers::Join.new(
on: :ent_num, aliases: ALT.read(raw[:alt]), addresses: ADD.read(raw[:add])
)

join.each(SDN.read(raw[:sdn])) do |row, related|
related[:aliases]     # => [Row, ...] -- always an Array, never nil
related[:addresses]   # => [Row, ...]
end

What streams and what does not

The primary file streams: one row at a time, and the caller decides what to keep. The child files are indexed, which means they are held in memory for the length of the join.

That asymmetry is not a shortcut, it is the only honest option. A join can stream both sides only if both are sorted on the key, and a publisher's sort order is not something to bet a parse on -- OFAC's files happen to arrive sorted today, and nothing says they will next quarter. So the smaller side is indexed and the larger side streams: for OFAC that is ~45k child rows resident while 19,321 primary rows pass through, which is a few tens of megabytes and entirely affordable. A source whose child files are genuinely too large for that wants a different strategy, and should say so rather than discovering it here.

Orphans

A child row whose key matches no primary row is dropped and counted. It is worth counting: a nonzero orphan count after a sync usually means the three files were downloaded at different moments and do not describe the same version of the list, which is a data problem no amount of careful parsing fixes.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(on:, **children) ⇒ void

Parameters:

  • on (T.untyped)
  • children (T.untyped)

Raises:



70
71
72
73
74
75
76
77
 
# File 'lib/active_sanction/parsers/join.rb', line 70

def initialize(on:, **children)
  raise InvalidArgument, "a join needs at least one child reader" if children.empty?

  @on = T.let(on.to_sym, Symbol)
  @children = T.let(children, T::Hash[Symbol, T.untyped])
  @orphans = T.let({}, T::Hash[Symbol, Integer])
  @warnings = T.let([], T::Array[Warning])
end

Instance Attribute Details

#children ⇒ Hash{Symbol => T.untyped} (readonly)

The child readers, by the name the caller gave each one; that name is what #each yields them back under.

Returns:

  • (Hash{Symbol => T.untyped}) 


61
62
63
 
# File 'lib/active_sanction/parsers/join.rb', line 61

def children
  @children
end

#on ⇒ Symbol (readonly)

warnings gathers every complaint from every file in the join, primary first, and orphans counts the child rows that matched nothing. Both are populated by #each, since that is when the files are actually read.

Returns:

  • (Symbol) 


56
57
58
 
# File 'lib/active_sanction/parsers/join.rb', line 56

def on
  @on
end

#orphans ⇒ Hash{Symbol => Integer} (readonly)

Returns:

  • (Hash{Symbol => Integer}) 


64
65
66
 
# File 'lib/active_sanction/parsers/join.rb', line 64

def orphans
  @orphans
end

#warnings ⇒ Array<Warning> (readonly)

Returns:



67
68
69
 
# File 'lib/active_sanction/parsers/join.rb', line 67

def warnings
  @warnings
end

Instance Method Details

#each(primary, &block) ⇒ T.untyped

Yields each primary row with its related child rows. Returns an Enumerator without a block, so join.each(rows).lazy works.

Re-running rebuilds the indexes rather than reusing them, because the readers reset their own warnings on re-enumeration and a join that kept a stale index would report a first pass's problems against a second pass's rows.

Parameters:

  • primary (T.untyped)
  • block (T.untyped)

Returns:

  • (T.untyped) 


87
88
89
90
91
92
93
94
95
96
 
# File 'lib/active_sanction/parsers/join.rb', line 87

def each(primary, &block)
  return enum_for(:each, primary) unless block

  indexes = build_indexes
  matched = Hash.new { |hash, name| hash[name] = Set.new }
  primary.each { |row| block.call(row, related(indexes, matched, row.fetch(on))) }
  @warnings = collect_warnings(primary)
  count_orphans(indexes, matched)
  self
end

#inspect ⇒ String

Returns:

  • (String) 


99
 
# File 'lib/active_sanction/parsers/join.rb', line 99

def inspect = "#<#{self.class} on=#{on.inspect} children=#{children.keys.join(", ")}>"