Class: Webhookdb::Replicator::Column

Inherits:
Object
  • Object
show all
Includes:
DBAdapter::ColumnTypes
Defined in:
lib/webhookdb/replicator/column.rb

Defined Under Namespace

Classes: IsomorphicProc

Constant Summary collapse

NOT_IMPLEMENTED = 
->(*) { raise NotImplementedError }
CONV_UNIX_TS =

Convert a Unix timestamp (fractional seconds) to a Datetime.

IsomorphicProc.new(
  ruby: lambda do |i, **_|
    return Time.at(i)
  rescue TypeError
    return nil
  end,
  sql: lambda do |i|
    # We do not have the 'rescue TypeError' behavior here yet.
    # It is a beast to add in because we can't easily check if something is convertable,
    # nor can we easily exception handle without creating a stored function.
    Sequel.function(:to_timestamp, Sequel.cast(i, :double))
  end,
)
CONV_TO_I =

Parse a value as an integer. Remove surrounding quotes.

IsomorphicProc.new(
  ruby: ->(i, **_) { i.nil? ? nil : i.delete_prefix('"').delete_suffix('"').to_i },
  sql: ->(i) { Sequel.cast(i, :integer) },
)
CONV_TO_UTC_DATE =

Given a Datetime, convert it to UTC and truncate to a Date.

IsomorphicProc.new(
  ruby: ->(t, **_) { t&.in_time_zone("UTC")&.to_date },
  sql: lambda do |i|
    ts = Sequel.cast(i, :timestamptz)
    in_utc = Sequel.function(:timezone, "UTC", ts)
    Sequel.cast(in_utc, :date)
  end,
)
CONV_PARSE_TIME =

Parse a value using Time.parse.

IsomorphicProc.new(
  ruby: ->(value, **_) { value.nil? ? nil : Time.parse(value) },
  sql: ->(i) { Sequel.cast(i, :timestamptz) },
)
CONV_PARSE_DATE =

Parse a value using Date.parse.

IsomorphicProc.new(
  ruby: ->(value, **_) { value.nil? ? nil : Date.parse(value) },
  sql: ->(i) { Sequel.cast(i, :date) },
)
CONV_COMMA_SEP = 
IsomorphicProc.new(
  ruby: ->(value, **_) { value.nil? ? [] : value.split(",").map(&:strip) },
  sql: lambda do |_e, json_path:, source_col:|
    e = source_col.get_text(json_path)
    parts = Sequel.function(:string_to_array, e, ",")
    parts = Sequel.function(:unnest, parts)
    sel = Webhookdb::Dbutil::MOCK_CONN.
      from(parts.as(:parts)).
      select(Sequel.function(:trim, :parts))
    f = Sequel.function(:array, sel)
    return f
  end,
)
DAYS_OF_WEEK = 
[
  "SUNDAY",
  "MONDAY",
  "TUESDAY",
  "WEDNESDAY",
  "THURSDAY",
  "FRIDAY",
  "SATURDAY",
].freeze
KNOWN_CONVERTERS = 
{
  date: CONV_PARSE_DATE,
  time: CONV_PARSE_TIME,
  to_i: CONV_TO_I,
  tsat: CONV_UNIX_TS,
}.freeze
DEFAULTER_NOW = 
IsomorphicProc.new(ruby: ->(*) { Time.now }, sql: ->(*) { Sequel.function(:now) })
DEFAULTER_FALSE = 
IsomorphicProc.new(ruby: ->(*) { false }, sql: ->(*) { false })
DEFAULTER_UUID4 = 
IsomorphicProc.new(ruby: ->(*) { Uuidx.v4 }, sql: ->(*) { Sequel.function(:gen_random_uuid) })
DEFAULTER_UUID7 = 
IsomorphicProc.new(ruby: ->(*) { Uuidx.v7 }, sql: NOT_IMPLEMENTED)
DEFAULTER_FROM_INTEGRATION_SEQUENCE = 
IsomorphicProc.new(
  ruby: ->(service_integration:, **_) { service_integration.sequence_nextval },
  sql: ->(service_integration:) { Sequel.function(:nextval, service_integration.sequence_name) },
)
KNOWN_DEFAULTERS = 
{
  now: DEFAULTER_NOW,
  tofalse: DEFAULTER_FALSE,
  uuid4: DEFAULTER_UUID4,
  uuid7: DEFAULTER_UUID7,
}.freeze
EACH_ITEM =

Use in data_key when a value is an array, and you want to map a value from the array.

:_each_item

Constants included from DBAdapter::ColumnTypes

DBAdapter::ColumnTypes::BIGINT, DBAdapter::ColumnTypes::BIGINT_ARRAY, DBAdapter::ColumnTypes::BOOLEAN, DBAdapter::ColumnTypes::COLUMN_TYPES, DBAdapter::ColumnTypes::DATE, DBAdapter::ColumnTypes::DECIMAL, DBAdapter::ColumnTypes::DOUBLE, DBAdapter::ColumnTypes::FLOAT, DBAdapter::ColumnTypes::INTEGER, DBAdapter::ColumnTypes::INTEGER_ARRAY, DBAdapter::ColumnTypes::OBJECT, DBAdapter::ColumnTypes::TEXT, DBAdapter::ColumnTypes::TEXT_ARRAY, DBAdapter::ColumnTypes::TIMESTAMP, DBAdapter::ColumnTypes::UUID

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, type, data_key: nil, event_key: nil, from_enrichment: false, optional: false, converter: nil, defaulter: nil, index: false, index_not_null: false, skip_nil: false, backfill_statement: nil, backfill_expr: nil) ⇒ Column

Returns a new instance of Column.

Raises:

  • (ArgumentError) 


355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
 
# File 'lib/webhookdb/replicator/column.rb', line 355

def initialize(
  name,
  type,
  data_key: nil,
  event_key: nil,
  from_enrichment: false,
  optional: false,
  converter: nil,
  defaulter: nil,
  index: false,
  index_not_null: false,
  skip_nil: false,
  backfill_statement: nil,
  backfill_expr: nil
)
  raise ArgumentError, "name must be a symbol" unless name.is_a?(Symbol)
  raise ArgumentError, "type #{type.inspect} is not supported" unless COLUMN_TYPES.include?(type)
  raise ArgumentError, "use :tofalse as the defaulter (or nil for no defaulter)" if defaulter == false
  @name = name
  @type = type
  @data_key = data_key || name.to_s
  @event_key = event_key
  @from_enrichment = from_enrichment
  @optional = optional
  @converter = KNOWN_CONVERTERS[converter] || converter
  @defaulter = KNOWN_DEFAULTERS[defaulter] || defaulter
  @index = index
  @index_not_null = index_not_null
  @skip_nil = skip_nil
  @backfill_statement = backfill_statement
  @backfill_expr = backfill_expr
end

Instance Attribute Details

#backfill_expr ⇒ String, ... (readonly)

If provided, use this expression as the UPDATE value when adding the column to an existing table.

Returns:

  • (String, Sequel, Sequel::SQL::Expression) 


353
354
355
 
# File 'lib/webhookdb/replicator/column.rb', line 353

def backfill_expr
  @backfill_expr
end

#backfill_statement ⇒ Object (readonly)

If provided, run this before backfilling as part of UPDATE. Usually used to add functions into pg_temp schema. This is an advanced use case; see unit tests for examples.



348
349
350
 
# File 'lib/webhookdb/replicator/column.rb', line 348

def backfill_statement
  @backfill_statement
end

#converter ⇒ IsomorphicProc (readonly)

Sometimes we need to do some processing on the value provided by the external service so that the we get the data we want in the format we want. A common example is parsing various DateTime formats into our desired timestamp format. In these cases, we use a ‘converter`, which is an `IsomorphicProc` where both procs take the value retrieved from the external service and the resource object and return a value consistent with the column’s type attribute.

Returns:

  • (IsomorphicProc) —

    The ‘ruby’ proc accepts (value, resource:, event:, enrichment:, service_integration:) and returns a value. The ‘sql’ proc takes an expression and returns a new expression.



327
328
329
 
# File 'lib/webhookdb/replicator/column.rb', line 327

def converter
  @converter
end

#data_key ⇒ String+ (readonly)

‘data_key` is the key we look for in the resource object. If this value is an array we will `_dig` through the object using each key successively. `data_key` defaults to the string version of whatever name you provide for the column.

Returns:

  • (String, Array<String>) 


296
297
298
 
# File 'lib/webhookdb/replicator/column.rb', line 296

def data_key
  @data_key
end

#defaulter ⇒ IsomorphicProc (readonly)

If the value we retrieve from the data provided by the external service is nil, we often want to use a default value instead of nil. The ‘defaulter` is an `IsomorphicProc` where both procs take the resource object and return a default value that is used in the upsert. A common example is the `now` defaulter, which uses the current time as the default value.

Returns:

  • (IsomorphicProc) —

    The ‘ruby’ proc accepts (resource:, event:, enrichment:, service_integration:) and returns a value. The ‘sql’ proc accepts (service_integration:) and returns an sql expression.



337
338
339
 
# File 'lib/webhookdb/replicator/column.rb', line 337

def defaulter
  @defaulter
end

#event_key ⇒ String+ (readonly)

‘event_key` is the key we look for in the event object. This defaults to nil, but note that if both an event object and event key are provided, we will always grab the value from the event object instead of from the resource object using the `data_key`. If this value is an array we will `_dig` through the object using each key successively, same as with `data_key`.

Returns:

  • (String, Array<String>) 


303
304
305
 
# File 'lib/webhookdb/replicator/column.rb', line 303

def event_key
  @event_key
end

#from_enrichment ⇒ Boolean (readonly)

If ‘from_enrichment` is set then we use the `data_key` value to find the desired value in the enrichment object. In this case, if the enrichment object is nil you will get an error.

Returns:

  • (Boolean) 


308
309
310
 
# File 'lib/webhookdb/replicator/column.rb', line 308

def from_enrichment
  @from_enrichment
end

#index ⇒ Boolean (readonly) Also known as: index?

Returns:

  • (Boolean) 


280
281
282
 
# File 'lib/webhookdb/replicator/column.rb', line 280

def index
  @index
end

#index_not_null ⇒ Boolean (readonly)

True if thie index should be a partial index, using WHERE (col IS NOT NULL). The #index attribute must be true.

Returns:

  • (Boolean) 


286
287
288
 
# File 'lib/webhookdb/replicator/column.rb', line 286

def index_not_null
  @index_not_null
end

#name ⇒ Symbol (readonly)

Returns:

  • (Symbol) 


276
277
278
 
# File 'lib/webhookdb/replicator/column.rb', line 276

def name
  @name
end

#optional ⇒ Boolean (readonly)

If ‘optional` is true then the column will be populated with a nil value instead of throwing an error if the desired value is not present in the object you’re ‘_dig`ging into, which could be any of the three (resource, event, and enrichment) according to the way the rest of the attributes are configured. Note that for nested values, `_dig` will return nil if any of the keys in the provided array are missing from the object.

Returns:

  • (Boolean) 


316
317
318
 
# File 'lib/webhookdb/replicator/column.rb', line 316

def optional
  @optional
end

#skip_nil ⇒ Boolean (readonly) Also known as: skip_nil?

If ‘skip_nil` is set to true, we only add the described value to the hash that gets upserted if it is not nil. This is so that we don’t override existing data in the database row with a nil value.

Returns:

  • (Boolean) 


342
343
344
 
# File 'lib/webhookdb/replicator/column.rb', line 342

def skip_nil
  @skip_nil
end

#type ⇒ Symbol (readonly)

Returns:

  • (Symbol) 


278
279
280
 
# File 'lib/webhookdb/replicator/column.rb', line 278

def type
  @type
end

Class Method Details

._assert_regex_converter_type(re) ⇒ Object

Raises:

  • (ArgumentError) 


486
487
488
489
 
# File 'lib/webhookdb/replicator/column.rb', line 486

def self._assert_regex_converter_type(re)
  return Regexp.new(re) if re.is_a?(String)
  raise ArgumentError, "regexp must be a string, not a Ruby regex, so it can be used in the database verbatim"
end

.converter_array_element(index:, sep:, cls: DECIMAL) ⇒ Object 



165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/webhookdb/replicator/column.rb', line 165

def self.converter_array_element(index:, sep:, cls: DECIMAL)
  case cls
    when DECIMAL
      to_ruby = ->(v) { BigDecimal(v) }
      to_sql = ->(e) { Sequel.cast(e, :decimal) }
    else
      raise ArgumentError, "Unsupported cls" unless valid_cls.include?(cls)
  end

  return IsomorphicProc.new(
    ruby: lambda do |value, **|
      break nil if value.nil?
      parts = value.split(sep)
      break nil if index >= parts.size
      to_ruby.call(parts[index])
    end,
    sql: lambda do |expr|
      # The expression may be a JSONB field, of the type jsonb (accessed with -> rather than ->>).
      # Make sure it's text. The CAST will turn 'a' into '"a"' though, so we also need to trim quotes.
      str_expr = Sequel.cast(expr, :text)
      str_expr = Sequel.function(:btrim, str_expr, '"')
      field_expr = Sequel.function(:split_part, str_expr, sep, index + 1)
      # If the field is invalid, we get ''. Use nil in this case.
      case_expr = Sequel.case({Sequel[field_expr => ""] => nil}, field_expr)
      to_sql.call(case_expr)
    end,
  )
end

.converter_array_pluck(key, coltype) ⇒ Object 



194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
 
# File 'lib/webhookdb/replicator/column.rb', line 194

def self.converter_array_pluck(key, coltype)
  pgtype = Webhookdb::DBAdapter::PG::COLTYPE_MAP.fetch(coltype)
  return IsomorphicProc.new(
    ruby: lambda do |value, **|
      break nil if value.nil?
      break nil unless value.respond_to?(:to_ary)
      value.map { |v| v[key] }
    end,
    sql: lambda do |expr|
      expr = Sequel.lit("'#{JSON.generate(expr)}'::jsonb") if expr.is_a?(Hash) || expr.is_a?(Array)
      Webhookdb::Dbutil::MOCK_CONN.
        from(Sequel.function(:jsonb_to_recordset, expr).as(Sequel.lit("x(#{key} #{pgtype})"))).
        select(Sequel.function(:array_agg, Sequel.lit(key)))
    end,
  )
end

.converter_from_regex(pattern, dbtype: nil) ⇒ Object

Note:

Only the first capture group can be extracted at this time.

Return a converter that parses a value using the given regex, and returns the capture group at index. The ‘coerce’ function can be applied to, for example, capture a number from a request path and store it as an integer.

Parameters:

  • pattern (String)
  • dbtype (Symbol) (defaults to: nil) —

    The DB type to use, like INTEGER or BIGINT.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/webhookdb/replicator/column.rb', line 78

def self.converter_from_regex(pattern, dbtype: nil)
  re = self._assert_regex_converter_type(pattern)
  case dbtype
    when INTEGER
      rcoerce = :to_i
      pgcast = :integer
    when BIGINT
      rcoerce = :to_i
      pgcast = :bigint
    when nil
      rcoerce = nil
      pgcast = nil
    else
      raise NotImplementedError, "unhandled converter_from_regex dbtype: #{dbtype}"
  end
  return IsomorphicProc.new(
    ruby: lambda do |value, **_|
      matched = value&.match(re) do |md|
        md.captures ? md.captures[0] : nil
      end
      (matched = matched.send(rcoerce)) if !matched.nil? && rcoerce
      matched
    end,
    sql: lambda do |e|
      f = Sequel.function(:substring, e.cast(:text), pattern)
      f = f.cast(pgcast) if pgcast
      f
    end,
  )
end

.converter_gsub(pattern, replacement) ⇒ Object 



153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/webhookdb/replicator/column.rb', line 153

def self.converter_gsub(pattern, replacement)
  re = self._assert_regex_converter_type(pattern)
  return Webhookdb::Replicator::Column::IsomorphicProc.new(
    ruby: lambda do |value, **|
      value&.gsub(re, replacement)
    end,
    sql: lambda do |e|
      Sequel.function(:regexp_replace, e, pattern, replacement, "g")
    end,
  )
end

.converter_int_or_sequence_from_regex(re, dbtype: BIGINT) ⇒ Object

Note:

This converter does not work for backfilling/UPDATE of existing columns.

Extract a number from a string using the given regexp. If nothing can be extracted, get the next value from the sequence.

Note this requires ‘requires_sequence=true` on the replicator.

Used primarily where the ID is sent by an API only in the request URL (not a key in the body), and the URL will not include an ID when it’s being sent for the first time. We see this in channel manager APIs primarily, that replicate their data to 3rd parties.

It is generally only of use for unique ids.



120
121
122
123
124
125
126
127
128
129
 
# File 'lib/webhookdb/replicator/column.rb', line 120

def self.converter_int_or_sequence_from_regex(re, dbtype: BIGINT)
  return Webhookdb::Replicator::Column::IsomorphicProc.new(
    ruby: lambda do |value, service_integration:, **kw|
      url_id = Webhookdb::Replicator::Column.converter_from_regex(re, dbtype:).
        ruby.call(value, service_integration:, **kw)
      url_id || service_integration.sequence_nextval
    end,
    sql: NOT_IMPLEMENTED,
  )
end

.converter_map_lookup(array:, map:) ⇒ Object

Convert a value or array by looking up its value in a map.

Parameters:

  • array (Boolean) —

    If true, the empty value is an array. If false, nil.

  • map (Hash) 


224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/webhookdb/replicator/column.rb', line 224

def self.converter_map_lookup(array:, map:)
  empty = array ? Sequel.pg_array([]) : nil
  return IsomorphicProc.new(
    ruby: lambda do |value, **|
      break empty if value.nil?
      is_ary = value.respond_to?(:to_ary)
      r = (is_ary ? value : [value]).map do |v|
        if (mapval = map[v])
          mapval
        else
          v
        end
      end
      break is_ary ? r : r[0]
    end,
    sql: NOT_IMPLEMENTED,
  )
end

.converter_strptime(format, sqlformat = nil, cls: Time) ⇒ Object

Parse the value in the column using the given strptime string.

To provide an ‘sql` proc, provide the sqlformat string, which is used in TO_TIMESTAMP(col, sqlformat). Note that TO_TIMESTAMP does not support timezone offsets, so the time will always be in UTC.

Future note: We may want to derive sqlformat from format, and handle timezone offsets in the timestamp strings.



139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/webhookdb/replicator/column.rb', line 139

def self.converter_strptime(format, sqlformat=nil, cls: Time)
  return Webhookdb::Replicator::Column::IsomorphicProc.new(
    ruby: lambda do |value, **|
      value.nil? ? nil : cls.strptime(value, format)
    end,
    sql: lambda do |e|
      raise NotImplementedError if sqlformat.nil?
      f = Sequel.function(:to_timestamp, e, sqlformat)
      f = f.cast(:date) if cls == Date
      f
    end,
  )
end

.defaulter_from_resource_field(key) ⇒ Object 



259
260
261
262
263
264
 
# File 'lib/webhookdb/replicator/column.rb', line 259

def self.defaulter_from_resource_field(key)
  return Webhookdb::Replicator::Column::IsomorphicProc.new(
    ruby: ->(resource:, **_) { resource.fetch(key.to_s) },
    sql: ->(*) { key.to_sym },
  )
end

Instance Method Details

#_dig(h, keys, optional) ⇒ Object 



468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
 
# File 'lib/webhookdb/replicator/column.rb', line 468

def _dig(h, keys, optional)
  v = h
  karr = Array(keys)
  karr.each do |key|
    begin
      v = optional ? v[key] : v.fetch(key)
    rescue KeyError
      raise KeyError, "key not found: '#{key}' in: #{v.keys}"
    rescue NoMethodError => e
      raise NoMethodError, "Element #{key} of #{karr}\n#{e}"
    end
    # allow optional nested values by returning nil as soon as key not found
    # the problem here is that you effectively set all keys in the sequence as optional
    break if optional && v.nil?
  end
  return v
end

#to_dbadapter(**more) ⇒ Object 



388
389
390
391
392
393
 
# File 'lib/webhookdb/replicator/column.rb', line 388

def to_dbadapter(**more)
  kw = {name:, type:, index:, backfill_expr:, backfill_statement:}
  kw[:index_where] = Sequel[self.name] !~ nil if self.index_not_null
  kw.merge!(more)
  return Webhookdb::DBAdapter::Column.new(**kw)
end

#to_ruby_value(resource:, event:, enrichment:, service_integration:) ⇒ Object 



432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
 
# File 'lib/webhookdb/replicator/column.rb', line 432

def to_ruby_value(resource:, event:, enrichment:, service_integration:)
  v = if self.from_enrichment
        self._dig(enrichment, self.data_key, self.optional)
  elsif event && self.event_key
    # Event keys are never optional since any API using them is going to have fixed keys
    self._dig(event, self.event_key, false)
  else
    self._dig(resource, self.data_key, self.optional)
  end
  (v = self.defaulter.ruby.call(resource:, event:, enrichment:, service_integration:)) if self.defaulter && v.nil?
  v = self.converter.ruby.call(v, resource:, event:, enrichment:, service_integration:) if self.converter
  if (self.type == INTEGER_ARRAY) && !v.nil?
    v = Sequel.pg_array(v, "integer")
  elsif (self.type == TEXT_ARRAY) && !v.nil?
    v = Sequel.pg_array(v, "text")
  elsif (self.type == BIGINT_ARRAY) && !v.nil?
    v = Sequel.pg_array(v, "bigint")
  elsif (self.type == TIMESTAMP) && !v.nil?
    # Postgres CANNOT handle timestamps with a 0000 year,
    # even if the actual time it represents is valid (due to timezone offset).
    # Repro with `SELECT '0000-12-31T18:10:00-05:50'::timestamptz`.
    # So if we are in the year 0, represent the time into UTC to get it out of year 0
    # (if it's still invalid, let it error).
    # NOTE: Only worry about times; if the value is a string, it will still error.
    # Let the caller parse the string into a Time to get this special behavior.
    # Time parsing is too loose to do it here.
    v = v.utc if v.is_a?(Time) && v.year.zero?
  end
  # pg_json doesn't handle thie ssuper well in our situation,
  # so JSON must be inserted as a string.
  if (_stringify_json = self.type == OBJECT && !v.nil? && !v.is_a?(String))
    v = v.to_json
  end
  return v
end

#to_sql_expr ⇒ Object

Convert this column to an expression that can be used to return the column’s value based on what is present in the row. This is generally used to ‘backfill’ column values from what is in the data and enrichment columns.

NOTE: this method assumes Postgres as the backing database. To support others will require additional work and some abstraction.



402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
 
# File 'lib/webhookdb/replicator/column.rb', line 402

def to_sql_expr
  source_col = @from_enrichment ? :enrichment : :data
  source_col_expr = Sequel.pg_json(source_col)
  # Have to use string keys here, PG handles it alright though.
  dkey = @data_key.respond_to?(:to_ary) ? @data_key.map(&:to_s) : @data_key
  expr = case self.type
    # If we're pulling out a normal value from JSON, get it as a 'native' value (not jsonb) (ie, ->> op).
    when TIMESTAMP, DATE, TEXT, INTEGER, BIGINT
      source_col_expr.get_text(dkey)
    else
      # If this is a more complex value, get it as jsonb (ie, -> op).
      # Note that this can be changed by the sql converter.
      source_col_expr[Array(dkey)]
  end
  if self.converter
    if self.converter.sql == NOT_IMPLEMENTED
      msg = "Converter SQL for #{self.name} is not implemented. This column cannot be added after the fact, " \
            "backfill_expr should be set on the column to provide a manual UPDATE/backfill converter, " \
            "or the :sql converter can be implemented (may not be possible or feasible in all cases)."
      raise TypeError, msg
    end
    conv_kwargs = self.converter.sql.arity == 1 ? {} : {json_path: dkey, source_col: source_col_expr}
    expr = self.converter.sql.call(expr, **conv_kwargs)
  end
  pgcol = Webhookdb::DBAdapter::PG::COLTYPE_MAP.fetch(self.type)
  expr = expr.cast(pgcol)
  (expr = Sequel.function(:coalesce, expr, self.defaulter.sql.call)) if self.defaulter
  return expr
end