def _execute_via_pt_osc(self, alter_statement, task,
      action=OnlineSchemaChangeActions.EXECUTE):
    """Executes an ALTER TABLE statement via pt-osc.
    """

    table, alter = _get_pt_osc_table_and_alter(alter_statement)
    username, password = task.credentials

    pt_osc_command = [
      'pt-online-schema-change',
      '--{}'.format(action),
      '--ask-pass',
      '--alter={}'.format(alter),
      '--user={}'.format(username),
    ] + PT_OSC_DEFAULT_FLAGS + [
      'h={0},D={1},t={2}'.format(task.database.cluster.get_master_hostname(),
          task.database.name, table),
    ]
    log.info('Executing online schema change: %s', alter_statement)

    pt_osc_process = None
    log_writer = LogWriter(log.info)
    try:
      log.info('Running command: %s', pt_osc_command)
      pt_osc_env = os.environ.copy()

      # If we're using a Packer-provided percona-toolkit, we should annotate our env's
      # paths to allow all Perl wrappers to link appropriately at runtime.
      if config.USE_PERCONA_TOOLKIT_FROM_PACKER:
        pt_library_paths = ':'.join(['./percona-toolkit/{}'.format(directory)
          for directory in ['lib', 'lib64', 'usr/lib', 'usr/lib64']])
        if 'LD_LIBRARY_PATH' not in pt_osc_env:
          pt_osc_env['LD_LIBRARY_PATH'] = pt_library_paths
        else:
          pt_osc_env['LD_LIBRARY_PATH'] = ':'.join(
              [pt_osc_env['LD_LIBRARY_PATH'], pt_library_paths])
        pt_osc_env['PATH'] = ':'.join(['./percona-toolkit/usr/bin', pt_osc_env['PATH']])

      pt_osc_process = pexpect.pty_spawn.spawn(command='/usr/bin/env',
          args=pt_osc_command, env=pt_osc_env)
      pt_osc_process.logfile_read = log_writer
      pt_osc_process.expect('Enter MySQL password: ')
      pt_osc_process.sendline(password)
      pt_osc_process.expect(pexpect.EOF)
    finally:
      if pt_osc_process is not None:
        pt_osc_process.close()

    if pt_osc_process.exitstatus != 0:
      self._add_result(task.id, text(alter_statement), (log_writer.all_output,))
      raise OnlineSchemaChangeError(log_writer.all_output)
    else:
      log.info('Successfully executed online schema change: %s', alter_statement)
      self._add_result(task.id, text(alter_statement), ('Statement completed successfully',))