#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Builtin function use - pylint: disable=W0141


# Copyright (c) 2013 Pivotal Software, Inc. All Rights Reserved
#
# This software contains the intellectual property of Pivotal Software, Inc.
# or is licensed to Pivotal Software, Inc. from third parties. Use of this
# software and the intellectual property contained therein is expressly
# limited to the terms and conditions of the License Agreement under which
# it is provided by or on behalf of Pivotal Software, Inc.

"""
gptransfer utility is for transferring one GPDB system to another. Currently
only table data or the entire system can be transferred.  In the future this
transfer will include dependent roles, UDFs, UDTs, views, resource queues,
and possibly a subset of system configuration of specified tables.
"""

import datetime
import hashlib
import inspect
import os
import re
import signal
import sys
import thread
import time
from collections import defaultdict
from optparse import OptionGroup, SUPPRESS_HELP
from threading import Thread, Event, Lock

from gppylib.gpversion import GpVersion
from gptransfer_modules.partition_comparators import PartitionComparatorFactory

try:
    from gppylib.mainUtils import simple_main, addStandardLoggingAndHelpOptions, ProgramArgumentValidationException
    from gppylib.gplog import get_default_logger
    from gppylib.gpparseopts import OptParser, OptChecker
    from gppylib.commands.base import Command, LOCAL, REMOTE, WorkerPool, ExecutionError, CommandResult
    from gppylib.commands.unix import MakeDirectory, RemoveDirectory, FileDirExists, RemoveGlob
    from gppylib.commands.gp import get_gphome
    from gppylib.db import dbconn
    from gppylib.db.dbconn import connect, DbURL, execSQL, execSQLForSingletonRow
    from gppylib.db.catalog import getUserDatabaseList, doesSchemaExist, dropSchemaIfExist
    from gppylib.gparray import GpArray
    from gppylib.userinput import ask_yesno
    from gppylib.operations.backup_utils import escapeDoubleQuoteInSQLString
    from pygresql.pg import DB
    from pygresql import pg
except ImportError, import_exception:
    sys.exit('Cannot import modules.  Please check that you have sourced'
             ' greenplum_path.sh.  Detail: %s' % str(import_exception))

# --------------------------------------------------------------------------
# pydoc variables
# --------------------------------------------------------------------------
__author__ = 'Chris Pedrotti <cpedrotti@gopivotal.com>'
__date__ = '1 February 2014'
__version__ = '0.0.1'

# --------------------------------------------------------------------------
# Description, help and usage constants
# --------------------------------------------------------------------------

__description__ = \
    """Transfers data from one Greenplum Database system to another."""

__help__ = []

# --------------------------------------------------------------------------
# Global Constants
# --------------------------------------------------------------------------

EXECNAME = os.path.split(__file__)[-1]
GPTRANSFER_PID_FILE = 'gptransfer.pid'

DEFAULT_BASE_WORKDIR = os.path.expanduser('~/')
GPTRANSFER_TMP_DIR = 'gptransfer_%d' % os.getpid()

now = datetime.datetime.now()
tableCountLock = Lock()
remaining_tables = 0
GPTRANSFER_FAILED_TABLES_FILE = 'failed_transfer_tables_%s.txt' % now.strftime(
    '%Y%m%d_%H%M%S')

DEFAULT_BATCH_SIZE = 2
MAX_BATCH_SIZE = 10
DEFAULT_SUB_BATCH_SIZE = 25
MAX_SUB_BATCH_SIZE = 50

SCHEMA_DELIMITER = '.'

DEFAULT_PORT = 5432
DEFAULT_USER = 'gpadmin'

DEFAULT_WAIT_TIME = 3

DEFAULT_MAX_GPFDIST_INSTANCES = 1
DEFAULT_GPFDIST_MAX_LINE_LENGTH = 1024 * 1024 * 10  # (10MB)
MIN_GPFDIST_MAX_LINE_LENGTH = 1024 * 32  # (32KB)
MAX_GPFDIST_MAX_LINE_LENGTH = 1024 * 1024 * 256  # (256MB)
DEFAULT_GPFDIST_TIMEOUT = 300
DEFAULT_GPFDIST_BASE_PORT = 8000
DEFAULT_GPFDIST_LAST_PORT = -1
MIN_GPFDIST_TIMEOUT = 2
MAX_GPFDIST_TIMEOUT = 600
GPFDIST_PORT_REGEX = r"Serving HTTP on port (\d+)"
VERSION_REGEX_PATTERN = re.compile(".*Greenplum Database (\d).*")

# --------------------------------------------------------------------------
# Logging
# --------------------------------------------------------------------------

logger = get_default_logger()

# --------------------------------------------------------------------------
# Global cancel flag and SIG method
# --------------------------------------------------------------------------

canceled = False
running_gpfdists = False


def wait_until_set_or_cancel(evt):
    while True:
        evt.wait(1)
        if evt.isSet() or canceled:
            return


def cancel_handler(_sig, _frame):  # pylint: disable=W0613
    """
    Sets canceled flag which starts the cancel process
    """
    global canceled
    logger.info('Canceling.  This may take some time...')
    canceled = True


def dump_threads(_sig, _frame):  # pylint: disable=W0613
    """
    Debug method that dumps threads
    """
    import traceback
    code = []
    for thread_id, stack in sys._current_frames().items():
        code.append("\n# ThreadID: %s" % thread_id)
        for filename, lineno, name, line in traceback.extract_stack(stack):
            code.append('File: "%s", line %d, in %s' %
                        (filename, lineno, name))
            if line:
                code.append("  %s" % (line.strip()))
    print "\n".join(code)


def dump_resource_usage(_sig, _frame):  # pylint: disable=W0613
    """
    Debug method that dumps resource usage
    """
    from resource import getrusage, RUSAGE_SELF

    keys = ['utime', 'stime', 'maxrss', 'ixrss', 'idrss',
            'isrss', 'minflt', 'majflt', 'nswap', 'inblock',
            'oublock', 'msgsnd', 'msgrcv', 'nsignals', 'nvcsw', 'nivcsw']
    usage = dict(zip(keys, list(getrusage(RUSAGE_SELF))))

    for key, val in usage.iteritems():
        logger.info("%s: %s", key, val)


# Set cancel on SIGINT
signal.signal(signal.SIGINT, cancel_handler)
# Set dump_threads on SIGUSR1
signal.signal(signal.SIGUSR1, dump_threads)
# Set dump_resource_usage on SIGUSR2
signal.signal(signal.SIGUSR2, dump_resource_usage)


# --------------------------------------------------------------------------
# Command line option parser
# --------------------------------------------------------------------------

def create_parser():
    """
    Creates the command line option parser for use by
    gppylib.mainUtils.simple_main
    """
    parser = OptParser(option_class=OptChecker,
                       version='%prog version $Revision: #1 $',
                       description=__description__)

    addStandardLoggingAndHelpOptions(parser, includeNonInteractiveOption=True)

    # General Options ####
    general_option_group = OptionGroup(parser, 'General Options')
    general_option_group.add_option(
        '--dry-run',
        dest='dry_run',
        default=False,
        action='store_true',
        help='Show what gptransfer will do without actually doing it'
    )
    general_option_group.add_option(
        '--batch-size',
        type='int',
        dest='batch_size',
        default=DEFAULT_BATCH_SIZE,
        action='store',
        metavar='<batch_size>',
        help='Table transfer concurrency (number of tables to transfer at one'
             ' time) [default: %d, maximum: %d]'
             % (DEFAULT_BATCH_SIZE, MAX_BATCH_SIZE)
    )
    general_option_group.add_option(
        '--sub-batch-size',
        type='int',
        dest='sub_batch_size',
        default=DEFAULT_SUB_BATCH_SIZE,
        action='store',
        metavar='<sub_batch_size>',
        help='Transfer operations concurrency [default: %d, maximum: %d]'
             % (DEFAULT_SUB_BATCH_SIZE, MAX_SUB_BATCH_SIZE)
    )
    general_option_group.add_option(
        '--enable-test',
        dest='enable_test',
        action='store_true',
        default=False,
        help=SUPPRESS_HELP
    )

    general_option_group.add_option(
        '--partition-transfer',
        dest='partition_transfer',
        default=False,
        action='store_true',
        help='Transfer partition tables if enabled, disabled by default, '
             'this is used in combination with -f option'
    )

    general_option_group.add_option(
        '--partition-transfer-non-partition-target',
        dest='partition_transfer_non_pt_target',
        default=False,
        action='store_true',
        help='Transfer partition tables to non-partition table if enabled, disabled by default, '
             'this is used in combination with -f option'
    )

    general_option_group.add_option(
        '--max-line-length',
        type='int',
        dest='max_line_length',
        default=DEFAULT_GPFDIST_MAX_LINE_LENGTH,
        action='store',
        metavar='<length>',
        help='Maximum line length for gpfdist [default: %d, minimum: %d,'
             ' maximum: %d]' % (DEFAULT_GPFDIST_MAX_LINE_LENGTH,
                                MIN_GPFDIST_MAX_LINE_LENGTH,
                                MAX_GPFDIST_MAX_LINE_LENGTH)
    )
    general_option_group.add_option(
        '--timeout',
        type='int',
        dest='timeout',
        default=DEFAULT_GPFDIST_TIMEOUT,
        action='store',
        metavar='<timeout>',
        help='gpfdist timeout in seconds [default: %d, minimum: %d, maximum: %d]'
             % (DEFAULT_GPFDIST_TIMEOUT, MIN_GPFDIST_TIMEOUT, MAX_GPFDIST_TIMEOUT)
    )
    general_option_group.add_option(
        '--delimiter',
        type='string',
        default=',',
        action='store',
        metavar='<delim>',
        help='Delimiter to use for external tables'
    )
    general_option_group.add_option(
        '--wait-time',
        type='int',
        dest='wait_time',
        default=DEFAULT_WAIT_TIME,
        action='store',
        metavar='<wait time>',
        help=SUPPRESS_HELP
    )
    general_option_group.add_option(
        '--force-standard-mode',
        dest='force_standard_mode',
        default=False,
        action='store_true',
        help=SUPPRESS_HELP
    )
    general_option_group.add_option(
        '--format',
        dest='format',
        default='CSV',
        action='store',
        choices=['CSV', 'csv', 'TEXT', 'text'],
        help='Transfer data in CSV(default) or TEXT format'
    )
    general_option_group.add_option(
        '--quote',
        dest='quote',
        default='\001',
        action='store',
        help='Specifies the quotation character for CSV mode. The default is "\001".'
    )

    parser.add_option_group(general_option_group)

    # Source Options ####
    source_option_group = OptionGroup(parser, 'Source Options')

    # source host
    source_option_group.add_option(
        '--source-host',
        type='string',
        dest='source_host',
        default='127.0.0.1',
        action='store',
        metavar='<host>',
        help='Source Greenplum Database hostname or IP address'
    )
    # source port
    source_option_group.add_option(
        '--source-port',
        type='int',
        dest='source_port',
        default=DEFAULT_PORT,
        action='store',
        metavar='<port>',
        help='Source Greenplum Database port number'
    )
    # source addresses file
    source_option_group.add_option(
        '--source-map-file',
        type='string',
        dest='source_map_file',
        default=None,
        action='store',
        metavar='<file>',
        help='File containing all source hostname to address mappings'
    )
    # source user
    source_option_group.add_option(
        '--source-user',
        type='string',
        dest='source_user',
        default=DEFAULT_USER,
        action='store',
        metavar='<user>',
        help='User to connect to source Greenplum Database as'
    )
    # base port for gpfdist instances
    source_option_group.add_option(
        '--base-port',
        type='int',
        dest='base_port',
        default=DEFAULT_GPFDIST_BASE_PORT,
        action='store',
        metavar='<base_port>',
        help='Base port for gpfdist on source systems',
    )
    source_option_group.add_option(
        '--last-port',
        type='int',
        dest='last_port',
        default=DEFAULT_GPFDIST_LAST_PORT,
        action='store',
        metavar='<last_port>',
        help=SUPPRESS_HELP
    )
    # full
    source_option_group.add_option(
        '--full',
        action='store_true',
        default=False,
        help='Full transfer of the entire GPDB source system to destination'
    )
    # tables
    source_option_group.add_option(
        '-t',
        type='string',
        dest='tables',
        default=[],
        action='append',
        metavar='<db.schema.table>',
        help='Table to transfer.  Can be specified multiple times to include '
             'multiple tables'
    )

    # tables
    source_option_group.add_option(
        '-T',
        type='string',
        dest='exclude_tables',
        default=[],
        action='append',
        metavar='<db.schema.table>',
        help='Table to exclude from transfer.  Can be specified multiple times to exclude '
             'multiple tables'
    )

    # database
    source_option_group.add_option(
        '-d',
        type='string',
        dest='databases',
        default=[],
        action='append',
        metavar='<database>',
        help='Database to transfer.  Can be specified multiple times to '
             'include multiple databases'
    )
    # file
    source_option_group.add_option(
        '-f',
        type='string',
        dest='input_file',
        default=None,
        action='store',
        metavar='<file>',
        help='File containing list of database.schema.table entries to transfer, '
             'if --partition-transfer option is specified, each line is a comma '
             'separated pair of full qualified source partition and destination '
             'partition table name, if dest partition table name is not specified, '
             'same source partition table name will be used in dest database'
    )
    # file
    source_option_group.add_option(
        '-F',
        type='string',
        dest='exclude_input_file',
        default=None,
        action='store',
        metavar='<file>',
        help='File containing list of database.schema.table entries to exclude from transfer'
    )

    # exclusive lock
    source_option_group.add_option(
        '-x',
        dest='exclusive_lock',
        default=False,
        action='store_true',
        help='Acquire exclusive lock on source table during the transfer '
             'to prevent insert/updates'
    )
    # work base directory
    source_option_group.add_option(
        '--work-base-dir',
        type='string',
        dest='work_base_dir',
        default=DEFAULT_BASE_WORKDIR,
        action='store',
        metavar='<directory>',
        help='Base directory to create gptransfer work directory'
    )

    # max gpfdist instances per host
    source_option_group.add_option(
        '--max-gpfdist-instances',
        type='int',
        dest='max_gpfdist_instances',
        default=DEFAULT_MAX_GPFDIST_INSTANCES,
        action='store',
        metavar='<instance count>',
        help=SUPPRESS_HELP  # 'Maximum gpfdist instances per host'
    )

    parser.add_option_group(source_option_group)

    # Dest Options ####
    dest_option_group = OptionGroup(parser, 'Destination Options')

    # destination host
    dest_option_group.add_option(
        '--dest-host',
        type='string',
        dest='dest_host',
        default='127.0.0.1',
        action='store',
        metavar='<host>',
        help='Destination Greenplum Database hostname or IP address'
    )
    # destination port
    dest_option_group.add_option(
        '--dest-port',
        type='int',
        dest='dest_port',
        default=DEFAULT_PORT,
        action='store',
        metavar='<port>',
        help='Destination Greenplum Database port number'
    )
    # destination user
    dest_option_group.add_option(
        '--dest-user',
        type='string',
        dest='dest_user',
        default=DEFAULT_USER,
        action='store',
        metavar='<user>',
        help='User to connect to destination Greenplum Database as'
    )
    # destination database
    dest_option_group.add_option(
        '--dest-database',
        type='string',
        dest='dest_database',
        default=None,
        action='store',
        metavar='<database>',
        help='Database to transfer all source tables into'
    )
    # skip tables that exist
    dest_option_group.add_option(
        '--skip-existing',
        dest='skip_existing',
        default=False,
        action='store_true',
        help='Skip tables that exist in destination Greenplum Database '
             'instead of erroring out'
    )
    # truncate tables that exist
    dest_option_group.add_option(
        '--truncate',
        default=False,
        action='store_true',
        help='Truncate destination table if it exists prior to transferring data'
    )

    # drop tables that exist
    dest_option_group.add_option(
        '--drop',
        default=False,
        action='store_true',
        help='Drop destination table if it exists prior to transferring data'
    )

    # analyze destination table after transfer
    dest_option_group.add_option(
        '--analyze',
        default=False,
        action='store_true',
        help='Analyze table after transfer'
    )

    dest_option_group.add_option(
        '--schema-only',
        dest='schema_only',
        default=False,
        action='store_true',
        help='Create schema only.  Do not transfer data'
    )

    parser.add_option_group(dest_option_group)

    # Validation Options ####
    validation_option_group = OptionGroup(parser, 'Validation Options')

    validation_option_group.add_option(
        '--validate',
        type='string',
        dest='validator',
        default=None,
        action='store',
        metavar='<validator>',
        help='Method to validate source and destination tables. '
             '[Available validators: %s]' %
             ', '.join(validator_factory.get_available_validators())
    )

    validation_option_group.add_option(
        '--no-final-count',
        dest='no_final_count_validation',
        default=False,
        action='store_true',
        help='Do not run final count validation'
    )

    parser.add_option_group(validation_option_group)

    # Gpfdist Options ####
    gpfdist_option_group = OptionGroup(parser, 'Gpfdist Options')

    gpfdist_option_group.add_option(
        '--gpfdist-verbose',
        dest='gpfdist_verbose',
        action='store_true',
        default=False,
        help='Run gpfdist with -v'
    )

    gpfdist_option_group.add_option(
        '--gpfdist-very-verbose',
        dest='gpfdist_very_verbose',
        action='store_true',
        default=False,
        help='Run gpfdist with -V'
    )

    parser.add_option_group(gpfdist_option_group)

    parser.setHelp(__help__)

    return parser


def wait_for_pool(pool, max_queued, wait_until_empty):
    """
    Waits for pool to empty below the number of workers.  This helps prevent
    high memory usage by queueing up lots of commands.

    pool: pool to wait on
    max_queued: maximum number of commands.  Once number queued drops below this
                it returns completed commands
    wait_until_empty: wait until the work queue is empty

    returns: successful and failed commands
    """
    successful_commands = list()
    failed_commands = list()
    # TODO: we should not be doing this but by using the max queue items.
    #       Unfortunately this is an arg to Queue() which isn't
    #       exposed through our worker pool class yet.
    while pool.work_queue.qsize() >= max_queued or wait_until_empty:
        if wait_until_empty:
            pool.join()

        has_completed_cmds = False
        for cmd in pool.getCompletedItems():
            has_completed_cmds = True
            res = cmd.get_results()
            if not res.wasSuccessful():
                failed_commands.append(cmd)
            else:
                successful_commands.append(cmd)
        if (has_completed_cmds and not (wait_until_empty and pool.work_queue.qsize() == 0)) or \
                ((canceled or wait_until_empty) and pool.work_queue.qsize() == 0):
            break
        time.sleep(1)

    return successful_commands, failed_commands


# --------------------------------------------------------------------------

def split_fqn(fqn):
    """
    Splits a fully qualified database table name (<db>.<schema>.<table>)
    into its three components.
    """

    index = 0
    partitions = ['', '', '']
    regexp_parsing = False

    for c in fqn:
        if index > 2:
            break
        if c != '/' and c != SCHEMA_DELIMITER:
            partitions[index] = partitions[index] + c
        elif c == '/':
            partitions[index] = partitions[index] + c
            if not regexp_parsing:
                regexp_parsing = True
            else:
                regexp_parsing = False
        else:
            if not regexp_parsing:
                index += 1
            else:
                partitions[index] = partitions[index] + c
    if index != 2:
        raise Exception('Invalid fully qualified table name %s' % fqn)

    (database, schema, table) = partitions

    return database, schema, table


def _process_regexp(regexp=None):
    """
    Add a dot before patterns which require preceding expression
    Add ^ and $ to match from begin to the end of line.
    """

    if regexp is None:
        return regexp

    regexp = validate_regexp(regexp)

    if not regexp.startswith('^'):
        regexp = '^' + regexp
    if not regexp.endswith('$'):
        regexp = regexp + '$'

    return regexp


def validate_regexp(regexp=None):
    """
    validating the user input and parse out the regular experssion
    """

    if regexp.count('/') % 2 != 0:
        raise Exception('Regular expression should be surround with /')

    return regexp.replace('/', '')


def get_user_databases(host, port, user):
    """
    get all user databases in the gpdb system
    """
    all_dbs = list()

    url = DbURL(host, port, 'template1', user)
    conn = connect(url)
    all_dbs = [db[0] for db in getUserDatabaseList(conn)]
    conn.close()

    return all_dbs


def get_databases_by_regexp(host, port, user, databases, warning=False):
    """
    Retrieves a list of all databases based on matching the regular expressions.
    """
    dbs = list()
    all_dbs = get_user_databases(host, port, user)

    for database in databases:
        db_matched = False
        for db in all_dbs:
            try:
                if re.match(r'%s' % database, db, re.L | re.U) and db not in dbs:
                    dbs.append(db)
                    db_matched = True
            except Exception, e:
                raise Exception("Error matching regular expression %s with %s\n%s" % (database, db, e))
        if not db_matched and warning:
            logger.warning('Find no user databases matching "%s" in source system' % database[1: len(database) - 1])

    return dbs


def get_user_tables(host, port, user, databases=None, full=False, partition_transfer=False):
    """
    Retrieves all tables from a list of databases on a GPDB system.

    host: hostname of GPDB system
    port: port of the GPDB system
    user: username to use to connect to GPDB system
    full: True if to retrieve tables of all databases
    """

    # We need to use this SQL query because we only want the top level
    # partition.
    table_sql_part1 = """
SELECT
    n.nspname, c.relname, c.relstorage
FROM
    pg_class c JOIN pg_namespace n ON (c.relnamespace=n.oid)
    JOIN pg_catalog.gp_distribution_policy p ON (c.oid = p.localoid)
WHERE """
    table_sql_filter_out_child_partitions = """c.oid NOT IN ( SELECT parchildrelid as oid FROM pg_partition_rule ) AND """
    table_sql_part2 = """ n.nspname NOT IN ('gpexpand', 'pg_bitmapindex', 'information_schema', 'gp_toolkit');"""
    all_tables_sql = table_sql_part1 + \
                     ('' if partition_transfer else table_sql_filter_out_child_partitions) + \
                     table_sql_part2

    all_tables = set()

    if full:
        databases = get_user_databases(host, port, user)
    elif databases is None or len(databases) == 0:
        return all_tables

    # get all the tables in the source system
    for database in databases:
        url = DbURL(host, port, database, user)
        conn = connect(url)
        cur = execSQL(conn, all_tables_sql)
        for row in cur:
            schema = row[0]
            table = row[1]
            external = (row[2] == 'x')
            all_tables.add(GpTransferTable(database, schema, table, external))
        cur.close()
        conn.close()
    return all_tables


def is_table_external(host, port, user, database, schema, table):
    """
    Checks if the table is an external table
    """
    is_external_sql = """
SELECT
    c.relstorage = 'x'
FROM
    pg_class c JOIN pg_namespace n ON (c.relnamespace=n.oid)
WHERE c.relname = '%s' AND n.nspname = '%s'"""
    conn = connect(DbURL(host, port, database, user))
    res = execSQLForSingletonRow(conn, is_external_sql % (table, schema))
    conn.close()
    return res[0]


def schema_exists_on_system(host, port, user, schema, databases=None):
    """
    host: hostname or ip address of Greenplum DB
    port: port of Greenplum DB
    user: user to connect as
    schema: schema name to search for
    databases: list of databases to search for schema

    Returns True if schema exists in the user database
    """

    if databases is None:
        url = DbURL(host, port, 'template1', user)
        conn = connect(url)
        databases = [db[0] for db in getUserDatabaseList(conn)]
        conn.close()

    for database in databases:
        url = DbURL(host, port, database, user)
        conn = connect(url)
        schema_exists = doesSchemaExist(conn, schema)
        conn.close()
        if schema_exists:
            return True


def drop_existing_schema_on_system(host, port, user, schema, databases=None):
    """
    host: hostname or ip address of Greenplum DB
    port: port of Greenplum DB
    user: user to connect as
    schema: schema name to search for
    databases: list of databases to search for schema
    """

    if databases is None:
        url = DbURL(host, port, 'template1', user)
        conn = connect(url)
        databases = [db[0] for db in getUserDatabaseList(conn)]
        conn.close()

    for database in databases:
        url = DbURL(host, port, database, user)
        conn = connect(url)
        dropSchemaIfExist(conn, schema)
        conn.close()

# --------------------------------------------------------------------------
# Validation classes
# --------------------------------------------------------------------------

class TableValidator(object):
    """
    Base class for table validators.
    """

    def __init__(self, work_dir, src_conn, dest_conn, src_sql, dest_sql):
        """
        src_conn: Source database connection
        dest_conn: Destination database connection
        src_sql: SQL to execute on source database
        dest_sql: SQL to execute on the destination database
        """

        self._src_conn = src_conn
        self._dest_conn = dest_conn
        self._src_sql = src_sql
        self._dest_sql = dest_sql
        self._src_res = None
        self._dest_res = None
        self._src_failed = False
        self._dest_failed = False
        self._work_dir = work_dir

    @staticmethod
    def setup(conn):
        """
        Child classes that need to perform some setup, such as create a UDF,
        should override this method.

        conn: Database connection to use for any setup SQL statements
        """
        pass

    @staticmethod
    def cleanup(conn):
        """
        Child classes that need to perform some cleanup, such as drop a UDF,
        should override this method.

        conn: Database connection to use for any cleanup SQL statements
        """
        pass

    def validate(self):
        """
        Starts threads that perform the validation SQL statements on the source
        and destination systems and waits for those threads to complete.  Also
        handles the canceling of validation.  In most cases child classes will
        not need to override this method.
        """

        source_thread = Thread(target=self._src_proc)
        source_thread.start()
        dest_thread = Thread(target=self._dest_proc)
        dest_thread.start()

        while True:
            if not source_thread.isAlive():
                if self._src_failed == True:
                    DB(self._dest_conn).cancel()
                    break

            if not dest_thread.isAlive():
                if self._dest_failed == True:
                    DB(self._src_conn).cancel()
                    break

            if not dest_thread.isAlive() and not source_thread.isAlive():
                break

            if canceled == True:
                try:
                    DB(self._src_conn).cancel()
                except:
                    pass
                try:
                    DB(self._dest_conn).cancel()
                except:
                    pass
                break

            time.sleep(1)

        source_thread.join()
        dest_thread.join()

        return self._compare()

    @staticmethod
    def get_name():
        """
        Gets the name of the validator which will be used as a valid arg to the
        --validator option.  The name must be unique amongst all validators.
        """
        return None

    def _src_proc(self):
        """
        Thread proc that executes the SQL statement on the source side.  To
        use this method the SQL statement must only return a single row.  If
        your query requires more complexity you should override this.
        """

        try:
            self._src_res = execSQLForSingletonRow(self._src_conn,
                                                   self._src_sql)
        except:
            self._src_failed = True

    def _dest_proc(self):
        """
        Thread proc that executes the SQL statement on the destination side.  To
        use this method the SQL statement must only return a single row.  If
        your query requires more complexity you should override this.
        """

        try:
            self._dest_res = execSQLForSingletonRow(self._dest_conn,
                                                    self._dest_sql)
        except:
            self._dest_failed = True

    def _compare(self):
        """
        Simple compare function that checks the source and destination rows
        returned are the same.
        """

        if not self._src_res or not self._dest_res:
            return False
        return self._src_res == self._dest_res


# --------------------------------------------------------------------------

class CountTableValidator(TableValidator):
    """
    Simple COUNT(*) "validation".
    """

    def __init__(self, work_dir, table_pair, src_conn=None, dest_conn=None):
        """
        table_pair: table pair to validate
        src_conn: Database connection to the source system
        dest_conn: Database connection to the destination system
        """
        sql = "SELECT count(*) FROM %s.%s"
        src_schema = table_pair.source.schema
        src_table = table_pair.source.table
        dest_schema = table_pair.dest.schema
        dest_table = table_pair.dest.table

        TableValidator.__init__(self, work_dir, src_conn, dest_conn,
                                sql % (escapeDoubleQuoteInSQLString(src_schema), escapeDoubleQuoteInSQLString(src_table)),
                                sql % (escapeDoubleQuoteInSQLString(dest_schema), escapeDoubleQuoteInSQLString(dest_table)))

    def init_dest_dict(self, dest_tables_dict=None, table_pair=None):
        try:
            dest_res = execSQLForSingletonRow(self._dest_conn, self._dest_sql)[0]
            dest_tables_dict[str(table_pair.dest)] = dest_res
        except:
            self._dest_failed = True

    def accumulate(self, dest_tables_dict=None, table_pair=None):
        try:
            src_res = execSQLForSingletonRow(self._src_conn, self._src_sql)[0]
            dest_tables_dict[str(table_pair.dest)] -= src_res
        except:
            self._dest_failed = True

    @staticmethod
    def get_name():
        """
        Returns 'count'
        """
        return 'count'


# --------------------------------------------------------------------------


class HashMergeTableValidator(TableValidator):
    """
    Validator class that defines some methods used by MD5 and SHA-256 validation
    """

    def _src_proc(self):
        """
        Thread proc that executes the SQL statement on the source side.  To
        use this method the SQL statement must only return a single row.  If
        your query requires more complexity you should override this.
        """

        try:
            self._src_res = execSQL(self._src_conn, self._src_sql)

        except:
            self._src_failed = True
        finally:
            cmd = GpCloseNamedPipe(
                'close source validation pipe', self._src_pipe, REMOTE, self._src_host)
            cmd.run()

    def _dest_proc(self):
        """
        Thread proc that executes the SQL statement on the destination side.  To
        use this method the SQL statement must only return a single row.  If
        your query requires more complexity you should override this.
        """

        try:
            self._dest_res = execSQL(self._dest_conn, self._dest_sql)
        except:
            self._dest_failed = True
        finally:
            cmd = GpCloseNamedPipe(
                'close dest validation pipe', self._dest_pipe, REMOTE, self._dest_host)
            cmd.run()

# --------------------------------------------------------------------------


class MD5MergeTableValidator(HashMergeTableValidator):
    """
    Validation that compares the MD5 hashes of all the rows in a table.
    """

    class MD5Sum(Command):
        """
        Prints md5sum of file.
        """

        def __init__(self, name, filename, ctxt=LOCAL, remoteHost=None):
            """
            name: command name
            filename: file to calculate md5sum of
            """
            cmdStr = "md5sum %s | awk '{print \\$1}'" % filename
            Command.__init__(self, name, cmdStr, ctxt, remoteHost)

    def __init__(self, work_dir, table_pair, src_conn, dest_conn):
        """
        table_pair: table pair to validate
        src_conn: Database connection to the source system
        dest_conn: Database connection to the destination system
        """

        sql = """COPY (SELECT md5(textin(record_out(t.*))) hash
                 FROM %s.%s t ORDER BY hash) TO '%s'"""
        src_schema = table_pair.source.schema
        src_table = table_pair.source.table
        dest_schema = table_pair.dest.schema
        dest_table = table_pair.dest.table
        self._pool = WorkerPool(2)
        self._src_pipe = os.path.join(
            work_dir, str(table_pair.source), 'src_md5_validation')
        self._dest_pipe = os.path.join(
            work_dir, str(table_pair.source), 'dest_md5_validation')

        self._src_host = DB(src_conn).host
        self._dest_host = DB(dest_conn).host

        src_pipe_cmd = GpCreateNamedPipe('Create source validation pipe',
                                         self._src_pipe,
                                         REMOTE, self._src_host)
        self._pool.addCommand(src_pipe_cmd)

        dest_pipe_cmd = GpCreateNamedPipe('Create destination validation pipe',
                                          self._dest_pipe,
                                          REMOTE, self._dest_host)
        self._pool.addCommand(dest_pipe_cmd)

        self._pool.join()
        self._pool.check_results()

        self._src_md5_cmd = MD5MergeTableValidator.MD5Sum(
            'Source MD5Sum', self._src_pipe, REMOTE, self._src_host)
        self._pool.addCommand(self._src_md5_cmd)

        self._dest_md5_cmd = MD5MergeTableValidator.MD5Sum(
            'Dest MD5Sum', self._dest_pipe, REMOTE, self._dest_host)
        self._pool.addCommand(self._dest_md5_cmd)

        TableValidator.__init__(self, work_dir, src_conn, dest_conn,
                                sql % (src_schema, src_table, self._src_pipe),
                                sql % (dest_schema, dest_table, self._dest_pipe))

    def _compare(self):
        """
        Diff the output of the hashes
        """
        res = False
        try:
            self._pool.join()
            self._pool.check_results()
            if not self._dest_failed and not self._src_failed:
                res = ((self._src_md5_cmd.get_results()).stdout.strip()
                       == (self._dest_md5_cmd.get_results()).stdout.strip())
        finally:
            pass
        return res

    @staticmethod
    def get_name():
        """
        Returns 'md5'
        """
        return 'md5'


# --------------------------------------------------------------------------


class SHA256MergeTableValidator(HashMergeTableValidator):
    """
    Validation that compares the SHA256 hashes of all the rows in a table.
    """

    class SHA256Sum(Command):
        """
        Prints sha256sum of file.
        """

        def __init__(self, name, filename, ctxt=LOCAL, remoteHost=None):
            """
            name: command name
            filename: file to calculate sha256sum of
            """
            cmdStr = "sha256sum %s | awk '{print \\$1}'" % filename
            Command.__init__(self, name, cmdStr, ctxt, remoteHost)

    def __init__(self, work_dir, table_pair, src_conn, dest_conn):
        """
        table_pair: table pair to validate
        src_conn: Database connection to the source system
        dest_conn: Database connection to the destination system
        """

        sql = """CREATE OR REPLACE FUNCTION pg_temp.digest(text, text)
RETURNS bytea
AS '$libdir/pgcrypto', 'pg_digest'
LANGUAGE C IMMUTABLE STRICT;

COPY (SELECT pg_temp.digest(textin(record_out(t.*)), 'sha256') hash
FROM %s.%s t ORDER BY hash) TO '%s'"""
        src_schema = table_pair.source.schema
        src_table = table_pair.source.table
        dest_schema = table_pair.dest.schema
        dest_table = table_pair.dest.table
        self._pool = WorkerPool(2)
        self._src_pipe = os.path.join(
            work_dir, str(table_pair.source), 'src_sha256_validation')
        self._dest_pipe = os.path.join(
            work_dir, str(table_pair.source), 'dest_sha256_validation')

        self._src_host = DB(src_conn).host
        self._dest_host = DB(dest_conn).host

        src_pipe_cmd = GpCreateNamedPipe('Create source validation pipe',
                                         self._src_pipe,
                                         REMOTE, self._src_host)
        self._pool.addCommand(src_pipe_cmd)

        dest_pipe_cmd = GpCreateNamedPipe('Create destination validation pipe',
                                          self._dest_pipe,
                                          REMOTE, self._dest_host)
        self._pool.addCommand(dest_pipe_cmd)

        self._pool.join()
        self._pool.check_results()

        self._src_sha256_cmd = SHA256MergeTableValidator.SHA256Sum(
            'Source SHA256Sum', self._src_pipe, REMOTE, self._src_host)
        self._pool.addCommand(self._src_sha256_cmd)

        self._dest_sha256_cmd = SHA256MergeTableValidator.SHA256Sum(
            'Dest SHA256Sum', self._dest_pipe, REMOTE, self._dest_host)
        self._pool.addCommand(self._dest_sha256_cmd)

        TableValidator.__init__(self, work_dir, src_conn, dest_conn,
                                sql % (src_schema, src_table, self._src_pipe),
                                sql % (dest_schema, dest_table, self._dest_pipe))

    def _compare(self):
        """
        Diff the output of the hashes
        """
        res = False
        try:
            self._pool.join()
            self._pool.check_results()
            if not self._dest_failed and not self._src_failed:
                res = ((self._src_sha256_cmd.get_results()).stdout.strip()
                       == (self._dest_sha256_cmd.get_results()).stdout.strip())
        finally:
            pass
        return res

    @staticmethod
    def get_name():
        """
        Returns 'sha256'
        """
        return 'sha256'


# --------------------------------------------------------------------------


class TableValidatorFactory(object):
    """
    Class reponsible for managing the various validation classes.
    """

    # dict of discovered validators
    __available_validators = {}

    def __init__(self):
        """
        Discovers validator classes using inpection
        """
        classes = inspect.getmembers(sys.modules[__name__],
                                     lambda m: inspect.isclass(m) and
                                               m.__module__ == __name__ and
                                               issubclass(m, TableValidator))
        for (_, c) in classes:
            name = c.get_name()
            if name:
                self.__available_validators[name] = c

    def get_available_validators(self):
        """
        Returns a list of available validator names
        """

        return TableValidatorFactory.__available_validators.keys()

    def get_validator(self, name):
        """
        Returns the class of the validator specified.

        name: name of validator to return class for
        """

        try:
            return TableValidatorFactory.__available_validators[name]
        except:
            raise Exception('Unknown validator %s' % name)


# --------------------------------------------------------------------------
# Validator factory
# --------------------------------------------------------------------------
validator_factory = TableValidatorFactory()


# --------------------------------------------------------------------------
# Command classes for data transfer
# --------------------------------------------------------------------------


class PsqlFile(Command):
    """
    psql command for executing file
    """

    def __init__(self, name, host, port, user, database, filename):
        """
        name: Command name
        host: GPDB hostname
        port: GPDB port
        user: GPDB user
        database: GPDB database
        filename: SQL file to execute
        """
        cmdStr = 'psql -U %s -h %s -p %d -f %s %s' \
                 % (user, host, port, filename, database)
        Command.__init__(self, name, cmdStr, LOCAL, None)


# --------------------------------------------------------------------------

class CreateDB(Command):
    """
    Create database command
    """

    def __init__(self, name, host, port, user, database):
        """
        name: Command name
        host: GPDB hostname
        port: GPDB port
        user: GPDB user
        database: GPDB database
        """
        cmdStr = 'createdb -U %s -h %s -p %d %s' \
                 % (user, host, port, database)
        Command.__init__(self, name, cmdStr, LOCAL, None)


# --------------------------------------------------------------------------

class GpCreateNamedPipe(Command):
    """
    Command for creating a named pipe on *nix systems.
    """

    def __init__(self, name, fifo_name, ctxt=LOCAL, remoteHost=None):
        """
        name: name of the command
        fifo_name: full path of the named pipe to create
        """

        cmdStr = 'mkfifo %s' % fifo_name
        Command.__init__(self, name, cmdStr, ctxt, remoteHost)


# --------------------------------------------------------------------------
class GpCloseNamedPipe(Command):
    """
    Opens and closes named pipe to make sure process at other end received
    EOF
    """

    def __init__(self, name, fifo_name, ctxt=LOCAL, remoteHost=None):
        """
        name: name of the command
        fifo_name: full path to the named pipe
        """
        cmdStr = """python -c 'f = open("%s", "w+"); f.close()'""" % fifo_name
        Command.__init__(self, name, cmdStr, ctxt, remoteHost)


# --------------------------------------------------------------------------


class GpCreateGpfdist(Command):
    """
    Command for starting a gpfdist instance.
    """

    def __init__(self, name, directory, data_file, port, last_port, max_line_length, timeout,
                 pid_file, log_file, ctxt=LOCAL, remoteHost=None, verbosity=""):
        """
        name: name of the command
        dir: directory for gpfdist to use as its root directory
        port: port for gpfdist to listen on
        last_port: last port for gpfdist to listen on
        max_line_length: maximum line length setting for gpfdist
        pid_file: full path of the pid file to create
        """

        self._data_file = data_file

        cmdStr = \
            'nohup gpfdist -d %s -p %d -P %d -m %d -t %d %s> %s 2>&1 < /dev/null & echo \\$! > ' \
            '%s && bash -c "(sleep 1 && kill -0 \`cat %s 2> /dev/null\` && cat %s) || (cat %s >&2 && exit 1)"' % \
            (directory, port, last_port, max_line_length, timeout, verbosity, log_file, pid_file, pid_file, log_file, log_file)

        Command.__init__(self, name, cmdStr, ctxt, remoteHost)

    def get_url(self):
        url = None
        res = self.get_results()
        if res and res.wasSuccessful():
            host = self.remoteHost
            m = re.search(GPFDIST_PORT_REGEX, res.stdout)
            if m:
                port = m.group(1)
                url = "gpfdist://%s:%s/%s" % (host, port, self._data_file)

        if url is None:
            raise Exception("Failed to find port in gpfdist log file")
        return url


# --------------------------------------------------------------------------

class GpCleanupGpfdist(Command):
    """
    Command for terminating a running gpfdist instance and removing its pid file
    """

    def __init__(self, name, pid_file, log_file, ctxt=LOCAL, remoteHost=None, keep_logs=False):
        """
        name: name of the command
        pid_file: full path of the pid file for the gpfdist instance to
                  terminate
        """

        if keep_logs:
            log_file = ""
        # We have to kill and then wait for the process to fully exit so we
        # can reuse the port.  Without this delay the next gpfdist instance
        # that tries to start using the same port can fail to bind.
        cmdStr = """python -c 'import os
import signal
import sys
import time

pid = -1

try:
    with open(sys.argv[1], "r") as f:
        pid = int(f.readline().strip())
except:
    pass

if len(sys.argv) == 3 and os.path.isfile(sys.argv[2]):
    os.unlink(sys.argv[2])

if pid == -1:
    sys.exit(0)

check_count = 0

try:
    os.kill(pid, signal.SIGTERM)
except:
    pass

while True:
    try:
        os.kill(pid,0)
    except:
        break
    time.sleep(1)
    check_count += 1
    if check_count >= 10:
        try:
            os.kill(pid, signal.SIGKILL)
            while True:
                try:
                    os.kill(pid,0)
                except:
                    break
                time.sleep(1)
                check_count += 1
                if check_count >= 20:
                    os.unlink(sys.argv[1])
                    sys.exit(1)
        except:
            pass

os.unlink(sys.argv[1])' %s %s""" % (pid_file, log_file)

        Command.__init__(self, name, cmdStr, ctxt, remoteHost)


# --------------------------------------------------------------------------

class GpSchemaDump(Command):
    """
    Command for dumping the schema of a GPDB table.  Currently it only dumps
    the table schema and does not include ownership or dependent database
    objects.
    """

    def __init__(self, name, host, port, user, full=True, database=None,
                 schema=None, table=None, ctxt=LOCAL, remoteHost=None):
        """
        name: name of the command
        host: GPDB host
        port: GPDB port
        user: GPDB user (must be admin account or table owner)
        full: Full dump of system
        database: database containing table to dump
        schema: schema containing table to dump
        table: table to dump
        """
        cmdStr = None
        self.full = full

        if not full and (database is None or schema is None or table is None):
            raise Exception('database, schema and table are required for '
                            'table dump')

        if full:
            cmdStr = 'pg_dumpall -s --gp-syntax -h %s -p %d -U %s' \
                     % (host, port, user)
        else:
            cmdStr = 'pg_dump -s -x -O --gp-syntax -h %s -p %d -U %s -t ' \
                     '\'\"%s\".\"%s\"\' %s' % (host, port, user, schema, table, database)

        Command.__init__(self, name, cmdStr, ctxt, remoteHost)

    def run(self):
        """
        Runs the command.
        """

        Command.run(self)

    def get_schema_sql(self):
        """
        Returns a string containing the SQL statement to create the table.
        """

        # Look for the start of the dump text to exclude anything like a MOTD
        sql = None
        res = self.get_results()
        if res is None:
            raise Exception('Command has not been run')

        if self.full:
            sql = res.stdout
        else:
            start_text = """--
-- Greenplum Database database dump
--"""
            start_pos = res.stdout.find(start_text)
            if start_pos == -1:
                return None
            sql = res.stdout[start_pos:]

        return sql


# --------------------------------------------------------------------------
class GpTransferCommand(Command):
    """
    Command to transfer data in a table from one GPDB system to another.

    """

    def __init__(
            self, name, src_host, src_port, src_user, dest_host, dest_port,
            dest_user, table_pair, dest_exists, truncate, analyze, drop,
            fast_mode, exclusive_lock, schema_only, work_dir,
            host_map, source_config, batch_size, gpfdist_port, gpfdist_last_port,
            gpfdist_instance_count, gpfdist_verbosity, max_line_length, timeout, wait_time,
            delimiter, validator, format, quote, table_transfer_set_total):
        """
        name: name of the command
        src_host: source GPDB host
        src_port: source GPDB port
        src_user: source GPDB user
        dest_host: destination GPDB host
        dest_port: destination GPDB port
        dest_user: destination GPDB user
        table_pair: description of source -> dest table transfer
        dest_exists: if the destination table exists already
        truncate: if the destination table should be truncated
        analyze: if the destination table should be analyzed after transfer
        drop: if the destination table should be dropped
        fast_mode: should fast mode of operation be used
        exclusive_lock: exclusive lock the source table
        schema_only: only create table
        work_dir: the work directory to create named pipes in
        host_map: the host to ip mapping of the source system
        source_config: the GpArray of the source GPDB system
        batch_size: the size of the WorkerPool for the command
        gpfdist_port: gpfdist port
        gpfdist_last_port: last gpfdist port
        gpfdist_instance_count: gpfdist instances per source host
        max_line_length: the gpfdist maximum line length
        timeout: the gpfdist timeout
        wait_time: time to wait on destination query (TODO: remove this in next release)
        delimiter: delimiter char to use for external tables
        validator: validator to use
        format: transfer data in CSV(default) or TEXT format
        quote: specifies the quotation character for CSV mode
        table_transfer_set_total: Number of tables need to be transferred
        """

        self._src_host = src_host
        self._src_port = src_port
        self._src_user = src_user
        self._dest_host = dest_host
        self._dest_port = dest_port
        self._dest_user = dest_user
        self._table_pair = table_pair
        self._dest_exists = dest_exists
        self._truncate = truncate
        self._analyze = analyze
        self._drop = drop
        self._fast_mode = fast_mode
        self._exclusive_lock = exclusive_lock
        self._schema_only = schema_only
        self._work_dir = work_dir
        self._host_map = host_map
        self._source_config = source_config
        self._batch_size = batch_size
        self._gpfdist_port = gpfdist_port
        self._gpfdist_last_port = gpfdist_last_port
        self._gpfdist_instance_count = gpfdist_instance_count
        self._gpfdist_verbosity = gpfdist_verbosity
        self._max_line_length = max_line_length
        self._timeout = timeout
        self._wait_time = wait_time
        self._delimiter = delimiter
        self._format = format
        self._quote = quote
        self._table_transfer_set_total = table_transfer_set_total
        # _used_ports is a dict where key is hostname and val is list of
        # ints (ports used)
        self._pipe = os.path.join(work_dir,
                                  str(table_pair.source),
                                  '%s.pipe' % str(table_pair.source))
        self._wext_gpfdist_urls = list()
        self._ext_gpfdist_urls = list()
        self._wext_name = ('w_ext_%s_%s' % (self._table_pair.source.table,
                                            hashlib.sha256(str(self._table_pair.source)).hexdigest()))[0:63]
        self._ext_name = ('ext_%s_%s' % (self._table_pair.source.table,
                                         hashlib.sha256(str(self._table_pair.source)).hexdigest()))[0:63]
        self._pool = None
        if validator:
            self._validator_class = validator_factory.get_validator(validator)
        else:
            self._validator_class = None

        # connections saved off so we can cancel executing SQL from outside
        # the thread executing the SQL statement
        self._src_conn = None
        self._dest_conn = None
        self._src_failed = False
        self._dest_failed = False
        self._status_msg = 'Success'
        self._src_exception = None
        self._dest_exception = None
        self._success = False
        self._failed_to_stop_gpfdists = False
        self._cleanup_gpfdist_needed = False
        self._cleanup_named_pipes_needed = False
        self._src_ready = Event()
        self._dest_ready = Event()

        Command.__init__(self, name, None, LOCAL, None)

    def run(self):
        """
        Runs the full table transfer, executing all steps needed.
        """
        global tableCountLock
        global remaining_tables

        try:
            if not canceled:
                self._pool = WorkerPool(self._batch_size)
                logger.info('Starting transfer of %s to %s...',
                            self._table_pair.source,
                            self._table_pair.dest)

                # NOTE: These connections are used for table validation
                url = DbURL(self._src_host, self._src_port,
                            self._table_pair.source.database, self._src_user)
                self._src_conn = connect(url)

                url = DbURL(self._dest_host, self._dest_port,
                            self._table_pair.dest.database, self._dest_user)
                self._dest_conn = connect(url)

                if not self._dest_exists:
                    self._create_target_table()
                elif self._truncate and not self._table_pair.dest.external:
                    self._truncate_target_table()
                elif self._drop:
                    self._drop_target_table()
                    self._create_target_table()
                if self._schema_only:
                    self._success = True
                    return 0
                if not self._table_pair.dest.external:
                    self._cleanup_named_pipes_needed = True
                    self._create_named_pipe()
                    self._cleanup_gpfdist_needed = True
                    self._start_write_gpfdist()
                    self._start_read_gpfdist()
                    self._create_source_wext()
                    self._create_dest_ext()

                    self._transfer_data()
                    if self._validator_class:
                        self._validate()
                    if self._analyze:
                        self._analyze_dest_table()
                    self._cleanup_dest_ext()
                    self._cleanup_source_wext()
            else:
                self.set_results(
                    CommandResult(1, 'Canceled', None, False, False))
        except ExecutionError, exerr:
            self.set_results(exerr.cmd.get_results())
        except Exception, ex:
            self.set_results(CommandResult(1, str(ex), None, False, False))
        else:
            self._success = True
        finally:
            if self._cleanup_gpfdist_needed:
                try:
                    self._cleanup_gpfdist()
                except:
                    global running_gpfdists
                    running_gpfdists = True

            if self._cleanup_named_pipes_needed:
                try:
                    self._cleanup_named_pipe()
                except:
                    pass  # will be cleaned up at the end

            if self._src_conn:
                self._src_conn.commit(
                ) if self._success else self._src_conn.rollback()
                self._src_conn.close()

            if self._dest_conn:
                self._dest_conn.commit(
                ) if self._success else self._dest_conn.rollback()
                self._dest_conn.close()

            if self._pool:
                self._pool.haltWork()
                self._pool.joinWorkers()

            if not self.get_results():
                self.set_results(
                    CommandResult(0, 'Success', None, True, False))

            if not canceled:
                if self._success:
                    with tableCountLock:
                        remaining_tables = remaining_tables - 1
                    logger.info(
                        "Finished transferring table %s, remaining %s of %s tables", str(self._table_pair.source),
                        remaining_tables, self._table_transfer_set_total)
                else:
                    logger.error(
                        "Failed to transfer table %s", str(self._table_pair.source))
                    if self._src_exception:
                        logger.error(self._src_exception)
                    if self._dest_exception:
                        logger.error(self._dest_exception)
                    logger.info('Remaining %s of %s tables', remaining_tables, self._table_transfer_set_total)

    def _get_distributed_by(self):
        sql = '''SELECT attname
FROM (
      SELECT localoid, UNNEST(attrnums) AS colnum, GENERATE_SERIES(1, ARRAY_UPPER(attrnums, 1)) AS colorder
      FROM gp_distribution_policy
 ) d
 JOIN pg_attribute a ON (d.localoid = a.attrelid AND d.colnum = a.attnum)
 JOIN pg_class c ON (d.localoid = c.oid)
 JOIN pg_namespace n ON (c.relnamespace = n.oid)
 WHERE nspname = '%s'
 AND relname = '%s'
 ORDER BY colorder;''' % (self._table_pair.source.schema,
                           self._table_pair.source.table)
        try:
            cur = execSQL(self._src_conn, sql)
            attnames = cur.fetchall()
            cur.close()
            if len(attnames) == 0:
                return 'DISTRIBUTED RANDOMLY'

            # NOTE: ideally, we can use pg.escape_identifier for this part, but
            # that is introduced in pygresql 4.1 and we are in 4.0
            escaped_attnames = [escapeDoubleQuoteInSQLString(attname[0]) for attname in attnames]
            distributed_by = ", ".join(escaped_attnames)
            return '''DISTRIBUTED BY (%s)''' % distributed_by
        except Exception as e:
            logger.warn('Unable to determine distribution policy for "%s". Using '
                        'distributed randomly instead.\nmessage: %s' % (self._table_pair.dest, str(e)))
            return 'DISTRIBUTED RANDOMLY'

    def _create_target_table(self):
        """
        Creates the table on the target GPDB system.
        """
        logger.info('Creating target table %s...', self._table_pair.dest)
        schema_sql = self._get_source_table_schema()
        cur = execSQL(self._dest_conn, schema_sql)
        cur.close()

    def _truncate_target_table(self):
        """
        Truncates the table on the target GPDB system.
        """

        logger.info('Truncating target table %s...', self._table_pair.dest)
        cur = execSQL(self._dest_conn, 'TRUNCATE TABLE \"%s\".\"%s\"'
                      % (self._table_pair.dest.schema,
                         self._table_pair.dest.table))
        cur.close()

    def _drop_target_table(self):
        """
        Drops the table on the target GPDB system.
        """

        logger.info('Dropping target table %s...', self._table_pair.dest)
        cur = execSQL(self._dest_conn, 'DROP TABLE \"%s\".\"%s\"'
                      % (self._table_pair.dest.schema,
                         self._table_pair.dest.table))
        cur.close()

    def _get_source_table_schema(self):
        """
        Gets the SQL to create the table from the source GPDB system.
        """

        self._pool.empty_completed_items()
        logger.info('Retrieving schema for table %s...',
                    self._table_pair.source)
        cmd = GpSchemaDump(
            'schema dump of %s' % self._table_pair.source,
            self._src_host,
            self._src_port,
            self._src_user,
            False,
            self._table_pair.source.database,
            self._table_pair.source.schema,
            self._table_pair.source.table,
            ctxt=REMOTE,
            remoteHost=self._src_host
        )
        self._pool.addCommand(cmd)
        self._pool.join()
        self._pool.check_results()
        return cmd.get_schema_sql()

    def _get_named_pipes(self):
        """
        """
        named_pipes = list()
        if self._fast_mode:
            for seg in self._source_config.getSegDbList():
                if seg.isSegmentMirror(True) or seg.isSegmentQD():
                    continue
                pipe = "%s.%d" % (self._pipe, seg.getSegmentContentId())
                address = iter(self._host_map[seg.getSegmentHostName()]).next()
                named_pipes.append((address, pipe))
        else:
            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()
                for i in xrange(0, self._gpfdist_instance_count):
                    pipe = "%s.%d" % (self._pipe, i)
                    named_pipes.append((address, pipe))
        return named_pipes

    def _create_named_pipe(self):
        """
        Creates all the named pipes needed for the table transfer on the
        source GPDB system.
        """

        self._pool.empty_completed_items()
        logger.debug('Creating FIFO pipes for source table %s...',
                     self._table_pair.source)

        for host in self._host_map.keys():
            address = iter(self._host_map[host]).next()
            cmd = MakeDirectory('create dir for table %s'
                                % self._table_pair.source,
                                os.path.dirname(self._pipe), REMOTE, address)
            self._pool.addCommand(cmd)

        cmd = MakeDirectory('create dir for table %s'
                            % self._table_pair.source,
                            os.path.dirname(self._pipe), REMOTE, self._src_host)
        self._pool.addCommand(cmd)
        cmd = MakeDirectory('create dir for table %s'
                            % self._table_pair.source,
                            os.path.dirname(self._pipe), REMOTE, self._dest_host)
        self._pool.addCommand(cmd)

        self._pool.join()
        self._pool.check_results()

        for (address, pipe) in self._get_named_pipes():
            cmd = GpCreateNamedPipe('Create pipe for table %s on %s'
                                    % (self._table_pair.source, address),
                                    pipe, REMOTE, address)

            self._pool.addCommand(cmd)
        self._pool.join()
        self._pool.check_results()

    def _create_source_wext(self):
        """
        Creates the writable external table on the source GPDB system.
        """

        logger.debug('Creating source writable external table for source '
                     'table %s...', self._table_pair.source)

        urls = ','.join(["'%s'" % url for url in self._wext_gpfdist_urls])

        if self._fast_mode:
            distributed_clause = self._get_distributed_by()
            wext_sql = \
                '''CREATE WRITABLE EXTERNAL WEB TABLE gptransfer.%s (LIKE \"%s\".\"%s\")
                   EXECUTE 'cat > %s.$GP_SEGMENT_ID'
                   FORMAT '%s'
                ''' % (self._wext_name,
                       self._table_pair.source.schema,
                       self._table_pair.source.table,
                       self._pipe,
                       self._format)
            if self._format.lower() == 'csv':
                wext_sql += """ (DELIMITER AS ',' QUOTE AS E'%s')""" % self._quote
            elif self._format.lower() == 'text':
                wext_sql += """ (DELIMITER AS E'%s' ESCAPE AS 'off')""" % self._delimiter
            wext_sql += """ ENCODING 'UTF8' %s""" % distributed_clause
        else:
            wext_sql = \
                '''CREATE WRITABLE EXTERNAL TABLE gptransfer.%s ( LIKE \"%s\".\"%s\")
                   LOCATION (%s)
                   FORMAT '%s' ''' \
                % (self._wext_name,
                   self._table_pair.source.schema,
                   self._table_pair.source.table,
                   urls,
                   self._format)
            if self._format.lower() == 'csv':
                wext_sql += """ (DELIMITER AS ',' QUOTE AS E'%s')""" % self._quote
            elif self._format.lower() == 'text':
                wext_sql += """ (DELIMITER AS E'%s' ESCAPE AS 'off')""" % self._delimiter
            wext_sql += """ ENCODING \'UTF8\' DISTRIBUTED RANDOMLY"""
        cur = execSQL(self._src_conn, wext_sql)
        cur.close()

    def _create_dest_ext(self):
        """
        Creates the readable external table on the destination GPDB system.
        """

        logger.debug('Creating external table for destination table %s...',
                     self._table_pair.dest)
        urls = ','.join(["'%s'" % url for url in self._ext_gpfdist_urls])
        ext_sql = \
            """CREATE EXTERNAL TABLE gptransfer.%s (LIKE \"%s\".\"%s\")
               LOCATION(%s)
               FORMAT '%s' """ \
            % (self._ext_name,
               self._table_pair.dest.schema,
               self._table_pair.dest.table,
               urls,
               self._format)
        if self._format.lower() == 'csv':
            ext_sql += """(DELIMITER AS ',' QUOTE AS E'%s')""" % self._quote
        elif self._format.lower() == 'text':
            ext_sql += """(DELIMITER AS E'%s' ESCAPE AS 'off')""" % self._delimiter
        cur = execSQL(self._dest_conn, ext_sql)
        cur.close()

    def _analyze_dest_table(self):
        """
        Analyzes destination table
        """

        logger.info('Analyzing destination table %s...', self._table_pair.dest)
        sql = 'ANALYZE \"%s\".\"%s\"' % (self._table_pair.dest.schema,
                                         self._table_pair.dest.table)
        cur = execSQL(self._dest_conn, sql)
        cur.close()

    def _start_write_gpfdist(self):
        """
        Starts gpfdist instances for write
        """
        self._pool.empty_completed_items()
        if self._fast_mode:
            return
        else:
            logger.info(
                'Starting gpfdist for writable external table for table %s...',
                self._table_pair.source)
            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()

                for i in xrange(0, self._gpfdist_instance_count):
                    cmd = GpCreateGpfdist(
                        'gpfdist for table %s on %s' % (self._table_pair.source,
                                                        address),
                        os.path.dirname(self._pipe),
                        "%s.%d" % (os.path.basename(self._pipe), i),
                        self._gpfdist_port,
                        self._gpfdist_last_port,
                        self._max_line_length,
                        self._timeout,
                        os.path.join(self._work_dir, 'gpfdist_write_%s_%d.pid'
                                     % (str(self._table_pair.source), i)),
                        os.path.join(self._work_dir, 'gpfdist_write_%s_%d.log'
                                     % (str(self._table_pair.source), i)),
                        ctxt = REMOTE,
                        remoteHost = address,
                        verbosity =  self._gpfdist_verbosity
                    )
                    self._pool.addCommand(cmd)

            self._pool.join()
            for cmd in self._pool.getCompletedItems():
                if not cmd.get_results().wasSuccessful():
                    raise ExecutionError("Error Executing Command: ", cmd)
                url = cmd.get_url()
                self._wext_gpfdist_urls.append(url)

            self._pool.empty_completed_items()

    def _start_read_gpfdist(self):
        """
        Starts gpfdist instances for read
        """
        self._pool.empty_completed_items()

        logger.debug(
            'Starting gpfdist for readable external table for table %s...',
            self._table_pair.source)

        if self._fast_mode:
            for seg in self._source_config.getSegDbList():
                if seg.isSegmentMirror(True) or seg.isSegmentQD():
                    continue
                address = iter(self._host_map[seg.getSegmentHostName()]).next()
                cmd = GpCreateGpfdist(
                    'gpfdist for table %s on %s' % (self._table_pair.source,
                                                    seg.getSegmentHostName()),
                    os.path.dirname(self._pipe),
                    "%s.%d" % (os.path.basename(self._pipe), seg.getSegmentContentId()),
                    self._gpfdist_port,
                    self._gpfdist_last_port,
                    self._max_line_length,
                    self._timeout,
                    os.path.join(self._work_dir, 'gpfdist_read_%s_%d.pid'
                                 % (str(self._table_pair.source), seg.getSegmentContentId())),
                    os.path.join(self._work_dir, 'gpfdist_read_%s_%d.log'
                                 % (str(self._table_pair.source), seg.getSegmentContentId())),
                    ctxt = REMOTE,
                    remoteHost = address,
                    verbosity =  self._gpfdist_verbosity
                )
                self._pool.addCommand(cmd)
        else:
            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()

                for i in xrange(0, self._gpfdist_instance_count):
                    cmd = GpCreateGpfdist(
                        'gpfdist for table %s on %s' % (self._table_pair.source,
                                                        address),
                        os.path.dirname(self._pipe),
                        "%s.%d" % (os.path.basename(self._pipe), i),
                        self._gpfdist_port,
                        self._gpfdist_last_port,
                        self._max_line_length,
                        self._timeout,
                        os.path.join(self._work_dir, 'gpfdist_read_%s_%d.pid'
                                     % (str(self._table_pair.source), i)),
                        os.path.join(self._work_dir, 'gpfdist_read_%s_%d.log'
                                     % (str(self._table_pair.source), i)),
                        ctxt = REMOTE,
                        remoteHost = address,
                        verbosity =  self._gpfdist_verbosity
                    )
                    self._pool.addCommand(cmd)
        self._pool.join()

        for cmd in self._pool.getCompletedItems():
            if not cmd.get_results().wasSuccessful():
                raise ExecutionError("Error Executing Command: ", cmd)
            url = cmd.get_url()
            self._ext_gpfdist_urls.append(url)

        self._pool.empty_completed_items()

    def _transfer_data(self):
        """
        Moves data from source to destination GPDB system.
        """
        self._pool.empty_completed_items()

        logger.info('Transfering data %s -> %s...',
                    self._table_pair.source, self._table_pair.dest)
        source_thread = Thread(target=self._do_wext_select_proc)
        source_thread.start()
        dest_thread = Thread(target=self._do_ext_insert_select_proc)
        dest_thread.start()

        while True:
            time.sleep(1)
            if not source_thread.isAlive():
                if self._src_failed == True:
                    DB(self._dest_conn).cancel()
                else:
                    # Make sure all named pipes get an EOF on them or empty tables
                    # will hang.
                    # TODO: currently no way to ensure gpfdist on the write side has
                    #       written all data available to the pipe.  Closing the pipe
                    #       while data is remaining causes and EPIPE on the writable
                    #       gpfdist side.  Need a better way to coordinate this.
                    time.sleep(self._wait_time)
                    logger.debug('Closing named pipes for table %s...', str(self._table_pair.source))
                    for (address, pipe) in self._get_named_pipes():
                        cmd = GpCloseNamedPipe('close pipe %s on %s' % (pipe, address),
                                               pipe, REMOTE, address)
                        self._pool.addCommand(cmd)

                    self._pool.join()
                    self._pool.check_results()
                break

            if not dest_thread.isAlive():
                if self._dest_failed == True:
                    DB(self._src_conn).cancel()

            if canceled == True:
                try:
                    logger.info('Canceling source query...')
                    DB(self._src_conn).cancel()
                    source_thread.join()
                except Exception, ex:
                    logger.info(
                        'Exception while canceling source query: %s', str(ex))

                try:
                    logger.info('Canceling destination query...')
                    DB(self._dest_conn).cancel()
                    dest_thread.join()
                except Exception, ex:
                    logger.info(
                        'Exception while canceling destination query: %s', str(ex))

                break

        if source_thread.isAlive():
            source_thread.join()
        if dest_thread.isAlive():
            dest_thread.join()

        if self._src_failed or self._dest_failed:
            raise Exception('Failed to transfer data')

        return True

    def _do_wext_select_proc(self):
        """
        Thread proc for the source side of the transfer.
        """
        cur = None
        try:
            query = """INSERT INTO gptransfer.%s SELECT * FROM \"%s\".\"%s\"""" \
                    % (self._wext_name,
                       self._table_pair.source.schema,
                       self._table_pair.source.table)
            if self._exclusive_lock:
                lock_query = "LOCK TABLE \"%s\".\"%s\" IN EXCLUSIVE MODE;" % \
                             (self._table_pair.source.schema, self._table_pair.source.table)
                execSQL(self._src_conn, lock_query)

            self._src_ready.set()
            wait_until_set_or_cancel(self._dest_ready)
            if canceled:
                return

            cur = execSQL(self._src_conn, query)
        except Exception, ex:
            self._src_failed = True
            self._src_exception = ex
        finally:
            if cur:
                cur.close()
            thread.exit()

    def _do_ext_insert_select_proc(self):
        """
        Thread proc for the destination side of the transfer.
        """
        cur = None
        try:
            query = """INSERT INTO \"%s\".\"%s\" SELECT * FROM gptransfer.%s""" \
                    % (self._table_pair.dest.schema,
                       self._table_pair.dest.table,
                       self._ext_name)
            if self._exclusive_lock:
                lock_query = "LOCK TABLE \"%s\".\"%s\" IN EXCLUSIVE MODE" % \
                             (self._table_pair.dest.schema, self._table_pair.dest.table)
                execSQL(self._dest_conn, lock_query)

            self._dest_ready.set()
            wait_until_set_or_cancel(self._src_ready)
            if canceled:
                return

            time.sleep(1)
            cur = execSQL(self._dest_conn, query)
            cur.close()
        except Exception, ex:
            self._dest_failed = True
            self._dest_exception = ex
        finally:
            if cur:
                cur.close()
            thread.exit()

    def _cleanup_source_wext(self):
        """
        Cleans up the writable external table on the source GPDB system.
        """

        try:
            logger.debug(
                'Dropping writable external table for source table %s...',
                self._table_pair.source)
            drop_wext_sql = """DROP EXTERNAL WEB TABLE gptransfer.%s""" \
                            % (self._wext_name)
            cur = execSQL(self._src_conn, drop_wext_sql)
            cur.close()
        except:
            # We'll drop at the end when we wipe out the gptransfer schema
            pass

    def _cleanup_dest_ext(self):
        """
        Cleans up the readable external table on the destination GPDB system.
        """

        try:
            logger.debug('Dropping external table for destination table %s...',
                         self._table_pair.dest)
            drop_ext_sql = """DROP EXTERNAL TABLE gptransfer.%s""" \
                           % (self._ext_name)
            cur = execSQL(self._dest_conn, drop_ext_sql)
            cur.close()
        except:
            # We'll drop at the end when we wipe out the gptransfer schema
            pass

    def _cleanup_gpfdist(self):
        """
        Stops the gpfdist instances on the source GPDB system that were started
        for this table transfer.
        """

        self._pool.empty_completed_items()
        logger.debug('Stopping gpfdist for source table %s...',
                     self._table_pair.source)
        keep_logs = self._gpfdist_verbosity != ""
        if self._fast_mode:
            for seg in self._source_config.getSegDbList():
                if seg.isSegmentMirror(True) or seg.isSegmentQD():
                    continue
                address = iter(self._host_map[seg.getSegmentHostName()]).next()
                cmd = GpCleanupGpfdist('stopping gpfdist on %s' % address,
                                       os.path.join(self._work_dir, 'gpfdist_read_%s_%d.pid' % (
                                           str(self._table_pair.source), seg.getSegmentContentId())),
                                       os.path.join(self._work_dir, 'gpfdist_read_%s_%d.log' % (
                                           str(self._table_pair.source), seg.getSegmentContentId())),
                                       REMOTE, address, keep_logs)
                self._pool.addCommand(cmd)
        else:
            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()

                for i in xrange(0, self._gpfdist_instance_count):
                    cmd = GpCleanupGpfdist('stopping gpfdist on %s' % address,
                                           os.path.join(self._work_dir,
                                                        'gpfdist_read_%s_%d.pid' % (str(self._table_pair.source), i)),
                                           os.path.join(self._work_dir,
                                                        'gpfdist_read_%s_%d.log' % (str(self._table_pair.source), i)),
                                           REMOTE, address, keep_logs)
                    self._pool.addCommand(cmd)
                    cmd = GpCleanupGpfdist('stopping gpfdist on %s' % address,
                                           os.path.join(self._work_dir,
                                                        'gpfdist_write_%s_%d.pid' % (str(self._table_pair.source), i)),
                                           os.path.join(self._work_dir,
                                                        'gpfdist_write_%s_%d.log' % (str(self._table_pair.source), i)),
                                           REMOTE, address, keep_logs)
                    self._pool.addCommand(cmd)

        self._pool.join()
        self._pool.check_results()

    def _cleanup_named_pipe(self):
        """
        Deletes the named pipes on the source GPDB system that were created for
        this table transfer.
        """

        self._pool.empty_completed_items()
        logger.debug('Removing FIFO pipes for source table %s...',
                     self._table_pair.source)

        for host in self._host_map.keys():
            address = iter(self._host_map[host]).next()
            cmd = RemoveDirectory('remove dir for table %s'
                              % self._table_pair.source,
                              os.path.dirname(self._pipe),
                              REMOTE, address)
            self._pool.addCommand(cmd)

        cmd = RemoveDirectory('remove dir for table %s'
                          % self._table_pair.source,
                          os.path.dirname(self._pipe),
                          REMOTE, self._dest_host)
        self._pool.addCommand(cmd)
        cmd = RemoveDirectory('remove dir for table %s'
                          % self._table_pair.source,
                          os.path.dirname(self._pipe),
                          REMOTE, self._src_host)
        self._pool.addCommand(cmd)

        self._pool.join()
        self._pool.check_results()

    def _validate(self):
        """
        Validates the table transfer.
        """

        logger.info(
            'Validating destination table %s...', self._table_pair.dest)

        validator = self._validator_class(self._work_dir,
                                          self._table_pair,
                                          self._src_conn,
                                          self._dest_conn)
        if not validator.validate():
            logger.error('Validation failed for %s', self._table_pair.dest)
            self._success = False
            self._status_msg = 'Validation failed'
        else:
            logger.info('Validation of %s successful', self._table_pair.dest)
            self._success = True

    def get_table_pair(self):
        """
        Returns table pair for this transfer task
        """
        return self._table_pair


# --------------------------------------------------------------------------

class GpTransferTable(object):
    """
    Class for describing one side of the data transfer.
    """

    def __init__(self, database, schema, table, external):
        """
        database: database name
        schema: schema name
        table: table name
        """

        assert database is not None
        assert schema is not None
        assert table is not None
        (self.database, self.schema, self.table) = (database, schema,
                                                    table)
        self.external = external

    def __str__(self):
        """
        Fully qualified string representation of the table.
        """

        return '%s.%s.%s' % (self.database, self.schema, self.table)

    def __eq__(self, other):
        """
        Equality operator.
        """

        return self.__dict__ == other.__dict__

    def __hash__(self):
        """
        Hash value of the GpTransferTable
        """

        return hash(str(self))

    def is_identifier_supported(self):
        return (self._is_identifier_supported(self.database) and
                self._is_identifier_supported(self.schema) and
                self._is_identifier_supported(self.table))

    def _is_identifier_supported(self, identifier):
        """
        Verify that the characters of identifier starts with a-z, or A-Z or underscore,
        followed by lower or upper case (a-z, A-Z), or digits(0-9), or underscore (_)
        """

        if not ((identifier[0:1] >= 'a' and identifier[0:1] <= 'z') or
                    (identifier[0:1] >= 'A' and identifier[0:1] <= 'Z') or
                        identifier[0:1] == '_'):
            return False

        for c in identifier:
            if not ((c >= 'a' and c <= 'z') or
                        (c >= 'A' and c <= 'Z') or
                        (c >= '0' and c <= '9') or
                        (c == '_')):
                return False
        return True


class GpTransferTablePair(object):
    """
    Wrapper the describes the source and destination table transfer.
    """

    def __init__(self, source, dest):
        """
        source: the source GpTransferTable
        dest: the destination GpTransferTable
        """

        assert source is not None
        assert dest is not None
        self.source = source
        self.dest = dest

    def __str__(self):
        """
        String representation of the transfer pair.
        """

        return '%s -> %s' % (self.source, self.dest)

    def __eq__(self, other):
        """
        Equality operator.

        other: obj to compare to
        """

        return self.source == other.source and self.dest == other.dest

    def __hash__(self):
        """
        Hash value of transfer pair.
        """

        return hash(str(self))


# --------------------------------------------------------------------------
#
# --------------------------------------------------------------------------

class GpTransfer(object):
    """
    Main application class for data transfer.
    """

    def __init__(self, options, _args):  # pylint: disable=W0613
        """
        options: options dict from the options parser.
        _args: args from the options parser.  Not used.
        """
        self._options = options
        self.src_dest_partition_table_mapping = None
        self._pre_validate_options()
        self._work_dir = os.path.join(self._options.work_base_dir,
                                      GPTRANSFER_TMP_DIR)

        self._cleanup_schemas = False
        self._excluding_table = True

        # --format would be 'TEXT' if delimiter is other than ','
        if self._options.delimiter != ",":
            self._options.format = 'text'

        # get GpArray objects of source and destination GPDB systems
        self._src_config = None
        self._dest_config = None
        self._get_configurations()

        self._src_databases = None
        self._dest_databases = None

        self._src_dest_partition_table_mapping = None

        self._all_src_databases = get_user_databases(self._options.source_host,
                                                     self._options.source_port,
                                                     self._options.source_user)
        self._all_dest_databases = get_user_databases(self._options.dest_host,
                                                      self._options.dest_port,
                                                      self._options.dest_user)

        self.multiple_same_target_set = set()
        self._src_tables = self._get_source_tables()

        self._validate_schema_not_exists()

        # get all the tables from the destination GPDB system
        logger.info("Retrieving list of destination tables...")
        self._dest_tables = \
            get_user_tables(self._options.dest_host,
                            self._options.dest_port,
                            self._options.dest_user,
                            [db for db in self._dest_databases if db in self._all_dest_databases],
                            False,
                                self._options.partition_transfer)
        self._dest_non_partition_tables = set()
        if self._options.partition_transfer:
            self._dest_non_partition_tables = \
                get_user_tables(self._options.dest_host,
                                self._options.dest_port,
                                self._options.dest_user,
                                [db for db in self._dest_databases if db in self._all_dest_databases],
                                False,
                                partition_transfer=False)
        # segment address list
        self._host_map = self._get_host_map()
        # build up table pairs to transfer and validate them
        self._table_transfer_set = self._build_table_transfer_list()
        self._table_transfer_set_total = len(self._table_transfer_set)

        self._fast_mode = \
            True if (self._dest_config.numPrimarySegments >= self._src_config.numPrimarySegments and
                     not options.force_standard_mode) \
                else False
        if self._fast_mode:
            logger.info('gptransfer will use "fast" mode for transfer.')
        else:
            logger.info('gptransfer will use "standard" mode for transfer.')

        if self._options.exclusive_lock:
            logger.info("Exclusive locks will be used during table transfers.")

        if not self._options.no_final_count_validation:
            self._validator_class = validator_factory.get_validator('count')
        else:
            self._validator_class = None

        self._validate_host_map()

        # Get number of gpfdists per source host
        source_host_count = len(self._host_map.keys())
        self._gpfdist_instance_count = \
            min(self._dest_config.numPrimarySegments / source_host_count,
                self._options.max_gpfdist_instances)

        # validate options
        self._post_validate_options()

        self._validate_table_transfer_set()
        if self._options.partition_transfer or self._options.partition_transfer_non_pt_target:
            self._validate_partition_table_transfer_set()

        self._validate_filespace(self._options.source_host, self._options.source_port, self._options.source_user,
                                 self._options.dest_host, self._options.dest_port, self._options.dest_user,
                                 self._options.full)

        # At most we'll execute primary segment count of commands in the subbatches
        # so we can reduce it if the value exceeds the number of primary segs.
        if self._src_config.get_primary_count() < self._options.sub_batch_size \
                and not self._options.enable_test:
            self._options.sub_batch_size = self._src_config.get_primary_count()
        logger.info('Using batch size of %d', self._options.batch_size)
        logger.info('Using sub-batch size of %d', self._options.sub_batch_size)

        self._pool = WorkerPool(self._options.batch_size)

    def run(self):
        """
        Runs the data transfer based on the options provided.
        """
        global remaining_tables
        remaining_tables = len(self._table_transfer_set)
        if not self._options.dry_run:
            self.setup()

            if self._options.full and self._options.schema_only:
                # we are done if full and schema only
                return 0

            # Hacky, but if we don't set num_assigned then isDone() will never
            # be done.
            self._pool.empty_completed_items()
            assert self._pool.work_queue.qsize() == 0
            self._pool.num_assigned = 0

            # Create all the commands to do the transfers
            # To avoid OOM errors with a large number of tables
            # we add commands in batches
            failed_tables = list()
            max_queued = self._options.batch_size + 1
            cmds_queued = 0

            self.before_dest_tables_dict = {}
            if self._options.partition_transfer_non_pt_target:
                self.before_dest_tables_dict = self.get_dest_row_count_before_transfer()

            gpfdist_verbosity = ""
            if self._options.gpfdist_verbose:
                gpfdist_verbosity = "-v "
            elif self._options.gpfdist_very_verbose:
                gpfdist_verbosity = "-V "

            for table_pair in self._table_transfer_set:
                truncate = (True if self._options.truncate and table_pair.dest in self._dest_tables else False)
                drop = (True if self._options.drop and table_pair.dest in self._dest_tables else False)

                cmd = GpTransferCommand(
                    'transfer of %s' % table_pair.source,
                    self._options.source_host,
                    self._options.source_port,
                    self._options.source_user,
                    self._options.dest_host,
                    self._options.dest_port,
                    self._options.dest_user,
                    table_pair,
                    True if self._options.full else
                    table_pair.dest in self._dest_tables,
                    truncate,
                    self._options.analyze,
                    drop,
                    self._fast_mode,
                    self._options.exclusive_lock,
                    self._options.schema_only,
                    self._work_dir,
                    self._host_map,
                    self._src_config,
                    self._options.sub_batch_size,
                    self._options.base_port,
                    self._options.last_port,
                    self._gpfdist_instance_count,
                    gpfdist_verbosity,
                    self._options.max_line_length,
                    self._options.timeout,
                    self._options.wait_time,
                    self._options.delimiter,
                    self._options.validator,
                    self._options.format,
                    self._options.quote,
                    self._table_transfer_set_total
                )
                self._pool.addCommand(cmd)
                cmds_queued += 1

                (successful_commands, failed_commands) = wait_for_pool(self._pool,
                                                                       max_queued,
                                                                       cmds_queued == len(self._table_transfer_set))
                for failed_cmd in failed_commands:
                    failed_tables.append(failed_cmd.get_table_pair().source)

            if len(failed_tables) > 0:
                with open(GPTRANSFER_FAILED_TABLES_FILE, 'w') as failed_file:
                    for table in failed_tables:
                        failed_file.write("%s%s%s%s%s\n" %
                                         (table.database, SCHEMA_DELIMITER, table.schema, SCHEMA_DELIMITER, table.table))
                logger.warn(
                    '%s tables failed to transfer.  A list of these tables', len(failed_tables))
                logger.warn(
                    'has been written to the file %s', GPTRANSFER_FAILED_TABLES_FILE)
                logger.warn(
                    'This file can be used with the -f option to continue')
                logger.warn('the data transfer.')
                return 1
            else:
                if self._validator_class and not self._options.schema_only:
                    self._final_count_validation()
        else:
            # dry run
            self._dry_run()

        return 0

    def setup(self):
        """
        Creates work directories, created needed databases, tables, etc., and
        initializes the validator if needed.
        """
        # create base pipe directory on each source system
        logger.info("Creating work directory '%s'" % self._work_dir)

        for host in self._host_map.keys():
            address = iter(self._host_map[host]).next()
            cmd = MakeDirectory('Create work directory on %s' % host,
                                self._work_dir, REMOTE, address)
            self._pool.addCommand(cmd)

        cmd = MakeDirectory('Create work directory',
                            self._work_dir, REMOTE, self._options.source_host)
        self._pool.addCommand(cmd)

        cmd = MakeDirectory('Create work directory',
                            self._work_dir, REMOTE, self._options.dest_host)
        self._pool.addCommand(cmd)

        self._pool.join()
        try:
            self._pool.check_results()
        except ExecutionError, e:
            logger.fatal("Unable to create work directory: %s" % e)
            os._exit(2)

        # if transferring entire system, dumpall and execute on destination
        if self._options.full:
            logger.info('Restoring full schema...')
            dump_schema_cmd = GpSchemaDump('dump full schema',
                                           '127.0.0.1',
                                           self._options.source_port,
                                           self._options.source_user,
                                           ctxt=REMOTE,
                                           remoteHost=self._options.source_host)
            dump_schema_cmd.run()
            res = dump_schema_cmd.get_results()
            if not res or not res.wasSuccessful():
                raise ExecutionError('Error Executing Command: ', dump_schema_cmd)
            schema_sql = dump_schema_cmd.get_schema_sql()

            schema_filename = os.path.join(self._work_dir, 'full_schema.sql')
            with open(schema_filename, 'w') as full_schema_file:
                full_schema_file.write(schema_sql)

            psql_cmd = PsqlFile('Create full schema',
                                self._options.dest_host,
                                self._options.dest_port,
                                self._options.dest_user,
                                'template1',
                                schema_filename)
            self._pool.addCommand(psql_cmd)
            self._pool.join()
            self._pool.check_results()
            os.unlink(schema_filename)
            if os.path.isfile(schema_filename):
                logger.warn(
                    'Failed to remove schema file %s.', schema_filename)
                logger.warn('This file should be removed manually.')
        else:
            # Make sure databases exist on destination system
            url = DbURL(self._options.dest_host, self._options.dest_port,
                        'template1', self._options.dest_user)
            conn = connect(url)
            dest_databases = set([db[0] for db in getUserDatabaseList(conn)])
            conn.close()

            needed_databases = \
                set([p.dest.database for p in self._table_transfer_set])

            for database in needed_databases:
                if database not in dest_databases:
                    logger.info('Creating database %s...', database)
                    createdb_cmd = CreateDB('Create database %s' % database,
                                            self._options.dest_host,
                                            self._options.dest_port,
                                            self._options.dest_user,
                                            database)
                    self._pool.addCommand(createdb_cmd)
            self._pool.join()
            self._pool.check_results()

            # Make sure schemas exist
            needed_schemas = dict()
            for p in self._table_transfer_set:
                if p.dest.database not in needed_schemas:
                    needed_schemas[p.dest.database] = list()
                if p.dest.schema not in needed_schemas[p.dest.database]:
                    needed_schemas[p.dest.database].append(p.dest.schema)

            for (database, schemas) in needed_schemas.iteritems():
                url = DbURL(self._options.dest_host, self._options.dest_port,
                            database, self._options.dest_user)
                conn = connect(url)

                for schema in schemas:
                    logger.info('Creating schema %s in database %s...',
                                schema, database)
                    if not doesSchemaExist(conn, schema):
                        execSQL(conn, 'CREATE SCHEMA \"%s\"' % schema)
                conn.commit()
                conn.close()

        # setup the validator
        if self._options.validator:
            logger.info('Setting up validator...')
            validator = validator_factory.get_validator(
                self._options.validator)
            for db in set([p.dest.database for p in self._table_transfer_set]):
                url = DbURL(self._options.dest_host, self._options.dest_port,
                            db, self._options.dest_user)
                conn = connect(url)
                validator.setup(conn)
                conn.commit()
                conn.close()

            for db in set([p.source.database for p in self._table_transfer_set]):
                url = DbURL(
                    self._options.source_host, self._options.source_port,
                    db, self._options.source_user)
                conn = connect(url)
                validator.setup(conn)
                conn.commit()
                conn.close()

        # setup gptransfer schema
        for db in set([p.dest.database for p in self._table_transfer_set]):
            url = DbURL(self._options.dest_host, self._options.dest_port,
                        db, self._options.dest_user)
            conn = connect(url)
            cur = execSQL(conn, 'CREATE SCHEMA gptransfer')
            cur.close()
            conn.commit()
            conn.close()

        for db in set([p.source.database for p in self._table_transfer_set]):
            url = DbURL(self._options.source_host, self._options.source_port,
                        db, self._options.source_user)
            conn = connect(url)
            cur = execSQL(conn, 'CREATE SCHEMA gptransfer')
            cur.close()
            conn.commit()
            conn.close()

        self._cleanup_schemas = True

    def cleanup(self):
        """
        Cleans up the data transfer.
        """
        if not self._options.dry_run:
            # Remove base pipe directory on each source address
            logger.info('Removing work directories...')

            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()
                cmd = RemoveGlob('stopping gpfdist on %s' % address,
                                 os.path.join(self._work_dir, '*.pid' ), REMOTE, address)
                self._pool.addCommand(cmd)
            # We join() here so that all pid files are removed before directories,
            # ensuring that empty directories are removed
            self._pool.join()

            for host in self._host_map.keys():
                address = iter(self._host_map[host]).next()
                cmd = RemoveDirectory('Remove work directory',
                                      self._work_dir, REMOTE, address)
                self._pool.addCommand(cmd)

            cmd = RemoveDirectory('Remove work directory',
                                  self._work_dir, REMOTE, self._options.dest_host)
            self._pool.addCommand(cmd)

            cmd = RemoveDirectory('Remove work directory',
                                  self._work_dir, REMOTE, self._options.source_host)
            self._pool.addCommand(cmd)

            self._pool.join()

            if self._options.gpfdist_verbose or self._options.gpfdist_very_verbose:
                logger.warn('gpfdist logs are present in %s on all hosts in the source', self._work_dir)
                logger.warn('database system. These logs need to be manually cleaned up.')

            if self._options.validator:
                logger.info('Cleaning up validator...')
                validator = validator_factory.get_validator(
                    self._options.validator)
                # cleanup the validator
                for db in set([p.dest.database for p in self._table_transfer_set]):
                    url = DbURL(
                        self._options.dest_host, self._options.dest_port,
                        db, self._options.dest_user)
                    conn = connect(url)
                    validator.cleanup(conn)
                    conn.commit()
                    conn.close()

                for db in set([p.source.database for p in self._table_transfer_set]):
                    url = DbURL(
                        self._options.source_host, self._options.source_port,
                        db, self._options.source_user)
                    conn = connect(url)
                    validator.cleanup(conn)
                    conn.commit()
                    conn.close()

            # setup gptransfer schema
            if self._cleanup_schemas:
                for db in set([p.dest.database for p in self._table_transfer_set]):
                    url = DbURL(self._options.dest_host, self._options.dest_port,
                                db, self._options.dest_user)
                    conn = connect(url)
                    cur = execSQL(conn, 'DROP SCHEMA gptransfer CASCADE')
                    cur.close()
                    conn.commit()
                    conn.close()

                for db in set([p.source.database for p in self._table_transfer_set]):
                    url = DbURL(
                        self._options.source_host, self._options.source_port,
                        db, self._options.source_user)
                    conn = connect(url)
                    cur = execSQL(conn, 'DROP SCHEMA gptransfer CASCADE')
                    cur.close()
                    conn.commit()
                    conn.close()

            self._pool.haltWork()
            self._pool.joinWorkers()

            if running_gpfdists:
                logger.warn('Some gpfdist processes failed to stop on the source')
                logger.warn('system in a timely manner.  These processes may need to be')
                logger.warn('manually cleaned up.')

        logger.info("Finished.")
        os._exit(0)

    def _dry_run(self):
        """
        Prints to the screen what gptransfer will do without actually doing it.
        """
        # print batch info
        logger.info('Data transfer will use a batch size of %d',
                    self._options.batch_size)
        # print source info
        logger.info('The source GPDB system is %s:%d',
                    self._options.source_host,
                    self._options.source_port)
        # print dest info
        logger.info('The destination GPDB system is %s:%d',
                    self._options.dest_host,
                    self._options.dest_port)
        # print table transfer info
        action = ''
        if self._options.skip_existing:
            action = 'skipped'
        elif self._options.drop:
            action = 'dropped'
        elif self._options.truncate:
            action = 'truncated'

        if self._options.schema_only:
            if self._options.full:
                logger.info(
                    'The entire schema of the source system will be created')
                logger.info(
                    'on the destination system but no data will be transfered.')
            else:
                logger.info(
                    'The following tables will be created on the destination ')
                logger.info('system but no data will be transfered:')
                for pair in self._table_transfer_set:
                    logger.info('  %s', str(pair.dest))
        else:
            logger.info('The following tables will be transfered:')
            for pair in self._table_transfer_set:
                action_msg = ''
                if pair.dest in self._dest_tables:
                    action_msg = '(Destination table exists and will be %s)' % action
                logger.info('  %s -> %s %s',
                            str(pair.source),
                            str(pair.dest),
                            action_msg)

    def _pre_validate_options(self):
        """
        Validates the options passed in to the application.
        """

        logger.info('Validating options...')
        if self._options.batch_size > MAX_BATCH_SIZE \
                or self._options.batch_size < 1:
            raise ProgramArgumentValidationException('Invalid value for --batch-size.  '
                                                     '--batch-size must be greater than 0 and '
                                                     'less than %s' % MAX_BATCH_SIZE)

        if self._options.sub_batch_size > MAX_SUB_BATCH_SIZE \
                or self._options.sub_batch_size < 1:
            raise ProgramArgumentValidationException('Invalid value for --sub-batch-size.  '
                                                     'Must be greater than 0 and less than %s'
                                                     % MAX_SUB_BATCH_SIZE)

        if self._options.full and len(self._options.exclude_tables) != 0:
            raise ProgramArgumentValidationException('Cannot specify -T and --full options '
                                                     'together')

        if self._options.full and self._options.exclude_input_file:
            raise ProgramArgumentValidationException('Cannot specify -F and --full options '
                                                     'together')

        if self._options.input_file and not os.path.isfile(self._options.input_file):
            raise ProgramArgumentValidationException('Input file %s is not a valid file' % self._options.input_file)

        if (self._options.partition_transfer and self._options.partition_transfer_non_pt_target):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be '
                                                     'used with --partition-transfer-non-partition-target option')

        if (self._options.partition_transfer and not self._options.input_file):
            raise ProgramArgumentValidationException('--partition-transfer option must be '
                                                     'used with -f option')

        if (self._options.partition_transfer and self._options.databases):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be '
                                                     'used with -d option')

        if (self._options.partition_transfer and self._options.dest_database):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be '
                                                     'used with --dest-database option')

        if (self._options.partition_transfer and self._options.drop):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be used '
                                                     'with --drop option')

        if (self._options.partition_transfer and self._options.tables):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be used '
                                                     'with -t option')

        if (self._options.partition_transfer and self._options.schema_only):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be used '
                                                     'with --schema-only option, please create partition '
                                                     'table in other ways')

        if (self._options.partition_transfer and (self._options.exclude_tables or self._options.exclude_input_file)):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be used '
                                                     'with any exclude table option')

        if (self._options.partition_transfer and self._options.full):
            raise ProgramArgumentValidationException('--partition-transfer option cannot be '
                                                     'used with --full option')

        if (self._options.partition_transfer_non_pt_target and not self._options.input_file):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option must be '
                                                     'used with -f option')

        if (self._options.partition_transfer_non_pt_target and self._options.databases):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be '
                                                     'used with -d option')

        if (self._options.partition_transfer_non_pt_target and self._options.dest_database):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be '
                                                     'used with --dest-database option')

        if (self._options.partition_transfer_non_pt_target and self._options.drop):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be used '
                                                     'with --drop option')

        if (self._options.partition_transfer_non_pt_target and self._options.tables):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be used '
                                                     'with -t option')

        if (self._options.partition_transfer_non_pt_target and self._options.schema_only):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be used '
                                                     'with --schema-only option')

        if (self._options.partition_transfer_non_pt_target and (self._options.exclude_tables or self._options.exclude_input_file)):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be used '
                                                     'with any exclude table option')

        if (self._options.partition_transfer_non_pt_target and self._options.full):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be '
                                                     'used with --full option')

        if (self._options.partition_transfer_non_pt_target and self._options.validator):
            raise ProgramArgumentValidationException('--partition-transfer-non-partition-target option cannot be '
                                                     'used with --validate option')

        if self._options.truncate and self._options.partition_transfer_non_pt_target:
            raise ProgramArgumentValidationException('--truncate is not allowed with option --partition-transfer-non-partition-target')

        if self._options.full and (self._options.truncate or self._options.drop or
                                       self._options.skip_existing):
            raise ProgramArgumentValidationException('Cannot specify --drop, --truncate or '
                                                     '--skip-existing with --full option')

        if self._options.truncate and self._options.skip_existing:
            raise ProgramArgumentValidationException('Cannot specify --truncate and '
                                                     '--skip-existing together')

        if self._options.truncate and self._options.drop:
            raise ProgramArgumentValidationException('Cannot specify --truncate and '
                                                     '--drop together')

        if self._options.skip_existing and self._options.drop:
            raise ProgramArgumentValidationException('Cannot specify --drop and '
                                                     '--skip-existing together')

        if self._options.schema_only and self._options.truncate:
            raise ProgramArgumentValidationException('Cannot specify --schema-only and '
                                                     '--truncate together')

        if not self._options.source_map_file and \
                        self._options.source_host != self._options.dest_host and \
                        self._options.source_port != self._options.dest_port:
            raise ProgramArgumentValidationException(
                '--source-addresses-file option is required')

        if self._options.source_map_file and not os.path.isfile(self._options.source_map_file):
            raise ProgramArgumentValidationException('File %s does not exist'
                                                     % self._options.source_map_file)

        if not self._options.source_host:
            raise ProgramArgumentValidationException('--source-host option is required.')

        if not self._options.full and len(self._options.tables) == 0 \
                and len(self._options.databases) == 0 \
                and not self._options.input_file:
            raise ProgramArgumentValidationException('One of --full, -f, -t, or -d must be specified')

        if sum(bool(option) for option in (self._options.full,
                                           self._options.databases,
                                           self._options.tables,
                                           self._options.input_file)) != 1:
            raise ProgramArgumentValidationException('Only one of --full, -f, -t, or -d can be specified')

        if self._options.base_port <= 1024 or self._options.base_port > 65535:
            raise ProgramArgumentValidationException('Invalid --base-port option')

        if self._options.last_port == -1:
            self._options.last_port = self._options.base_port + 1000

        if self._options.last_port > 65535:
            logger.warn('Invalid --last-port value.  Setting last port to 65535')
            self._options.last_port = 65535

        if self._options.base_port >= self._options.last_port:
            self._options.last_port = self._options.base_port + 1000
            if self._options.last_port > 65535:
                raise ProgramArgumentValidationException('--base-port value causes last port to be invalid.')
            logger.warn(
                '--last-port value is less than --base-port.  Setting last port to %d' % (self._options.last_port))

        if self._options.max_line_length < MIN_GPFDIST_MAX_LINE_LENGTH or \
                        self._options.max_line_length > MAX_GPFDIST_MAX_LINE_LENGTH:
            raise ProgramArgumentValidationException('Invalid --max-line-length option.  Value must be '
                                                     'between %d and %d' % (MIN_GPFDIST_MAX_LINE_LENGTH,
                                                                            MAX_GPFDIST_MAX_LINE_LENGTH))

        if self._options.timeout < MIN_GPFDIST_TIMEOUT or \
                        self._options.timeout > MAX_GPFDIST_TIMEOUT:
            raise ProgramArgumentValidationException('Invalid --timeout option.  Value must be between '
                                                     '%d and %d' % (MIN_GPFDIST_TIMEOUT, MAX_GPFDIST_TIMEOUT))

        if self._options.wait_time < 0:
            raise ProgramArgumentValidationException('--wait-time must be greater than 0')

        self._gphome = get_gphome()
        if not FileDirExists.remote('remote gphome check',
                                    self._options.source_host, self._gphome):
            raise ProgramArgumentValidationException('GPHOME directory does not exist on %s'
                                                     % self._options.source_host)

        if self._options.validator and self._options.validator not in \
                validator_factory.get_available_validators():
            raise ProgramArgumentValidationException(
                'Unknown validator %s' % self._options.validator)

        if self._options.source_host == self._options.dest_host and \
                        self._options.source_port == self._options.dest_port and \
                not self._options.dest_database and \
                not self._options.partition_transfer:
            raise ProgramArgumentValidationException('Source and destination systems cannot be '
                                                     'the same unless --dest-database or --partition-transfer option is '
                                                     'set')

        if self._options.full and self._options.dest_database:
            raise ProgramArgumentValidationException('--dest-database option cannot be used with'
                                                     ' the --full option')

        if (self._options.delimiter[0] == "\\" and len(self._options.delimiter) != 4) and \
                (len(self._options.delimiter) != 1):
            raise ProgramArgumentValidationException(
                'Invalid --delimiter.  Must be single character')

        if self._options.delimiter != "," and self._options.format.lower() == 'csv':
            raise ProgramArgumentValidationException(
                'Invalid --delimiter. Delimiter %s is not allowed in CSV(default) format. Specify --format=TEXT' % self._options.delimiter)

        if self._options.delimiter == ',' and self._options.format.lower() == 'text':
            raise ProgramArgumentValidationException(
                'Invalid --delimiter. Default delimiter "," is not allowed in TEXT format. Specify --delimiter option for TEXT')

        if self._options.gpfdist_verbose and self._options.gpfdist_very_verbose:
            raise ProgramArgumentValidationException('--gpfdist-verbose option cannot be used with the --gpfdist-very-verbose option')

    def _post_validate_options(self):
        """
        Validates the options passed in to the application.
        """

        if self._options.full and self._dest_databases != None and len(self._dest_databases) != 0:
            raise ProgramArgumentValidationException('--full option specified but databases exist '
                                                     'in destination system')

        if self._gpfdist_instance_count == 0:
            raise Exception('Destination system is too small')

        if (self._options.exclude_tables or self._options.exclude_input_file) and not self._excluding_table:
            logger.warning('Found no tables to exclude from transfer table list')
            sys.exit(0)

        if self._options.validator == 'sha256':
            cmdStr = 'test -e `pg_config  --pkglibdir`/pgcrypto.so'
            cmd = Command('test pgcrypto on source', cmdStr, REMOTE, self._options.source_host)
            cmd.run()
            if cmd.get_results().rc != 0:
                raise Exception('Source system must have pgcrypto installed when using sha256 validator')

            cmd = Command('test pgcrypto on destination', cmdStr, REMOTE, self._options.dest_host)
            cmd.run()
            if cmd.get_results().rc != 0:
                raise Exception('Destination system must have pgcrypto installed when using sha256 validator')

    def _final_count_validation(self):
        logger.info('Running final table row count validation on destination tables...')

        # build a hash of dest tables
        # add the count of tables into the hash
        # remove # of tables from source to the hash of dest table
        #
        dest_table_dict = {}
        for table_pair in self._table_transfer_set:
            dest_conn = None
            src_conn = None
            try:
                if not str(table_pair.dest) in dest_table_dict:
                    dest_table_dict[str(table_pair.dest)] = 0
                    url = DbURL(self._options.dest_host, self._options.dest_port, table_pair.dest.database,
                                self._options.dest_user)
                    dest_conn = connect(url)

                url = DbURL(self._options.source_host, self._options.source_port, table_pair.source.database,
                            self._options.source_user)
                src_conn = connect(url)
                validator = self._validator_class(self._work_dir,
                                                  table_pair,
                                                  src_conn,
                                                  dest_conn)
                if dest_conn:
                    validator.init_dest_dict(dest_table_dict, table_pair)
                validator.accumulate(dest_table_dict, table_pair)
            except Exception, e:
                raise Exception("Final count validation failed for %s" % e)
            finally:
                if src_conn:
                    src_conn.close()
                if dest_conn:
                    dest_conn.close()

        for dest_table, validation_count in dest_table_dict.items():
            before_transfer_row_count = self.before_dest_tables_dict[dest_table] \
                if self.before_dest_tables_dict else 0
            if (validation_count - before_transfer_row_count) != 0:
                if self._options.partition_transfer or self._options.partition_transfer_non_pt_target:
                    logger.warning('Validation failed for %s', dest_table)
                else:
                    logger.error('Validation failed for %s', dest_table)
            else:
                logger.info('Validation of %s successful', dest_table)

    def get_dest_row_count_before_transfer(self):
        before_dest_tables_dict = {}
        for table_pair in self._table_transfer_set:
            dest_conn = None
            try:
                url = DbURL(self._options.dest_host, self._options.dest_port, table_pair.dest.database,
                            self._options.dest_user)
                dest_conn = connect(url)
                schemaname = escapeDoubleQuoteInSQLString(table_pair.dest.schema)
                tablename = escapeDoubleQuoteInSQLString(table_pair.dest.table)
                dest_sql = "SELECT count(*) FROM %s.%s" % (schemaname, tablename)
                dest_res = execSQLForSingletonRow(dest_conn, dest_sql)[0]
                before_dest_tables_dict[str(table_pair.dest)] = dest_res
            except Exception, e:
                raise Exception("Failed to get table row count before transfer: %s" % e)
            finally:
                if dest_conn:
                    dest_conn.close()
        return before_dest_tables_dict

    # Validate if destination system is missing required filespaces
    def _validate_filespace(self, source_host, source_port, source_user, dest_host, dest_port, dest_user, full):
        # Get all the filespaces of destination system
        list_dest_fs = self._get_destination_filespaces(dest_host, dest_port, 'template1', dest_user)

        # Check if source filespace exists in destination system
        if full:
            # For transferring in full mode
            source_fs_sql = """SELECT fsname FROM pg_catalog.pg_filespace"""
            self._check_filespace(source_host, source_port, 'template1', source_user, list_dest_fs, source_fs_sql)
        else:
            # For transferring in table mode
            template_source_fs_sql = """SELECT fsname FROM pg_catalog.pg_filespace WHERE oid IN
                                        (SELECT spcfsoid FROM pg_catalog.pg_tablespace
                                        WHERE oid IN (SELECT dattablespace FROM pg_catalog.pg_database WHERE datname = '%s')
                                        OR oid IN (SELECT reltablespace FROM pg_catalog.pg_class WHERE relname = '%s'));"""
            for table_pair in self._table_transfer_set:
                dbname = pg.escape_string(table_pair.source.database)
                tablename = pg.escape_string(table_pair.source.table)
                source_fs_sql = template_source_fs_sql % (dbname, tablename)
                self._check_filespace(source_host, source_port, table_pair.source.database, source_user, list_dest_fs,
                                      source_fs_sql)

    def _get_destination_filespaces(self, dest_host, dest_port, database, dest_user):
        list_dest_fs = []
        dest_fs_sql = """SELECT fsname FROM pg_catalog.pg_filespace"""
        with dbconn.connect(dbconn.DbURL(dest_host, dest_port, database, dest_user)) as dest_conn:
            dest_cur = execSQL(dest_conn, dest_fs_sql)
            for row in dest_cur:
                list_dest_fs.append(row[0])
            dest_cur.close()
        return list_dest_fs

    def _check_filespace(self, source_host, source_port, database, source_user, list_dest_fs, source_fs_sql):
        with dbconn.connect(dbconn.DbURL(source_host, source_port, database, source_user)) as source_conn:
            source_cur = execSQL(source_conn, source_fs_sql)
            for row in source_cur:
                fs = row[0]
                if fs not in list_dest_fs:
                    source_cur.close()
                    raise Exception(
                        "Filespace '%s' does not exist on destination. Please create the filespace to run gptransfer" % fs)
            source_cur.close()

    def _get_configurations(self):
        """
        Retrieves the configurations of the source and destination GPDB systems.
        """

        logger.info('Retrieving configuration of source Greenplum Database...')
        url = DbURL(self._options.source_host,
                    self._options.source_port, 'template1',
                    self._options.source_user)
        self._src_config = GpArray.initFromCatalog(url)
        logger.info('Retrieving configuration of destination Greenplum '
                    'Database...')
        url = DbURL(self._options.dest_host, self._options.dest_port,
                    'template1', self._options.dest_user)
        self._dest_config = GpArray.initFromCatalog(url)

    def _validate_schema_not_exists(self):
        """
        Check to see if we have any gptransfer schemas on either system
        """
        logger.info('Checking for gptransfer schemas...')

        schemas_exists_on_source = False
        schemas_exists_on_destination = False
        if self._options.full:
            if schema_exists_on_system(self._options.source_host,
                                       self._options.source_port,
                                       self._options.source_user,
                                       'gptransfer'):
                schemas_exists_on_source = True

            if schema_exists_on_system(self._options.dest_host,
                                       self._options.dest_port,
                                       self._options.dest_user,
                                       'gptransfer'):
                schemas_exists_on_destination = True

        else:
            if schema_exists_on_system(self._options.source_host,
                                       self._options.source_port,
                                       self._options.source_user,
                                       'gptransfer',
                                       self._src_databases):
                schemas_exists_on_source = True

            if schema_exists_on_system(self._options.dest_host,
                                       self._options.dest_port,
                                       self._options.dest_user,
                                       'gptransfer',
                                       [db for db in self._dest_databases if db in self._all_dest_databases]):
                schemas_exists_on_destination = True

        if schemas_exists_on_source:
            logger.warning('The gptransfer schema already exists on the source system.')
            logger.warning('This is likely due to a previous run on gptransfer')
            logger.warning('being forcefully terminated and not properly cleaned up.')
            logger.warning('Removing existing gptransfer schema on source system.')
            if self._options.full:
                drop_existing_schema_on_system(self._options.source_host,
                                               self._options.source_port,
                                               self._options.source_user,
                                               'gptransfer')
            else:
                drop_existing_schema_on_system(self._options.source_host,
                                               self._options.source_port,
                                               self._options.source_user,
                                               'gptransfer',
                                               self._src_databases)

        if schemas_exists_on_destination:
            logger.warning('The gptransfer schema already exists on destination system.')
            logger.warning('This is likely due to a previous run on gptransfer')
            logger.warning('being forcefully terminated and not properly cleaned up.')
            logger.warning('Removing existing gptransfer schema on destination system.')
            if self._options.full:
                drop_existing_schema_on_system(self._options.dest_host,
                                               self._options.dest_port,
                                               self._options.dest_user,
                                               'gptransfer')
            else:
                drop_existing_schema_on_system(self._options.dest_host,
                                               self._options.dest_port,
                                               self._options.dest_user,
                                               'gptransfer',
                                               [db for db in self._dest_databases if db in self._all_dest_databases])

    def _get_source_tables(self):
        source_tables = list()
        # --partition-transfer will have the exact database name for each destination table,
        # not support the regexp
        if (self._options.partition_transfer or self._options.partition_transfer_non_pt_target):

            self._src_dest_partition_table_mapping = dict()

            self._get_tables_from_partition_input_file()
            user_tables = get_user_tables(self._options.source_host,
                                          self._options.source_port,
                                          self._options.source_user,
                                          self._src_databases,
                                          False,
                                          (self._options.partition_transfer or self._options.partition_transfer_non_pt_target))

            # filter out non-exist source tables from source databases.
            # key in dict: self._src_dest_partition_table_mapping maps to index of list: source_tables
            key = 0
            for i in range(len(self._src_prt_tables)):
                tbl = self._src_prt_tables[i]
                for user_tbl in user_tables:
                    if tbl == str(user_tbl):
                        source_tables.append(user_tbl)
                        self._src_dest_partition_table_mapping[key] = self._dest_prt_tables[i]
                        key += 1
                        break

            del self._src_prt_tables[:]
            del self._dest_prt_tables[:]

            return source_tables

        if self._options.full:
            for src_table in get_user_tables(self._options.source_host,
                                             self._options.source_port,
                                             self._options.source_user,
                                             full=True,
                                             partition_transfer=False):
                if src_table.database != 'gpperfmon':
                    source_tables.append(src_table)

            self._src_databases = self._all_src_databases
            self._dest_databases = self._all_dest_databases

            if 'gpperfmon' in self._src_databases:
                logger.warn('Tables in database "gpperfmon" will not be transferred with full option')
                self._src_databases.remove('gpperfmon')
        else:
            # get all source databases for both full transfer and table transfer
            databases_full_transfer = list()
            databases_table_transfer = list()

            # for databases to transfer with option -d
            databases = list()

            for database in self._options.databases:
                databases.append(_process_regexp(database))
            databases_full_transfer = get_databases_by_regexp(
                self._options.source_host,
                self._options.source_port,
                self._options.source_user,
                databases, True)

            # add all tables of source databases
            for table in get_user_tables(self._options.source_host,
                                         self._options.source_port,
                                         self._options.source_user,
                                         databases_full_transfer,
                                         False,
                                         (self._options.partition_transfer or self._options.partition_transfer_non_pt_target)):
                source_tables.append(table)

            del databases[:]
            tables = list()
            # add specified tables and parse wildcard of table names
            if self._options.tables:
                self._get_tables_from_option(databases, tables)

            # gptransfer with -f option
            # add all tables from input file and parse wildcard of table names
            if self._options.input_file:
                self._get_tables_from_input_file(databases, tables)

            databases_table_transfer = get_databases_by_regexp(
                self._options.source_host,
                self._options.source_port,
                self._options.source_user,
                databases, True)

            # get all source databases to transfer tables from
            self._src_databases = databases_full_transfer + databases_table_transfer

            # get all destination databases
            if self._options.dest_database:
                self._dest_databases = [self._options.dest_database]
            else:
                self._dest_databases = [db for db in self._all_dest_databases if db in self._src_databases]

            # get all the tables of specified databases from the source GPDB system
            logger.info("Retrieving source tables...")
            user_tables = get_user_tables(self._options.source_host,
                                          self._options.source_port,
                                          self._options.source_user,
                                          databases_table_transfer,
                                          False,
                                          (self._options.partition_transfer or self._options.partition_transfer_non_pt_target))

            # Filter down to only the tables needed to added
            seen = set()
            for (db_, schema_, table_) in tables:
                for user_table in user_tables:
                    database, schema, table = user_table.database, user_table.schema, user_table.table

                    try:
                        if (re.match(r'%s' % db_, database, re.L | re.U) and
                                re.match(r'%s' % schema_, schema, re.L | re.U) and
                                re.match(r'%s' % table_, table, re.L | re.U)):
                            if user_table not in seen:
                                seen.add(user_table)
                                source_tables.append(user_table)

                    except Exception, e:
                        raise Exception('Error matching regular expression, %s with database name %s'
                                        ' or %s with schema name %s, or %s with table name %s\n%s' %
                                        (db_, database, schema_, schema, table_, table, e))

        return self.filter_tables(source_tables, self._get_exclude_tables())

    def _process_tables_details(self, table_info, databases, tables):
        (database, schema, table) = self._process_full_qualified_regexp_name(table_info)
        if database not in databases:
            databases.append(database)
        if (database, schema, table) not in tables:
            tables.append((database, schema, table))

    def _get_tables_from_input_file(self, databases, tables):
        """
        if users provide an input file, add all tables specified.
        """
        with open(self._options.input_file) as input_file:
            for line in input_file:
                line = line.strip()
                if line.startswith('#'):
                    continue
                if ',' in line:
                    raise Exception("Destination tables (comma separated) are only allowed for partition tables. "
                                    "line: %s" % line)
                self._process_tables_details(line, databases, tables)

    def _get_tables_from_partition_input_file(self):
        """
        read comma ',' separated partition table pair: <source, destination>,
        populate all source-destination pairs with their exact name(does not expect regexp).
        """
        self._src_prt_tables = []
        self._dest_prt_tables = []
        self._dest_databases = []
        self._src_databases = []
        fqn_format = '<db_name>.<schema_name>.<table_name>'
        logger.info('Reading partition table transfer pairs from input file %s', self._options.input_file)
        with open(self._options.input_file) as input_file:
            for line in input_file:
                line = line.strip()
                if line.startswith('#') or line == '':
                    logger.debug('Skipping blank lines and commented lines beginning with # (pound sign)')
                    continue
                if ',' in line:
                    if line.count(',') > 1:
                        raise Exception('Wrong format found for table transfer pair "%s", use single comma '
                                        'to separate source and destination table' % line)
                    src_table, dest_table = line.split(',')
                else:
                    src_table = dest_table = line

                # populate the mapping between source and destination table pair
                src_table, dest_table = src_table.strip(), dest_table.strip()

                if src_table == '':
                    raise Exception('Cannot specify empty source table name')

                if dest_table == '':
                    raise Exception('Cannot specify empty destination table name')

                # gptransfer does not support special char '.' in db name, schema name or table name
                if src_table.count(SCHEMA_DELIMITER) != 2:
                    raise Exception('Source table name "%s" isn\'t fully qualified format: %s' %
                                    (src_table, fqn_format))

                if dest_table.count(SCHEMA_DELIMITER) != 2:
                    raise Exception('Destination table name "%s" isn\'t fully qualified format: %s' %
                                    (dest_table, fqn_format))

                if (src_table == dest_table and self._options.source_host == self._options.dest_host and
                            self._options.source_port == self._options.dest_port):
                    raise Exception('Cannot transfer between same partition table %s' % src_table)

                if src_table in self._src_prt_tables:
                    indexes = [i for i, tbl in enumerate(self._src_prt_tables) if tbl == src_table]
                    for i in indexes:
                        if self._dest_prt_tables[i] == dest_table:
                            logger.warning(
                                'Duplicate entries found, source partition table %s, destination partition table %s ' %
                                (src_table, dest_table))
                    continue

                logger.debug(
                    'Getting source partition table "%s", destination partition table "%s"' % (src_table, dest_table))
                self._src_prt_tables.append(src_table)
                self._dest_prt_tables.append(dest_table)

                src_db = split_fqn(src_table)[0]
                if src_db not in self._src_databases:
                    self._src_databases.append(src_db)
                dest_db = split_fqn(dest_table)[0]
                if dest_db not in self._dest_databases:
                    self._dest_databases.append(dest_db)

    def _get_tables_from_option(self, databases, tables):
        """
        If use provided tables to transfer from command line, add all of them
        """
        for fqn in self._options.tables:
            self._process_tables_details(fqn, databases, tables)

    def _process_full_qualified_regexp_name(self, full_qualified_name):
        """
        split the full qualified table name(possibly in regexp format), and
        return (database, schema, table) tuple
        """
        (database, schema, table) = split_fqn(full_qualified_name)

        database = _process_regexp(database)
        schema = _process_regexp(schema)
        table = _process_regexp(table)

        return database, schema, table

    def _get_exclude_tables(self):
        """
        get all tables to exclude from transfer table list
        """

        exclude_tables = set()

        # excluded tables
        for fqn in self._options.exclude_tables:
            (database, schema, table) = split_fqn(fqn)
            database = _process_regexp(database)
            schema = _process_regexp(schema)
            table = _process_regexp(table)
            exclude_tables.add((database, schema, table))

        # add all tables in exclude input file and parse wildcard of table names
        if self._options.exclude_input_file:
            with open(self._options.exclude_input_file) as input_file:
                for line in input_file:
                    line = line.strip()
                    if line.startswith('#'):
                        continue

                    (database, schema, table) = split_fqn(line)
                    database = _process_regexp(database)
                    schema = _process_regexp(schema)
                    table = _process_regexp(table)

                    exclude_tables.add((database, schema, table))

        return exclude_tables

    def filter_tables(self, source_tables, exclude_tables):
        """
        filter out the source tables from transferring
        """
        if (source_tables is None or len(source_tables) == 0 or
                    exclude_tables is None or len(exclude_tables) == 0):
            return source_tables

        tables_to_transfer = list()
        for src_table in source_tables:
            database, schema, table = src_table.database, src_table.schema, src_table.table

            is_exclude_table_match = False
            for (db_, schema_, table_) in exclude_tables:
                try:
                    if (re.match(r'%s' % db_, database, re.L | re.U) and
                            re.match(r'%s' % schema_, schema, re.L | re.U) and
                            re.match(r'%s' % table_, table, re.L | re.U)):
                        is_exclude_table_match = True
                        break
                except Exception, e:
                    raise Exception('Error matching regular expression, %s with database name %s'
                                    ' or %s with schema name %s, or %s with table name %s\n%s' %
                                    (db_, database, schema_, schema, table_, table, e))

            if is_exclude_table_match:
                logger.debug("Excluding table %s", src_table)
            else:
                logger.debug('Adding table %s', src_table)
                tables_to_transfer.append(src_table)

        if len(tables_to_transfer) == len(source_tables):
            self._excluding_table = False

        return tables_to_transfer

    def _build_table_transfer_list(self):
        """
        Builds up the list of tables to transfer.
        """

        logger.info('Building list of source tables to transfer...')
        table_pairs = list()
        seen = set()

        for i in range(len(self._src_tables)):

            src_table = self._src_tables[i]

            if self._options.partition_transfer or self._options.partition_transfer_non_pt_target:
                dest_table = self._src_dest_partition_table_mapping[i]
                dest_db, dest_schema, dest_tbl = split_fqn(dest_table)
                table_pair = GpTransferTablePair(src_table,
                                                 GpTransferTable(dest_db,
                                                                 dest_schema,
                                                                 dest_tbl,
                                                                 src_table.external))

            else:
                dest_database = src_table.database
                if self._options.dest_database and not self._options.full:
                    dest_database = self._options.dest_database

                table_pair = GpTransferTablePair(src_table,
                                                 GpTransferTable(dest_database,
                                                                 src_table.schema,
                                                                 src_table.table,
                                                                 src_table.external))
            if table_pair not in seen:
                seen.add(table_pair)
                table_pairs.append(table_pair)

        if self._src_dest_partition_table_mapping:
            self._src_dest_partition_table_mapping.clear()

        logger.info('Number of tables to transfer: %s', len(table_pairs))

        return table_pairs

    def _validate_table_transfer_set(self):
        """
        Validates the table transfers, checking if tables exist, etc.
        """

        logger.info('Validating transfer table set...')

        # make sure we don't collide on dest table names
        non_supported_table_name_list = list()
        target_tables = set()
        remove_list = list()
        truncate_list = list()
        drop_list = list()

        for table_pair in self._table_transfer_set:
            if table_pair.dest.external:
                continue

            if table_pair.dest in self._dest_tables and self._options.skip_existing:
                remove_list.append(table_pair)
                continue

            if not table_pair.source.is_identifier_supported():
                non_supported_table_name_list.append(table_pair.source)

            if not table_pair.dest.is_identifier_supported():
                non_supported_table_name_list.append(table_pair.dest)

            if table_pair.dest in target_tables:
                if self._options.partition_transfer_non_pt_target:
                    self.multiple_same_target_set.add(table_pair.dest)
                else:
                    raise Exception('Multiple tables map to %s.  Remove one of '
                                    'the tables ' % table_pair.dest
                                    + 'from the list or do not use the '
                                      '--dest-database option.')
            target_tables.add(table_pair.dest)

            self._create_special_case_lists(table_pair, truncate_list, drop_list)

        self._report_non_supported_names(non_supported_table_name_list)

        [self._table_transfer_set.remove(table) for table in remove_list]

        if len(self._table_transfer_set) == 0 and not self._options.dry_run:
            logger.info('Found no tables to transfer.')
            sys.exit(0)

        self._prompt_interactive_mode(drop_list, truncate_list)

    def _report_non_supported_names(self, non_supported_table_name_list):
        if len(non_supported_table_name_list) > 0:
            with open(GPTRANSFER_FAILED_TABLES_FILE, 'w') as failed_file:
                for table_obj in non_supported_table_name_list:
                    failed_file.write("%s%s%s%s%s\n" %
                                      (table_obj.database, SCHEMA_DELIMITER,
                                       table_obj.schema, SCHEMA_DELIMITER, table_obj.table))
            logger.error('Found unsupported identifiers')
            logger.error('Valid identifiers must start with a-z, or A-Z or underscore followed by a-z, '
                         'or A-Z, or 0-9, or underscore')
            logger.error('A list of these tables has been written to the file %s', GPTRANSFER_FAILED_TABLES_FILE)
            sys.exit(1)

    def _prompt_interactive_mode(self, drop_list, truncate_list):
        if self._options.interactive and not self._options.dry_run:
            ask = False
            if self._options.truncate and len(truncate_list) > 0:
                ask = True
                logger.info(
                    'The following tables on the destination system will be truncated:')
                for table in truncate_list:
                    logger.info('   %s', str(table))

            elif self._options.drop and len(drop_list) > 0:
                ask = True
                logger.info(
                    'The following tables on the destination system will be dropped:')
                for table in drop_list:
                    logger.info('   %s', str(table))

            if ask:
                cont = ask_yesno(None, 'Do you want to continue?', 'N')
                if not cont:
                    raise Exception('Canceled')

    def _create_special_case_lists(self, table_pair, truncate_list, drop_list):
        if table_pair.dest in self._dest_tables:
            if not self._options.truncate and not self._options.drop and not \
                    self._options.partition_transfer and not self._options.partition_transfer_non_pt_target:
                raise Exception('Table %s exists in database %s.'
                                % (table_pair.dest,
                                   table_pair.dest.database))
            else:
                if self._options.truncate:
                    truncate_list.append(table_pair.dest)
                elif not self._options.partition_transfer:
                    drop_list.append(table_pair.dest)

    def _validate_partition_table_transfer_set(self):
        """
        validate that the table pair to transfer are actually leaf partition tables.
        validate that the pair have same layout and partition criteria.
        validate that the destination tables are non partition table for --partition-transfer-non-partition-target
        """

        logger.info('Validating partition table transfer set...')

        if self._options.partition_transfer_non_pt_target:
            self._validate_partition_sources_to_a_common_dest_must_have_same_parent()

        for table_pair in self._table_transfer_set:
            if table_pair.dest not in self._dest_tables:
                current_option = "--partition-transfer" if self._options.partition_transfer else "--partition-transfer-non-partition-target"
                looking_for = "leaf partitions" if self._options.partition_transfer else "non-partition tables"
                error_message = 'Table %s does not exist in destination database ' \
                                'when transferring from partition tables ' \
                                '(filtering for destination %s because of option "%s")' % \
                                (str(table_pair.dest), looking_for, current_option)
                raise Exception(error_message)

            self._check_leaf_partition_set(table_pair)

            if not self._has_same_column_types(table_pair):
                raise Exception('Source partition table %s has different column layout or types from '
                                'destination table %s' % (str(table_pair.source), str(table_pair.dest)))

            if self._options.partition_transfer:
                if not self._has_same_partition_criteria(table_pair):
                    raise Exception('Source partition table %s has different partition criteria from '
                                    'destination table %s' % (str(table_pair.source), str(table_pair.dest)))

    def _validate_partition_sources_to_a_common_dest_must_have_same_parent(self):
        diff_parented_sources_for_common_dest = self._get_table_pair_with_different_parent_for_common_destinations()
        if len(diff_parented_sources_for_common_dest) > 0:
            sources = diff_parented_sources_for_common_dest[0]["sources"]
            dest = diff_parented_sources_for_common_dest[0]["dest"]
            raise Exception("partition sources: %s when transferred to the "
                            "same destination: table %s , must share the same parent" % (sources, dest))

    def _get_table_pair_with_different_parent_for_common_destinations(self):
        mismatch = list()
        # assemble all common destinations and validate
        all_destinations = dict()
        for table_pair in self._table_transfer_set:
            all_destinations.setdefault(table_pair.dest, list()).append(table_pair.source)

        # when a common destination has multiple sources, validate that all come from same parent partition
        for dest, source_list in all_destinations.iteritems():
            if len(source_list) > 1:
                parent_schema_previous = None
                parent_table_previous = None
                for source in source_list:
                    parent_schema, parent_table = self._get_parentpartable(source.database,
                                                                           source.schema,
                                                                           source.table,
                                                                           self._options.source_host,
                                                                           self._options.source_port,
                                                                           self._options.source_user)

                    if parent_schema_previous and parent_schema_previous != parent_schema:
                        sources = ""
                        for source in source_list:
                            sources += source.schema + "." + source.table + ", "
                        mismatch.append(dict(sources=sources, dest=dest.schema + "." + dest.table))
                    if parent_table_previous and parent_table_previous != parent_table:
                        sources = ""
                        for source in source_list:
                            sources += source.schema + "." + source.table + ", "
                        mismatch.append(dict(sources=sources, dest=dest.schema + "." + dest.table))
                    parent_table_previous = parent_table
                    parent_schema_previous = parent_schema

        return mismatch

    def _has_same_column_types(self, table_pair):
        """
        Columns at the same position of two partition tables need to be of same type
        """
        src_tbl_columns = self._get_table_columns(table_pair.source, self._options.source_host, self._options.source_port, self._options.source_user)
        dest_tbl_columns = self._get_table_columns(table_pair.dest, self._options.dest_host, self._options.dest_port, self._options.dest_user)
        return src_tbl_columns == dest_tbl_columns

    def _get_table_columns(self, tbl, host, port, user):
        get_columns_sql = ''' select is_nullable, data_type, character_maximum_length,
                             character_octet_length, numeric_precision, numeric_precision_radix, numeric_scale,
                             datetime_precision, interval_type, udt_name from information_schema.columns
                             where table_schema = '%s' and table_name = '%s' order by ordinal_position asc;''' % (pg.escape_string(tbl.schema), pg.escape_string(tbl.table))
        with dbconn.connect(dbconn.DbURL(host, port, tbl.database, user)) as conn:
            cursor = execSQL(conn, get_columns_sql)
        return list(cursor.fetchall())

    def _has_same_partition_levels(self, table_pair):
        logger.debug('Verifying that partition table transfer pair has same level of partition.')
        src_prt_levels = self._get_partition_levels(table_pair.source, self._options.source_host,
                                                    self._options.source_port, self._options.source_user)
        dest_prt_levels = self._get_partition_levels(table_pair.dest, self._options.dest_host,
                                                     self._options.dest_port, self._options.dest_user)
        if src_prt_levels != dest_prt_levels:
            return False
        return True

    def _get_partition_levels(self, tbl, host, port, user):
        get_max_partition_sql = ''' select max(p1.partitionlevel) from pg_partitions p1, pg_partitions p2
                                    where p1.schemaname = p2.schemaname and p1.tablename = p2.tablename and
                                    p2.partitionschemaname = '%s' and p2.partitiontablename = '%s';''' % (
        pg.escape_string(tbl.schema),
        pg.escape_string(tbl.table))
        with dbconn.connect(dbconn.DbURL(host, port, tbl.database, user)) as conn:
            res = execSQLForSingletonRow(conn, get_max_partition_sql)
            return res[0]

    def _has_same_partition_criteria(self, table_pair):
        # verify the current subpartition pair:
        logger.debug('Verifying that partition table transfer pair has same partition criteria')
        source_dest_info = 'source partition table %s and destination partition table %s' % (
        str(table_pair.source), str(table_pair.dest))
        if not self._has_same_partition_levels(table_pair):
            logger.error('Max level of partition is not same between %s' % source_dest_info)
            return False
        if not self._has_same_partition_type_and_key_columns(table_pair):
            logger.error('Partition type or key is different between %s' % source_dest_info)
            return False
        if not self._has_same_parent_partition_value(
                self._get_partition_levels(
                    table_pair.source,
                    self._options.source_host,
                    self._options.source_port,
                    self._options.source_user),
                table_pair.source.database,
                table_pair.source.schema,
                table_pair.source.table,
                table_pair.dest.database,
                table_pair.dest.schema,
                table_pair.dest.table):
            logger.error('Partition value is different in the partition hierarchy between %s' % source_dest_info)
            return False
        return True

    def _has_same_parent_partition_value(self, level, src_db, src_schema, src_table, dest_db, dest_schema, dest_table):
        if not self._has_same_partition_value(src_db, src_schema, src_table, dest_db, dest_schema, dest_table):
            return False
        if level == 0:
            return True
        src_schema, src_table = self._get_parentpartable(src_db, src_schema, src_table, self._options.source_host,
                                                         self._options.source_port, self._options.source_user)
        dest_schema, dest_table = self._get_parentpartable(dest_db, dest_schema, dest_table, self._options.dest_host,
                                                           self._options.dest_port, self._options.dest_user)
        result = self._has_same_parent_partition_value(level - 1, src_db, src_schema, src_table, dest_db, dest_schema,
                                                       dest_table)
        if not result:
            logger.error("Partitions have different parents at level: %d" % (level - 1))
        return result

    def _get_parentpartable(self, db, schema, table, host, port, user):
        """
        get the parent partition table, return its schema and table name
        """
        table_info = 'table %s.%s.%s, host %s' % (db, schema, table, host)
        logger.debug('Getting parent table information for %s' % (table_info))
        oid = self._get_oid(db, schema, table, host, port, user)
        get_parentpartable_sql = ''' select n.nspname, c.relname
                                 from pg_class c,
                                      pg_namespace n,
                                      pg_inherits i
                                 where i.inhparent = c.oid
                                       and c.relnamespace = n.oid
                                       and i.inhrelid = %s; ''' % oid
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            row = execSQLForSingletonRow(conn, get_parentpartable_sql)
            if not row:
                raise Exception('Found no parent table for %s' % table_info)
            return row[0], row[1]

    def _get_top_level_parent_table(self, db, schema, table, host, port, user):
        """
        get the top level parent partition table's name and its namespace
        """
        table_info = 'table %s.%s.%s, host %s' % (db, schema, table, host)
        logger.debug('Getting top level parent table information for %s' % (table_info))
        get_top_level_table_sql = ''' select schemaname, tablename from pg_catalog.pg_partitions
                                      where partitionschemaname = '%s'
                                            and partitiontablename = '%s' ''' % (
            pg.escape_string(schema),
            pg.escape_string(table))
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            row = execSQLForSingletonRow(conn, get_top_level_table_sql)
            if not row:
                raise Exception('Failed to find the top level table for %s' % table_info)
            return row[0], row[1]

    def _get_top_level_parent_table_oid(self, db, schema, table, host, port, user):
        root_schema, root_table = self._get_top_level_parent_table(db, schema, table, host, port, user)
        root_oid = self._get_oid(db, root_schema, root_table, host, port, user)
        return root_oid

    def _has_same_partition_type_and_key_columns(self, table_pair):
        source_dest_info = 'source table %s and destination table %s' % (str(table_pair.source), str(table_pair.dest))
        source_partition_column_info = self._get_partition_column_info(table_pair.source.database,
                                                                 table_pair.source.schema,
                                                                 table_pair.source.table,
                                                                 self._options.source_host,
                                                                 self._options.source_port,
                                                                 self._options.source_user)

        dest_partition_column_info = self._get_partition_column_info(table_pair.dest.database,
                                                                 table_pair.dest.schema,
                                                                 table_pair.dest.table,
                                                                 self._options.dest_host,
                                                                 self._options.dest_port,
                                                                 self._options.dest_user)

        source_ordinal_positions = self._get_ordinal_positions_list(table_pair.source.database,
                                                                 table_pair.source.schema,
                                                                 table_pair.source.table,
                                                                 self._options.source_host,
                                                                 self._options.source_port,
                                                                 self._options.source_user)

        dest_ordinal_positions = self._get_ordinal_positions_list(table_pair.dest.database,
                                                                 table_pair.dest.schema,
                                                                 table_pair.dest.table,
                                                                 self._options.dest_host,
                                                                 self._options.dest_port,
                                                                 self._options.dest_user)

        for level_key in source_partition_column_info:
            if source_partition_column_info[level_key]['parkind'] != dest_partition_column_info[level_key]['parkind']:
                logger.error('Partition type is different at level %s between %s' % (level_key, source_dest_info))
                return False
            if source_partition_column_info[level_key]['parnatts'] != dest_partition_column_info[level_key]['parnatts']:
                logger.error(
                    'Number of partition columns is different at level %s between %s' % (level_key, source_dest_info))
                return False

            if not self._has_same_paratts(
                                     source_partition_column_info[level_key]['paratts'],
                                     dest_partition_column_info[level_key]['paratts'],
                                     source_ordinal_positions,
                                     dest_ordinal_positions):
                logger.error('Partition column attributes are different at level %s between %s' %
                             (level_key, source_dest_info))
                return False
        return True

    def _has_same_paratts(self, source_paratts_str, dest_paratts_str, source_ordinal_positions, dest_ordinal_positions):
        # this handles multi column partitions by sorting them before comparison.
        # we compare the index of the column within the list of ordinal positions to account for cases where
        # actual ordinal positions may not match, but the order of the columns is the same (i.e. dropping and re-adding)
        source_paratts = [int(att.strip()) for att in source_paratts_str.split(' ')]
        dest_paratts = [int(att.strip()) for att in dest_paratts_str.split(' ')]

        source_and_dest_paratts = zip(sorted(source_paratts), sorted(dest_paratts))
        for source_paratt, dest_paratt in source_and_dest_paratts:
            if source_paratt not in source_ordinal_positions or dest_paratt not in dest_ordinal_positions or source_ordinal_positions.index(source_paratt) != dest_ordinal_positions.index(dest_paratt):
                return False
        return True

    def _get_ordinal_positions_list(self, db, schema, table, host, port, user):
        ordinal_positions_list = []
        ordinal_pos_sql = ''' select ordinal_position from information_schema.columns
                              where table_schema = '%s' and table_name = '%s'
                              order by ordinal_position ''' % (pg.escape_string(schema), pg.escape_string(table))
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            cursor = execSQL(conn, ordinal_pos_sql)
            for row in cursor:
                ordinal_positions_list.append(row[0])

        return ordinal_positions_list

    def _get_partition_column_info(self, db, schema, table, host, port, user):
        level_partition_info = defaultdict(dict)
        top_level_table_oid = self._get_top_level_parent_table_oid(db, schema, table, host, port, user)
        get_partition_colum = ''' select parkind, parlevel, parnatts, paratts from pg_partition
                                  where paristemplate = 'f'
                                  and parrelid = %s ''' % top_level_table_oid
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            cursor = execSQL(conn, get_partition_colum)
            for row in cursor:
                level = row[1]
                level_partition_info[level]['parkind'] = row[0]
                level_partition_info[level]['parnatts'] = row[2]
                level_partition_info[level]['paratts'] = row[3]
        return level_partition_info

    def _has_same_partition_value(self, src_db, src_schema, src_table, dest_db, dest_schema, dest_table):
        """
        Verify that the partition value is the same based on the partition type: [range, list]
        """
        source_partition_info = self._get_partition_info(src_db, src_schema, src_table, self._options.source_host,
                                                         self._options.source_port, self._options.source_user)
        dest_partition_info = self._get_partition_info(dest_db, dest_schema, dest_table, self._options.dest_host,
                                                       self._options.dest_port, self._options.dest_user)

        part_transfer_pair_msg = '''source partition table %s.%s.%s and
                                    destination partition table %s.%s.%s.''' % (src_db, src_schema,
                                                                                src_table, dest_db,
                                                                                dest_schema, dest_table)
        part_type = source_partition_info['partitiontype']
        if source_partition_info['parisdefault'] != dest_partition_info['parisdefault']:
            logger.error('One of the subpartition table is a default partition, %s' % part_transfer_pair_msg)
            return False

        if part_type == 'list':
            return self._has_same_list_partition_value(source_partition_info, dest_partition_info,
                                                       part_transfer_pair_msg)
        elif part_type == 'range':
            return self._has_same_range_partition_value(source_partition_info, dest_partition_info,
                                                        part_transfer_pair_msg)
        else:
            raise Exception('Unknown partitioning type %s found, %s' % (part_type, part_transfer_pair_msg))

    def _has_same_range_partition_value(self, source_partition_info, dest_partition_info, part_transfer_pair_msg):
        comparator = PartitionComparatorFactory().get(source_partition_info, dest_partition_info)
        result = comparator.is_same(source_partition_info, dest_partition_info)
        if not result:
            logger.error('Range partition value is different between %s' % part_transfer_pair_msg)

        return result

    def _has_same_list_partition_value(self, source_partition_info, dest_partition_info, part_transfer_pair_msg):
        comparator = PartitionComparatorFactory().get(source_partition_info, dest_partition_info)
        result = comparator.is_same(source_partition_info, dest_partition_info)
        if not result:
            logger.error('List partition value is different between %s' % part_transfer_pair_msg)

        return result

    def _process_list_partition_value(self, partition_info):
        """
        In the scenario of multiple partition columns, the partition values will
        be in the format: (({constraint1} {constraint2})) in pg_catalog.pg_partition_rule
        """
        partition_values = partition_info[2: len(partition_info) - 2].split('} {')
        partition_values = [val.strip('{') if '{' in val else val.strip('}') for val in partition_values]
        return partition_values

    def _get_oid(self, db, schema, table, host, port, user):
        get_oid_sql = '''select c.oid
                     from pg_catalog.pg_class c, pg_catalog.pg_namespace n
                     where c.relname = '%s'
                           and n.oid = c.relnamespace
                           and n.nspname = '%s'
                           ''' % (pg.escape_string(table), pg.escape_string(schema))
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            res = execSQLForSingletonRow(conn, get_oid_sql)
            if not res:
                raise Exception('Failed to retrieve the oid of table %s.%s.%s on host %s' % (db, schema, table, host))
            return res[0]

    def _get_partition_info(self, db, schema, table, host, port, user):
        partition_info = dict()
        oid = self._get_oid(db, schema, table, host, port, user)

        get_partition_type = '''select partitiontype
                                 from pg_catalog.pg_partitions
                                 where partitionschemaname = '%s'
                                       and partitiontablename = '%s';''' % (
        pg.escape_string(schema), pg.escape_string(table))
        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            part_type = execSQLForSingletonRow(conn, get_partition_type)
            partition_info['partitiontype'] = part_type[0]

        get_partition_info_sql = '''select parisdefault, parruleord, parrangestartincl, parrangeendincl, parrangestart,
                                           parrangeend, parrangeevery, parlistvalues
                                    from pg_catalog.pg_partition_rule
                                    where parchildrelid = %s''' % oid

        with dbconn.connect(dbconn.DbURL(host, port, db, user)) as conn:
            row = execSQLForSingletonRow(conn, get_partition_info_sql)
            partition_info['parisdefault'] = row[0]
            partition_info['parruleord'] = row[1]
            partition_info['parrangestartincl'] = row[2]
            partition_info['parrangeendincl'] = row[3]
            partition_info['parrangestart'] = row[4]
            partition_info['parrangeend'] = row[5]
            partition_info['parrangeevery'] = row[6]
            partition_info['parlistvalues'] = row[7]

            row = execSQLForSingletonRow(conn, 'SELECT version()')
            version = GpVersion(row[0])
            partition_info['version'] = version.getVersionRelease()
        return partition_info

    def _check_leaf_partition_set(self, table_pair):
        # check if the source and destination table are all leaf partition tables

        logger.debug('Verifying partition table is leaf partition table.')
        if not self._is_leaf_partition(table_pair.source, self._options.source_host,
                                       self._options.source_port, self._options.source_user):
            raise Exception('Source table %s is not a leaf partition table.' % str(table_pair.source))

        if self._options.partition_transfer and not self._is_leaf_partition(table_pair.dest, self._options.dest_host,
                                                                            self._options.dest_port,
                                                                            self._options.dest_user):
            raise Exception('Destination table %s is not a leaf partition table.' % str(table_pair.dest))

    def _is_leaf_partition(self, tbl, host, port, user):
        check_leaf_partition_sql = '''select relname from pg_class r
                             where not relhassubclass and exists
                             (select oid from pg_partition_rule p where r.oid = p.parchildrelid) and not exists
                             (select oid from pg_partition p where r.oid = p.parrelid) and
                             relnamespace = (select oid from pg_namespace where nspname = '%s') and
                             relname = '%s' ;''' % (pg.escape_string(tbl.schema), pg.escape_string(tbl.table))
        with dbconn.connect(dbconn.DbURL(host, port, tbl.database, user)) as conn:
            cursor = execSQL(conn, check_leaf_partition_sql)
            if cursor.rowcount != 1:
                return False
        return True

    def _get_host_map(self):
        """
        Reads in the host map file.

        The format of the input file is:
            hostname1,addr1[,addr2...]
            ...
            hostnameN,addr1[,addr2...]

        Can use a # to comment a line.
        """
        host_map = None

        if self._options.source_map_file:
            logger.info('Reading source host map file...')
            with open(self._options.source_map_file) as map_file:
                host_map = dict()
                for line in map_file:
                    line = line.strip()
                    if line.startswith('#') or len(line) == 0:
                        continue
                    line_list = line.split(',')
                    (hostname, addresses) = (line_list[0], line_list[1:])
                    if hostname not in host_map:
                        host_map[hostname] = set()
                    map(host_map[hostname].add, [address.strip()
                                                 for address in addresses])
        elif self._options.source_host == self._options.dest_host and \
                        self._options.source_port == self._options.dest_port and \
                self._options.dest_database and not self._options.full:
            # src and dest are same so we don't need map file, we can build it
            host_map = dict()
            for seg in self._src_config.getSegDbList():
                hostname = seg.getSegmentHostName()
                if seg.isSegmentMaster() or hostname in host_map:
                    continue
                host_map[hostname] = [hostname]
        else:
            raise Exception('--source-map-file option must be used when '
                            'source and destination systems are not the same')

        return host_map

    def _validate_host_map(self):
        """
        Validates that all hosts in the source system have an entry in the map
        file.  It validates against the hostname column in
        gp_segment_configuration catalog table.
        """
        logger.info('Validating source host map...')

        if not self._host_map or len(self._host_map) == 0:
            raise Exception('No hosts in map')

        for seg in self._src_config.getSegDbList():
            if seg.isSegmentMaster():
                continue
            seg_host = seg.getSegmentHostName()
            if seg_host not in self._host_map:
                raise Exception('Hostname %s missing from map file' % seg_host)
            if len(self._host_map[seg_host]) == 0:
                raise Exception('No addresses for hostname %s' % seg_host)


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
if __name__ == '__main__':
    # Preemptively check for command -h | --help.
    # This will automacially exit gptransfer and print the help text if -h | --help is found
    exit_on_help = create_parser()
    exit_on_help.parse_args()

    sys.argv[0] = EXECNAME
    simple_main(create_parser, GpTransfer,
                {'pidfilename': GPTRANSFER_PID_FILE,
                 'programNameOverride': EXECNAME})
