Module: PartitionGardener::Migration::HotSwitchConcern

Defined in:
lib/partition_gardener/migration/hot_switch_concern.rb

Overview

Hot-switch from a live non-partitioned table to a declarative partitioned shadow table.

Recommended at switch time:

  • create default + minimal premake (months_ahead: 1)
  • run nightly PartitionGardener maintenance after cutover for sliding-window layout

partition_config may be an inline hash or resolved from Registry via PartitionGardener::Registry.hot_switch_partition_config.

Defined Under Namespace

Modules: ClassMethods

Constant Summary collapse

DEFAULT_SWAP_LOCK_TIMEOUT = 
"5s"

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(base) ⇒ Object 



14
15
16
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 14

def self.included(base)
  base.extend(ClassMethods)
end

Instance Method Details

#add_write_block_trigger(table_name) ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 74

def add_write_block_trigger(table_name)
  trigger_name = "#{table_name}_write_block_trigger"
  function_name = "#{table_name}_write_block_function"

  sql = <<~SQL
    CREATE OR REPLACE FUNCTION #{function_name}()
    RETURNS TRIGGER AS $$
    BEGIN
      RAISE EXCEPTION 'Table % is read-only. Write operations are blocked during hot switch.', '#{table_name}';
    END;
    $$ LANGUAGE plpgsql;
  SQL
  execute(sql)

  execute "DROP TRIGGER IF EXISTS #{trigger_name} ON #{quoted_table(table_name)}"
  sql = <<~SQL
    CREATE TRIGGER #{trigger_name}
    BEFORE INSERT OR UPDATE OR DELETE ON #{quoted_table(table_name)}
    FOR EACH ROW
    EXECUTE FUNCTION #{function_name}();
  SQL
  execute(sql)

  say "Added write-block trigger to #{table_name}"
end

#analyze_shadow_partitions! ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 59

def analyze_shadow_partitions!
  config = hot_switch_config
  partitioned_table = config[:partitioned_table]

  say "Analyzing shadow partitions for #{partitioned_table}"

  fetch_partitions(partitioned_table).each do |partition|
    execute "ANALYZE #{partition}"
    say "Analyzed #{partition}"
  end

  execute "ANALYZE #{quoted_table(partitioned_table)}"
  say "Analyzed #{partitioned_table}"
end

#compare_table_counts ⇒ Object 



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 265

def compare_table_counts
  config = hot_switch_config
  current_table = config[:current_table]
  partitioned_table = config[:partitioned_table]

  current_count = execute("SELECT COUNT(*) AS count FROM #{quoted_table(current_table)}").first["count"].to_i
  partitioned_count = execute("SELECT COUNT(*) AS count FROM #{quoted_table(partitioned_table)}").first["count"].to_i

  say "Count comparison:"
  say "  #{current_table}: #{current_count}"
  say "  #{partitioned_table}: #{partitioned_count}"
  say "  Difference: #{current_count - partitioned_count}"

  {
    current: current_count,
    partitioned: partitioned_count,
    difference: current_count - partitioned_count
  }
end

#ensure_future_partitions_exist(months_ahead: 1) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 24

def ensure_future_partitions_exist(months_ahead: 1)
  config = hot_switch_config
  partition_config = resolve_partition_config(config)
  return unless partition_config

  if months_ahead > 3
    say "Warning: months_ahead=#{months_ahead} creates many partitions at switch time; prefer 1 and let PartitionGardener maintain the sliding window"
  end

  today = PartitionGardener.configuration.today
  partitions_to_create = partition_config[:partitions_to_create].call(today)
  additional_months = (1..months_ahead).map { |month_offset| DateCalendar.add_months(today, month_offset) }
  all_months = (partitions_to_create + additional_months).uniq.sort

  partition_name_format = partition_config[:partition_name_format]
  partition_definition = partition_config[:partition_definition]
  partitioned_table = config[:partitioned_table]
  current_table = config[:current_table]

  all_months.each do |identifier|
    partition_name = partition_name_format.call(identifier).gsub(/^#{current_table}_/, "#{partitioned_table}_")
    next if connection.table_exists?(partition_name)

    for_values_clause = partition_definition.call(identifier)
    sql = <<~SQL
      CREATE TABLE IF NOT EXISTS #{quoted_table(partition_name)} PARTITION OF #{quoted_table(partitioned_table)}
      FOR VALUES #{for_values_clause}
    SQL
    execute(sql)
    say "Created partition #{partition_name}"
  end

  say "Ensured #{all_months.size} partitions exist (including #{months_ahead} months ahead)"
end

#fetch_partitions(table_name) ⇒ Object 



285
286
287
288
289
290
291
292
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 285

def fetch_partitions(table_name)
  sql = <<~SQL
    SELECT inhrelid::regclass AS child
    FROM pg_catalog.pg_inherits
    WHERE inhparent = #{connection.quote(table_name)}::regclass
  SQL
  execute(sql).map { |row| row["child"] }
end

#hot_switch_tables ⇒ Object 



294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 294

def hot_switch_tables
  config = hot_switch_config
  current_table = config[:current_table]
  partitioned_table = config[:partitioned_table]
  old_table = "#{current_table}_old"

  return if connection.table_exists?(old_table)

  say "Performing hot switch: #{current_table} -> #{partitioned_table}"

  sequence_pairs = serial_sequence_pairs(current_table)

  transaction do
    apply_swap_lock_timeout!

    execute "ALTER TABLE #{quoted_table(current_table)} RENAME TO #{quoted_table(old_table)}"
    say "Renamed #{current_table} to #{old_table}"

    execute "ALTER TABLE #{quoted_table(partitioned_table)} RENAME TO #{quoted_table(current_table)}"
    say "Renamed #{partitioned_table} to #{current_table}"

    rename_partition_children!(current_table, partitioned_table, current_table)
    repoint_serial_sequences!(current_table, sequence_pairs)
    remove_write_block_trigger(old_table, current_table)
  end

  say "Hot switch completed successfully"
end

#hot_unswitch_tables ⇒ Object 



323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 323

def hot_unswitch_tables
  config = hot_switch_config
  current_table = config[:current_table]
  partitioned_table = config[:partitioned_table]
  old_table = "#{current_table}_old"

  return unless connection.table_exists?(old_table)
  return if connection.table_exists?(partitioned_table)

  say "Performing hot unswitch: #{current_table} -> #{partitioned_table}"

  sequence_pairs = serial_sequence_pairs(current_table)

  transaction do
    apply_swap_lock_timeout!

    execute "ALTER TABLE #{quoted_table(current_table)} RENAME TO #{quoted_table(partitioned_table)}"
    say "Renamed #{current_table} to #{partitioned_table}"

    execute "ALTER TABLE #{quoted_table(old_table)} RENAME TO #{quoted_table(current_table)}"
    say "Renamed #{old_table} to #{current_table}"

    rename_partition_children!(partitioned_table, current_table, partitioned_table)
    repoint_serial_sequences!(current_table, sequence_pairs)
  end

  say "Hot unswitch completed successfully"
end

#remove_write_block_trigger(table_name, initial_table_name) ⇒ Object 



100
101
102
103
104
105
106
107
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 100

def remove_write_block_trigger(table_name, initial_table_name)
  trigger_name = "#{initial_table_name}_write_block_trigger"
  function_name = "#{initial_table_name}_write_block_function"

  execute "DROP TRIGGER IF EXISTS #{trigger_name} ON #{quoted_table(table_name)}"
  execute "DROP FUNCTION IF EXISTS #{function_name}()"
  say "Removed write-block trigger from #{table_name}"
end

#serial_sequence_pairs(table_name) ⇒ Object 



352
353
354
355
356
357
358
359
360
361
362
363
364
365
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 352

def serial_sequence_pairs(table_name)
  qualified_table = qualified_table_name(table_name)

  table_columns(table_name).filter_map do |column_name|
    sql = <<~SQL
      SELECT pg_get_serial_sequence(#{connection.quote(qualified_table)}, #{connection.quote(column_name)}) AS sequence_name
    SQL
    result = execute(sql).first
    sequence_name = result["sequence_name"]
    next if Blank.blank?(sequence_name)

    [column_name, sequence_name]
  end
end

#sync_delta_data(batch_size: nil, source_table: nil, target_table: nil, swapped: false, sleep_seconds: 0) ⇒ Object 



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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
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
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 143

def sync_delta_data(batch_size: nil, source_table: nil, target_table: nil, swapped: false, sleep_seconds: 0)
  config = hot_switch_config
  batch_size = batch_size || config[:sync_batch_size] || PartitionGardener::MOVE_BATCH_SIZE
  source_table, target_table = resolve_sync_tables(source_table: source_table, target_table: target_table, swapped: swapped)
  partition_key_column = config[:partition_key_column]
  conflict_key = config[:conflict_key] || default_conflict_key(partition_key_column)
  derived_columns = config[:derived_columns] || {}
  stale_column = config[:sync_stale_column] || "updated_at"
  base_key = partition_key_column.to_s.split("::").first.strip
  current_partition_key_expression = config[:current_partition_key_expression] || "s.#{connection.quote_column_name(base_key)}"
  today = PartitionGardener.configuration.today
  start_date = DateCalendar.add_months(today, -3)
  end_date = DateCalendar.add_months(today, 3)
  start_date_sql = connection.quote(start_date)
  end_date_sql = connection.quote(end_date)

  say "Syncing delta data from #{source_table} to #{target_table}"

  current_columns = table_columns(source_table)
  partitioned_columns = table_columns(target_table)
  common_columns = current_columns & partitioned_columns
  insert_columns = (common_columns + derived_columns.keys) & partitioned_columns
  columns_str = insert_columns.map { |column| connection.quote_column_name(column) }.join(", ")
  inner_select_columns_str = insert_columns.map do |column|
    if derived_columns.key?(column)
      "#{derived_columns[column]} AS #{connection.quote_column_name(column)}"
    else
      "s.#{connection.quote_column_name(column)}"
    end
  end.join(", ")

  conflict_key_str = conflict_key.map { |column| connection.quote_column_name(column) }.join(", ")
  conflict_match = conflict_key.map { |column| "d.#{connection.quote_column_name(column)} = s.#{connection.quote_column_name(column)}" }.join(" AND ")
  update_columns = insert_columns - conflict_key
  update_clause = update_columns.map { |column| "#{connection.quote_column_name(column)} = EXCLUDED.#{connection.quote_column_name(column)}" }.join(", ")
  stale_check = if partitioned_columns.include?(stale_column)
    "d.#{connection.quote_column_name(stale_column)} >= s.#{connection.quote_column_name(stale_column)}"
  else
    "FALSE"
  end
  window_predicate = [
    "#{current_partition_key_expression} >= #{start_date_sql}",
    "AND #{current_partition_key_expression} < #{end_date_sql}",
    "AND NOT EXISTS (",
    "SELECT 1 FROM #{quoted_table(target_table)} d",
    "WHERE #{conflict_match}",
    "AND #{stale_check}",
    ")"
  ].join(" ")
  order_clause = conflict_key.map { |column| "s.#{connection.quote_column_name(column)}" }.join(", ")

  count_sql = <<~SQL
    SELECT COUNT(*) AS count
    FROM #{quoted_table(source_table)} s
    WHERE #{window_predicate}
  SQL

  records_to_sync = execute(count_sql).first["count"].to_i
  if records_to_sync.zero?
    say "No records to sync - tables are already in sync"
    return
  end

  say "Found #{records_to_sync} records to sync"

  conflict_update_sql = if update_columns.empty?
    <<~SQL.chomp
      ON CONFLICT (#{conflict_key_str}) DO NOTHING
    SQL
  elsif partitioned_columns.include?(stale_column)
    <<~SQL.chomp
      ON CONFLICT (#{conflict_key_str}) DO UPDATE SET
        #{update_clause}
      WHERE #{quoted_table(target_table)}.#{connection.quote_column_name(stale_column)} < EXCLUDED.#{connection.quote_column_name(stale_column)}
    SQL
  else
    <<~SQL.chomp
      ON CONFLICT (#{conflict_key_str}) DO UPDATE SET
        #{update_clause}
    SQL
  end

  synced_total = 0
  loop do
    sync_sql = <<~SQL
      INSERT INTO #{quoted_table(target_table)} (#{columns_str})
      SELECT #{columns_str}
      FROM (
        SELECT #{inner_select_columns_str}
        FROM #{quoted_table(source_table)} s
        WHERE #{window_predicate}
        ORDER BY #{order_clause}
        LIMIT #{batch_size.to_i}
      ) batch
      #{conflict_update_sql}
    SQL

    result = execute(sync_sql)
    synced_batch = result.cmd_tuples
    break if synced_batch.zero?

    synced_total += synced_batch
    sleep(sleep_seconds) if sleep_seconds.positive?
  end

  say "Synced #{synced_total} records"
end

#table_columns(table_name) ⇒ Object Also known as: get_table_columns 



251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 251

def table_columns(table_name)
  sql = <<~SQL
    SELECT column_name
    FROM information_schema.columns
    WHERE table_schema = #{connection.quote(PartitionGardener.configuration.schema_name)}
      AND table_name = #{connection.quote(table_name)}
    ORDER BY ordinal_position
  SQL

  execute(sql).map { |row| row["column_name"] }
end

#wait_for_active_transactions(table_name, timeout_seconds: 300, check_interval_seconds: 1) ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/partition_gardener/migration/hot_switch_concern.rb', line 109

def wait_for_active_transactions(table_name, timeout_seconds: 300, check_interval_seconds: 1)
  start_time = Time.current
  say "Waiting for active transactions on #{table_name} to complete..."

  loop do
    active_transactions_sql = <<~SQL
      SELECT COUNT(*) AS count
      FROM pg_stat_activity
      WHERE state IN ('active', 'idle in transaction')
        AND pid != pg_backend_pid()
        AND query NOT LIKE '%pg_stat_activity%'
        AND query NOT LIKE '%information_schema%'
        AND (
          query ILIKE '%#{table_name}%'
          OR query ILIKE '%LOCK TABLE #{table_name}%'
        )
    SQL

    active_count = execute(active_transactions_sql).first["count"].to_i
    if active_count.zero?
      say "No active transactions on #{table_name}"
      return true
    end

    elapsed = Time.current - start_time
    if elapsed > timeout_seconds
      raise "Timeout waiting for active transactions on #{table_name}. #{active_count} transactions still active."
    end

    say "Waiting... #{active_count} active transactions on #{table_name} (elapsed: #{elapsed.round}s)"
    sleep(check_interval_seconds)
  end
end