Module: PtOnlineSchemaChange::Helper

Defined in:
app/my_lib/pt_online_schema_change/helper.rb

Constant Summary collapse

TYPE_MAP = 
{
  string: 'VARCHAR(255)',
  text: 'TEXT',
  integer: 'INT',
  bigint: 'BIGINT',
  float: 'FLOAT',
  boolean: 'TINYINT(1)',
  datetime: 'DATETIME',
  timestamp: 'TIMESTAMP',
  date: 'DATE',
  time: 'TIME',
  decimal: 'DECIMAL(10,2)',
  json: 'JSON',
  binary: 'BLOB',
  uuid: 'CHAR(36)',
  enum: 'ENUM',
  set: 'SET',
}.freeze

Instance Method Summary collapse

Instance Method Details

#pt_add_column(table_name, column_name, type, options = {}) ⇒ void

This method returns an undefined value.

Adds a new column to a database table using pt-online-schema-change.

Examples:

Add a non-nullable string column with default value

pt_add_column('users', 'status', :string, null: false, default: 'active', limit: 50)

Parameters:

  • table_name (String) —

    The name of the table to alter

  • column_name (String) —

    The name of the column to add

  • type (Symbol) —

    The data type of the column (e.g., :string, :integer)

  • options (Hash) (defaults to: {}) —

    Additional column options

Options Hash (options):

  • :limit (Integer) —

    For string types, specifies the maximum length

  • :null (Boolean) — default: true —

    If false, adds NOT NULL constraint

  • :default (String) —

    Default value for the column

Raises:

  • (RuntimeError) —

    If an unsupported column type is provided



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 136

def pt_add_column(table_name, column_name, type, options = {})
  type_sql = TYPE_MAP[type] || raise("Unsupported type: #{type}. Supported types are: #{TYPE_MAP.keys.map(&:to_s).sort.join(', ')}")

  if type.to_sym == :string && options[:limit]
    type_sql = "VARCHAR(#{options[:limit]})"
  end

  # Handle column options
  column_options = []
  column_options << 'NOT NULL' if options[:null] == false
  if options.key?(:default)
    default_value = options[:default]
    column_options << if default_value.nil?
                        'DEFAULT NULL'
                      elsif default_value == true
                        'DEFAULT 1'
                      elsif default_value == false
                        'DEFAULT 0'
                      else
                        # Always quote the default value to prevent SQL injection
                        "DEFAULT #{ActiveRecord::Base.connection.quote(default_value)}"
                      end
  end

  alter_statement = "ADD COLUMN #{column_name} #{type_sql}"
  alter_statement += " #{column_options.join(' ')}" unless column_options.empty?

  pt_online_alter_table(table_name, alter_statement)
end

#pt_add_columns(table_name, columns, options = {}) ⇒ void

This method returns an undefined value.

Adds multiple columns to a database table using pt-online-schema-change in a single operation.

Examples:

Add multiple columns

pt_add_columns('users', [
  { name: 'first_name', type: :string, options: { limit: 100, null: false } },
  { name: 'last_name', type: :string, options: { limit: 100, null: false } },
  { name: 'age', type: :integer, options: { default: 0 } }
])

Parameters:

  • table_name (String) —

    The name of the table to alter

  • columns (Array<Hash>) —

    Array of column definitions

  • options (Hash) (defaults to: {}) —

    Additional options to pass to pt_online_alter_table

Options Hash (columns):

  • :name (String) —

    The name of the column

  • :type (Symbol) —

    The data type of the column

  • :options (Hash) —

    Column options (limit, null, default)

Raises:

  • (RuntimeError) —

    If an unsupported column type is provided or batch size exceeds recommendations



300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
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
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 300

def pt_add_columns(table_name, columns, options = {})
  # Always validate batch size against recommendations
  recommendations = pt_get_recommendations(table_name)
  max_columns = recommendations[:max_columns_per_batch]

  if columns.length > max_columns
    error_message = 'Too many columns for single batch operation. ' \
                   "Table '#{table_name}' (#{recommendations[:size_category]}, #{recommendations[:estimated_rows]} rows) " \
                   "supports max #{max_columns} columns per batch, but #{columns.length} provided. " \
                   'Consider splitting into smaller batches.'

    BUSINESS_LOGGER.error('Batch size validation failed for pt_add_columns', {
                            table_name: table_name,
                            columns_count: columns.length,
                            max_allowed: max_columns,
                            size_category: recommendations[:size_category],
                            estimated_rows: recommendations[:estimated_rows],
                            operation: 'pt_add_columns_validation_failed',
                          })

    raise ArgumentError, error_message
  end

  BUSINESS_LOGGER.info('Batch size validation passed for pt_add_columns', {
                         table_name: table_name,
                         columns_count: columns.length,
                         max_allowed: max_columns,
                         size_category: recommendations[:size_category],
                       })

  column_definitions = columns.map do |column|
    column_name = column[:name]
    type = column[:type]
    column_options = column[:options] || {}

    type_sql = TYPE_MAP[type] || raise("Unsupported type: #{type}. Supported types are: #{TYPE_MAP.keys.map(&:to_s).sort.join(', ')}")

    if type.to_sym == :string && column_options[:limit]
      type_sql = "VARCHAR(#{column_options[:limit]})"
    end

    # Handle column options
    column_opts = []
    column_opts << 'NOT NULL' if column_options[:null] == false
    if column_options.key?(:default)
      default_value = column_options[:default]
      column_opts << if default_value.nil?
                       'DEFAULT NULL'
                     elsif default_value == true
                       'DEFAULT 1'
                     elsif default_value == false
                       'DEFAULT 0'
                     else
                       # Always quote the default value to prevent SQL injection
                       "DEFAULT #{ActiveRecord::Base.connection.quote(default_value)}"
                     end
    end

    column_def = "ADD COLUMN #{column_name} #{type_sql}"
    column_def += " #{column_opts.join(' ')}" unless column_opts.empty?
    column_def
  end

  alter_statement = column_definitions.join(', ')
  pt_online_alter_table(table_name, alter_statement, options)
end

#pt_add_index(table_name, column_names, options = {}) ⇒ void

This method returns an undefined value.

Adds an index to a database table using pt-online-schema-change.

Examples:

Add index to email on users with unique index (please make sure there is no duplication data first)

pt_add_index('users', 'email', unique: true, check_unique_key_change: false)

Add index to email on users with custom index name

pt_add_index('users', 'email', name: 'custom_index_name')

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • column_names (Array<String, Symbol>) —

    The names of the columns to include in the index

  • options (Hash) (defaults to: {}) —

    Additional index options

Options Hash (options):

  • :name (String) —

    The name of the index (optional)

  • :unique (Boolean) — default: false —

    If true, creates a unique index

  • :check_unique_key_change (Boolean) — default: true —

    If false, skips unique key change validation (use with caution)

Raises:

  • (RuntimeError) —

    If an error occurs during index creation



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
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 183

def pt_add_index(table_name, column_names, options = {})
  index_name = options[:name] || "idx_#{table_name}_#{Array(column_names).join('_')}"
  columns = Array(column_names).join(', ')

  alter_statement = if options[:unique]
                      "ADD UNIQUE INDEX #{index_name} (#{columns})"
                    else
                      "ADD INDEX #{index_name} (#{columns})"
                    end

  # Extract pt-osc specific options including replication monitoring
  pt_osc_options = {}
  if options.key?(:check_unique_key_change)
    pt_osc_options[:check_unique_key_change] =
      options[:check_unique_key_change]
  end

  # Log warning if bypassing unique key change check
  if options[:unique] && options[:check_unique_key_change] == false
    BUSINESS_LOGGER.warn('Bypassing unique key change validation - potential data loss risk', {
                           table_name: table_name,
                           columns: columns,
                           index_name: index_name,
                           operation: 'pt_add_unique_index_unsafe',
                         })
  end

  pt_online_alter_table(table_name, alter_statement, pt_osc_options)
end

#pt_add_indexes(table_name, indexes, options = {}) ⇒ void

This method returns an undefined value.

Adds multiple indexes to a database table using pt-online-schema-change in a single operation.

Examples:

Add multiple indexes

pt_add_indexes('users', [
  { columns: ['email'], unique: true, name: 'idx_unique_email' },
  { columns: ['created_at', 'status'] },
  { columns: ['last_login_at'] }
])

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • indexes (Array<Hash>) —

    Array of index definitions

  • options (Hash) (defaults to: {}) —

    Additional options to pass to pt_online_alter_table

Options Hash (indexes):

  • :columns (Array<String, Symbol>) —

    The names of the columns to include in the index

  • :name (String) —

    The name of the index (optional)

  • :unique (Boolean) — default: false —

    If true, creates a unique index

Options Hash (options):

  • :check_unique_key_change (Boolean) — default: true —

    If false, skips unique key change validation (use with caution)

Raises:

  • (RuntimeError) —

    If an error occurs during index creation or batch size exceeds recommendations



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
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 436

def pt_add_indexes(table_name, indexes, options = {})
  # Always validate batch size against recommendations
  recommendations = pt_get_recommendations(table_name)
  max_indexes = recommendations[:max_columns_per_batch] # Use same limit for indexes

  if indexes.length > max_indexes
    error_message = 'Too many indexes for single batch operation. ' \
                   "Table '#{table_name}' (#{recommendations[:size_category]}, #{recommendations[:estimated_rows]} rows) " \
                   "supports max #{max_indexes} indexes per batch, but #{indexes.length} provided. " \
                   'Consider splitting into smaller batches.'

    BUSINESS_LOGGER.error('Batch size validation failed for pt_add_indexes', {
                            table_name: table_name,
                            indexes_count: indexes.length,
                            max_allowed: max_indexes,
                            size_category: recommendations[:size_category],
                            estimated_rows: recommendations[:estimated_rows],
                            operation: 'pt_add_indexes_validation_failed',
                          })

    raise ArgumentError, error_message
  end

  BUSINESS_LOGGER.info('Batch size validation passed for pt_add_indexes', {
                         table_name: table_name,
                         indexes_count: indexes.length,
                         max_allowed: max_indexes,
                         size_category: recommendations[:size_category],
                       })

  has_unique_index = indexes.any? { |index| index[:unique] }

  # Log warning if bypassing unique key change check for any unique index
  if has_unique_index && options[:check_unique_key_change] == false
    BUSINESS_LOGGER.warn('Bypassing unique key change validation for multiple indexes - potential data loss risk', {
                           table_name: table_name,
                           indexes: indexes.select { |idx| idx[:unique] }.pluck(:columns),
                           operation: 'pt_add_multiple_indexes_unsafe',
                         })
  end

  index_definitions = indexes.map do |index|
    columns = Array(index[:columns]).join(', ')
    index_name = index[:name] || "idx_#{table_name}_#{Array(index[:columns]).join('_')}"

    if index[:unique]
      "ADD UNIQUE INDEX #{index_name} (#{columns})"
    else
      "ADD INDEX #{index_name} (#{columns})"
    end
  end

  alter_statement = index_definitions.join(', ')

  # Extract pt-osc specific options
  pt_osc_options = {}
  if options.key?(:check_unique_key_change)
    pt_osc_options[:check_unique_key_change] = options[:check_unique_key_change]
  end

  pt_online_alter_table(table_name, alter_statement, pt_osc_options)
end

#pt_add_reference(table_name, column_name, referenced_table, referenced_column, options = {}) ⇒ void

This method returns an undefined value.

Executes a schema change to add a foreign key constraint to a table using pt-online-schema-change.

Examples:

Add a foreign key from users.role_id to roles.id

pt_add_reference('users', 'role_id', 'roles', 'id')

Parameters:

  • table_name (String) —

    The name of the table to add the constraint to

  • column_name (String) —

    The name of the column that will be the foreign key

  • referenced_table (String) —

    The name of the table being referenced

  • referenced_column (String) —

    The name of the column in the referenced table

  • options (Hash) (defaults to: {}) —

    Additional options to pass to pt_online_alter_table

Raises:

  • (RuntimeError) —

    If an error occurs during foreign key creation



259
260
261
262
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 259

def pt_add_reference(table_name, column_name, referenced_table, referenced_column, options = {})
  alter_statement = "ADD CONSTRAINT fk_#{table_name}_#{column_name} FOREIGN KEY (#{column_name}) REFERENCES #{referenced_table}(#{referenced_column})"
  pt_online_alter_table(table_name, alter_statement, options)
end

#pt_drop_column(table_name, column_name) ⇒ void

This method returns an undefined value.

Drops a column from a database table using pt-online-schema-change.

Examples:

Drop a column from the users table

pt_drop_column('users', 'status')

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • column_name (String, Symbol) —

    The name of the column to drop

Raises:

  • (RuntimeError) —

    If an error occurs during column removal



224
225
226
227
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 224

def pt_drop_column(table_name, column_name)
  alter_statement = "DROP COLUMN #{column_name}"
  pt_online_alter_table(table_name, alter_statement)
end

#pt_drop_columns(table_name, column_names, options = {}) ⇒ void

This method returns an undefined value.

Drops multiple columns from a database table using pt-online-schema-change in a single operation.

Examples:

Drop multiple columns from the users table

pt_drop_columns('users', ['status', 'deprecated_field', 'old_column'])

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • column_names (Array<String, Symbol>) —

    The names of the columns to drop

  • options (Hash) (defaults to: {}) —

    Additional options to pass to pt_online_alter_table

Raises:

  • (RuntimeError) —

    If an error occurs during column removal or batch size exceeds recommendations



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 379

def pt_drop_columns(table_name, column_names, options = {})
  column_names_array = Array(column_names)

  # Always validate batch size against recommendations
  recommendations = pt_get_recommendations(table_name)
  max_columns = recommendations[:max_columns_per_batch]

  if column_names_array.length > max_columns
    error_message = 'Too many columns for single batch operation. ' \
                   "Table '#{table_name}' (#{recommendations[:size_category]}, #{recommendations[:estimated_rows]} rows) " \
                   "supports max #{max_columns} columns per batch, but #{column_names_array.length} provided. " \
                   'Consider splitting into smaller batches.'

    BUSINESS_LOGGER.error('Batch size validation failed for pt_drop_columns', {
                            table_name: table_name,
                            columns_count: column_names_array.length,
                            max_allowed: max_columns,
                            size_category: recommendations[:size_category],
                            estimated_rows: recommendations[:estimated_rows],
                            operation: 'pt_drop_columns_validation_failed',
                          })

    raise ArgumentError, error_message
  end

  BUSINESS_LOGGER.info('Batch size validation passed for pt_drop_columns', {
                         table_name: table_name,
                         columns_count: column_names_array.length,
                         max_allowed: max_columns,
                         size_category: recommendations[:size_category],
                       })

  column_drops = column_names_array.map { |column_name| "DROP COLUMN #{column_name}" }
  alter_statement = column_drops.join(', ')
  pt_online_alter_table(table_name, alter_statement, options)
end

#pt_drop_foreign_key(table_name, constraint_name, options = {}) ⇒ void

This method returns an undefined value.

Drops a foreign key constraint from a table using pt-online-schema-change.

Examples:

Drop a foreign key constraint

pt_drop_foreign_key('users', 'fk_users_role_id')

Parameters:

  • table_name (String) —

    the name of the table containing the foreign key constraint

  • constraint_name (String) —

    the name of the foreign key constraint to drop

  • options (Hash) (defaults to: {}) —

    additional options to pass to pt-online-schema-change

Raises:

  • (RuntimeError) —

    if the pt_online_alter_table operation fails



276
277
278
279
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 276

def pt_drop_foreign_key(table_name, constraint_name, options = {})
  alter_statement = "DROP FOREIGN KEY #{constraint_name}"
  pt_online_alter_table(table_name, alter_statement, options)
end

#pt_drop_index(table_name, index_name) ⇒ void

This method returns an undefined value.

Drops an index from a database table using pt-online-schema-change.

Examples:

Drop the index on the email column

pt_drop_index('users', 'idx_users_email')

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • index_name (String, Symbol) —

    The name of the index to drop

Raises:

  • (RuntimeError) —

    If an error occurs during index removal



240
241
242
243
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 240

def pt_drop_index(table_name, index_name)
  alter_statement = "DROP INDEX #{index_name}"
  pt_online_alter_table(table_name, alter_statement)
end

#pt_execute_query(table_name, query, options = {}) ⇒ void

This method returns an undefined value.

Executes a custom ALTER statement using pt-online-schema-change. This method provides direct access to pt-osc for complex operations not covered by other helper methods.

Examples:

Execute a custom ALTER statement

pt_execute_query('users', 'MODIFY COLUMN email VARCHAR(320) NOT NULL, ADD INDEX idx_email_domain ((SUBSTRING_INDEX(email, "@", -1)))')

Add a generated column (MySQL 5.7+)

pt_execute_query('orders', 'ADD COLUMN total_with_tax DECIMAL(10,2) AS (subtotal * 1.07) STORED')

Parameters:

  • table_name (String, Symbol) —

    The name of the table to alter

  • query (String) —

    The raw ALTER statement without “ALTER TABLE table_name” prefix

  • options (Hash) (defaults to: {}) —

    Additional options to pass to pt_online_alter_table

Raises:

  • (RuntimeError) —

    If an error occurs during query execution



515
516
517
518
519
520
521
522
523
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 515

def pt_execute_query(table_name, query, options = {})
  BUSINESS_LOGGER.warn('Executing custom pt-osc query - ensure query syntax is correct', {
                         table_name: table_name,
                         query: query,
                         operation: 'pt_execute_custom_query',
                       })

  pt_online_alter_table(table_name, query, options)
end

#pt_get_recommendations(table_name) ⇒ Hash

Provides recommended batch sizing and configuration for pt-osc operations based on table size. This helps optimize performance and avoid timeouts on large tables.

Examples:

Get recommendations for a large table

recommendations = pt_get_recommendations('users')

Parameters:

  • table_name (String, Symbol) —

    The name of the table to analyze

Returns:

  • (Hash) —

    Recommended configuration options



534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 534

def pt_get_recommendations(table_name)
  # Get approximate row count
  row_count = begin
    result = ActiveRecord::Base.connection.exec_query(<<~SQL)
      SELECT table_rows
      FROM information_schema.tables
      WHERE table_schema = DATABASE()
        AND table_name = #{ActiveRecord::Base.connection.quote(table_name)}
    SQL

    if result.first.present?
      result.first['table_rows']&.to_i || result.first['TABLE_ROWS']&.to_i || 0
    else
      0
    end
  rescue StandardError => e
    BUSINESS_LOGGER.warn('Could not determine table size for recommendations', {
                           table_name: table_name,
                           error: e.message,
                         })
    0
  end

  # Provide recommendations based on table size
  case row_count
  when 0...100_000
    {
      max_columns_per_batch: 20,
      chunk_size: 1000,
      chunk_time: 0.5,
      monitor_interval: 30,
      recommended_batching: false,
      size_category: 'small',
    }
  when 100_000...1_000_000
    {
      max_columns_per_batch: 12,
      chunk_size: 500,
      chunk_time: 1.0,
      monitor_interval: 60,
      recommended_batching: true,
      size_category: 'medium',
    }
  when 1_000_000...10_000_000
    {
      max_columns_per_batch: 8,
      chunk_size: 200,
      chunk_time: 2.0,
      monitor_interval: 120,
      recommended_batching: true,
      size_category: 'large',
    }
  else
    {
      max_columns_per_batch: 5,
      chunk_size: 100,
      chunk_time: 3.0,
      monitor_interval: 180,
      recommended_batching: true,
      size_category: 'very_large',
    }
  end.merge(
    estimated_rows: row_count,
    table_name: table_name,
  )
end

#pt_online_alter_table(table_name, alter_statement, options = {}) ⇒ void

This method returns an undefined value.

Executes pt-online-schema-change on a table to safely alter large MySQL tables without blocking operations. Automatically performs a dry run first in development environments and includes optional monitoring.

Examples:

Basic usage with monitoring

pt_online_alter_table(:users, "ADD COLUMN email VARCHAR(255) NOT NULL")

Disable monitoring

pt_online_alter_table(:users, "ADD INDEX idx_status (status)", monitor: false)

Parameters:

  • table_name (String, Symbol) —

    The table to alter

  • alter_statement (String) —

    The ALTER statement without “ALTER TABLE table_name” prefix

  • options (Hash) (defaults to: {}) —

    Configuration options

Options Hash (options):

  • :skip_dry_run (Boolean) —

    Skip automatic dry run in development

  • :dry_run (Boolean) —

    Force dry run mode

  • :monitor (Boolean) — default: true —

    Enable progress monitoring

  • :monitor_interval (Integer) — default: 30 —

    Monitoring poll interval in seconds

  • :lag_threshold (Integer) — default: 10 —

    Replication lag threshold for warnings

  • :chunk_size (Integer) —

    Number of rows per chunk

  • :chunk_time (Float) —

    Sleep time between chunks in seconds

  • :max_load (String) —

    MySQL status variables threshold to pause

  • :critical_load (String) —

    MySQL status variables threshold to abort

  • :check_interval (Integer) —

    How often to check load in seconds

  • :progress (String) —

    Progress reporting format (e.g., “time,30”)

  • :no_drop_old_table (Boolean) —

    Keep original table as backup

  • :check_unique_key_change (Boolean) —

    Enable unique key validation (default: true)

  • :max_lag (Integer) —

    Maximum acceptable replication lag in seconds



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'app/my_lib/pt_online_schema_change/helper.rb', line 51

def pt_online_alter_table(table_name, alter_statement, options = {})
  # Set business context for comprehensive logging
  BUSINESS_LOGGER.set_business_context(
    {
      table_name: table_name,
      alter_statement: alter_statement,
      options: options.except(:password), # Don't log sensitive data
    }.compact,
  )

  BUSINESS_LOGGER.info('Starting pt-online-schema-change operation', {
                         table_name: table_name,
                         alter_statement: alter_statement,
                         options: options.except(:password),
                       })

  say "Running pt-online-schema-change on #{table_name}: #{alter_statement}"

  # Start monitoring if enabled (default: true)
  monitor_thread = nil
  if options.fetch(:monitor, true) && !options[:dry_run]
    monitor_options = {
      poll_interval: options[:monitor_interval] || 30,
      lag_threshold: options[:lag_threshold] || 10,
    }
    monitor_thread = PtOnlineSchemaChange::Monitor.start_monitoring(table_name.to_s, monitor_options)
  end

  begin
    # Always do a dry run first in development
    if Rails.env.development? && !options[:skip_dry_run]
      pt_osc_options = {
        dry_run: true,
        check_unique_key_change: options.fetch(:check_unique_key_change, true),
      }

      say 'Performing dry run first...'
      PtOnlineSchemaChange.execute_pt_osc(table_name, alter_statement, pt_osc_options)
    end

    # Execute the actual change
    PtOnlineSchemaChange.execute_pt_osc(table_name, alter_statement, options)

    BUSINESS_LOGGER.info('Completed pt-online-schema-change operation', {
                           table_name: table_name,
                           alter_statement: alter_statement,
                         })
  ensure
    # Stop monitoring
    if monitor_thread
      PtOnlineSchemaChange::Monitor.stop_monitoring(monitor_thread, 30)
    end
  end
rescue StandardError => e
  BUSINESS_LOGGER.error('Failed pt-online-schema-change operation', {
                          table_name: table_name,
                          alter_statement: alter_statement,
                          error: e.message,
                        })

  APMErrorHandler.report(
    e, context: {
      table_name: table_name,
      alter_statement: alter_statement,
    }
  )
  raise
end