# Database import/export

import copy

import csv

import operator

import warnings

import sys

import csvs

import exc

import dicts

import sql

import sql_gen

import streams

import strings

import util

##### Exceptions

# Can't use built-in SyntaxError because it stringifies to only the first line

class SyntaxError(Exception): pass

##### Data cleanup

def table_nulls_mapped__set(db, table):

    assert isinstance(table, sql_gen.Table)

    sql.run_query(db, 'SELECT util.table_nulls_mapped__set('

        +sql_gen.table2regclass_text(db, table)+')')

def table_nulls_mapped__get(db, table):

    assert isinstance(table, sql_gen.Table)

    return sql.value(sql.run_query(db, 'SELECT util.table_nulls_mapped__get('

        +sql_gen.table2regclass_text(db, table)+')'))

null_strs = ['', '-', r'\N', 'NULL', 'UNKNOWN', 'nulo']

def cleanup_table(db, table):

    '''idempotent'''

    table = sql_gen.as_Table(table)

    assert sql.table_exists(db, table)

    if table_nulls_mapped__get(db, table): return # already cleaned up

    cols = filter(lambda c: sql_gen.is_text_col(db, c),

        sql.table_cols(db, table))

    try: pkey_col = sql.table_pkey_col(db, table)

    except sql.DoesNotExistException: pass

    else:

        try: cols.remove(pkey_col)

        except ValueError: pass

    if not cols: return

    db.log_debug('Cleaning up table', level=1.5)

    expr = 'trim(both from %s)'

    for null in null_strs: expr = 'nullif('+expr+', '+db.esc_value(null)+')'

    changes = [(v, sql_gen.CustomCode(expr % v.to_str(db))) for v in cols]

    while True:

        try:

            sql.update(db, table, changes, in_place=True, recover=True)

            break # successful

        except sql.NullValueException, e:

            db.log_debug('Caught exception: '+exc.str_(e))

            col, = e.cols

            sql.drop_not_null(db, col)

    db.log_debug('Vacuuming and reanalyzing table', level=1.5)

    sql.vacuum(db, table)

    table_nulls_mapped__set(db, table)

##### Error tracking

def track_data_error(db, errors_table, cols, value, error_code, error):

    '''

    @param errors_table If None, does nothing.

    '''

    if errors_table == None: return

    col_names = [c.name for c in cols]

    if not col_names: col_names = [None] # need at least one entry

    for col_name in col_names:

        try:

            sql.insert(db, errors_table, dict(column=col_name, value=value,

                error_code=error_code, error=error), recover=True,

                cacheable=True, log_level=4)

        except sql.DuplicateKeyException: pass

class ExcToErrorsTable(sql_gen.ExcToWarning):

    '''Handles an exception by saving it or converting it to a warning.'''

    def __init__(self, return_, srcs, errors_table, value=None):

        '''

        @param return_ See sql_gen.ExcToWarning

        @param srcs The column names for the errors table

        @param errors_table None|sql_gen.Table

        @param value The value (or an expression for it) that caused the error

        @pre The invalid value must be in a local variable "value" of type text.

        '''

        sql_gen.ExcToWarning.__init__(self, return_)

        value = sql_gen.as_Code(value)

        self.srcs = srcs

        self.errors_table = errors_table

        self.value = value

    def to_str(self, db):

        if not self.srcs or self.errors_table == None:

            return sql_gen.ExcToWarning.to_str(self, db)

        errors_table_cols = map(sql_gen.Col,

            ['column', 'value', 'error_code', 'error'])

        col_names_query = sql.mk_select(db, sql_gen.NamedValues('c', None,

            [[c.name] for c in self.srcs]), order_by=None)

        insert_query = sql.mk_insert_select(db, self.errors_table,

            errors_table_cols,

            sql_gen.Values(errors_table_cols).to_str(db))+';\n'

        return '''\

-- Save error in errors table.

DECLARE

    error_code text := SQLSTATE;

    error text := SQLERRM;

    value text := '''+self.value.to_str(db)+''';

    "column" text;

BEGIN

    -- Insert the value and error for *each* source column.

'''+strings.indent(sql_gen.RowExcIgnore(None, col_names_query, insert_query,

    row_var=errors_table_cols[0]).to_str(db))+'''

END;

'''+self.return_.to_str(db)

def data_exception_handler(*args, **kw_args):

    '''Handles a data_exception by saving it or converting it to a warning.

    For params, see ExcToErrorsTable().

    '''

    return sql_gen.data_exception_handler(ExcToErrorsTable(*args, **kw_args))

def cast(db, type_, col, errors_table=None):

    '''Casts an (unrenamed) column or value.

    If errors_table set and col has srcs, saves errors in errors_table (using

    col's srcs attr as source columns). Otherwise, converts errors to warnings.

    @param col str|sql_gen.Col|sql_gen.Literal

    @param errors_table None|sql_gen.Table|str

    '''

    col = sql_gen.as_Col(col)

    # Don't convert exceptions to warnings for user-supplied constants

    if isinstance(col, sql_gen.Literal): return sql_gen.Cast(type_, col)

    assert not isinstance(col, sql_gen.NamedCol)

    function_name = strings.first_word(type_)

    srcs = col.srcs

    save_errors = errors_table != None and srcs

    if save_errors: # function will be unique for the given srcs

        function_name = strings.ustr(sql_gen.FunctionCall(function_name,

            *map(sql_gen.to_name_only_col, srcs)))

    function = db.TempFunction(function_name)

    # Create function definition

    modifiers = 'STRICT'

    if not save_errors: modifiers = 'IMMUTABLE '+modifiers

    value_param = sql_gen.FunctionParam('value', 'anyelement')

    handler = data_exception_handler('RETURN NULL;\n', srcs, errors_table,

        value_param.name)

    body = sql_gen.CustomCode(handler.to_str(db, '''\

/* The explicit cast to the return type is needed to make the cast happen

inside the try block. (Implicit casts to the return type happen at the end

of the function, outside any block.) */

RETURN '''+sql_gen.Cast(type_, sql_gen.CustomCode('value')).to_str(db)+''';

'''))

    body.lang='plpgsql'

    sql.define_func(db, sql_gen.FunctionDef(function, type_, body,

        [value_param], modifiers))

    return sql_gen.FunctionCall(function, col)

def func_wrapper_exception_handler(db, return_, args, errors_table):

    '''Handles a function call's data_exceptions.

    Supports PL/Python functions.

    @param return_ See data_exception_handler()

    @param args [arg...] Function call's args

    @param errors_table See data_exception_handler()

    '''

    args = filter(sql_gen.has_srcs, args)

    srcs = sql_gen.cross_join_srcs(args)

    value = sql_gen.merge_not_null(db, ',', args)

    return sql_gen.NestedExcHandler(

        data_exception_handler(return_, srcs, errors_table, value)

        , sql_gen.plpythonu_error_handler

        )

def cast_temp_col(db, type_, col, errors_table=None):

    '''Like cast(), but creates a new column with the cast values if the input

    is a column.

    @return The new column or cast value

    '''

    def cast_(col): return cast(db, type_, col, errors_table)

    try: col = sql_gen.underlying_col(col)

    except sql_gen.NoUnderlyingTableException: return sql_gen.wrap(cast_, col)

    table = col.table

    new_col = sql_gen.suffixed_col(col, '::'+strings.first_word(type_))

    expr = cast_(col)

    # Add column

    new_typed_col = sql_gen.TypedCol(new_col.name, type_)

    sql.add_col(db, table, new_typed_col, comment=strings.urepr(col)+'::'+type_)

    new_col.name = new_typed_col.name # propagate any renaming

    sql.update(db, table, [(new_col, expr)], in_place=True, recover=True)

    return new_col

def errors_table(db, table, if_exists=True):

    '''

    @param if_exists If set, returns None if the errors table doesn't exist

    @return None|sql_gen.Table

    '''

    table = sql_gen.as_Table(table)

    if table.srcs != (): table = table.srcs[0]

    errors_table = sql_gen.suffixed_table(table, '.errors')

    if if_exists and not sql.table_exists(db, errors_table): return None

    return errors_table

def mk_errors_table(db, table):

    errors_table_ = errors_table(db, table, if_exists=False)

    if sql.table_exists(db, errors_table_, cacheable=False): return

    typed_cols = [

        sql_gen.TypedCol('column', 'text'),

        sql_gen.TypedCol('value', 'text'),

        sql_gen.TypedCol('error_code', 'character varying(5)', nullable=False),

        sql_gen.TypedCol('error', 'text', nullable=False),

        ]

    sql.create_table(db, errors_table_, typed_cols, has_pkey=False)

    index_cols = ['column', sql_gen.CustomCode('md5(value)'), 'error_code',

        sql_gen.CustomCode('md5(error)')]

    sql.add_index(db, index_cols, errors_table_, unique=True)

##### Import

row_num_col_def = copy.copy(sql.row_num_col_def)

row_num_col_def.name = 'row_num'

row_num_col_def.type = 'integer'

def append_csv(db, table, reader, header):

    def esc_name_(name): return sql.esc_name(db, name)

    def log(msg, level=1): db.log_debug(msg, level)

    # Wrap in standardizing stream

    cols_ct = len(header)

    stream = csvs.InputRewriter(streams.ProgressInputStream(csvs.StreamFilter(

        csvs.ColCtFilter(reader, cols_ct)), sys.stderr, msg='Read %d row(s)',

        n=1000))

    dialect = stream.dialect # use default dialect

    # Create COPY FROM statement

    if header == sql.table_col_names(db, table): cols_str = ''

    else: cols_str =' ('+(', '.join(map(esc_name_, header)))+')'

    copy_from = ('COPY '+table.to_str(db)+cols_str+' FROM STDIN DELIMITER '

        +db.esc_value(dialect.delimiter)+' NULL '+db.esc_value(''))

    assert not csvs.is_tsv(dialect)

    copy_from += ' CSV'

    if dialect.quoting != csv.QUOTE_NONE:

        quote_str = db.esc_value(dialect.quotechar)

        copy_from += ' QUOTE '+quote_str

        if dialect.doublequote: copy_from += ' ESCAPE '+quote_str

    copy_from += ';\n'

    log(copy_from, level=2)

    try: db.db.cursor().copy_expert(copy_from, stream)

    except Exception, e: sql.parse_exception(db, e, recover=True)

def import_csv(db, table, reader, header):

    def log(msg, level=1): db.log_debug(msg, level)

    # Get format info

    col_names = map(strings.to_unicode, header)

    for i, col in enumerate(col_names): # replace empty column names

        if col == '': col_names[i] = 'column_'+str(i)

    # Select schema and escape names

    def esc_name(name): return db.esc_name(name)

    typed_cols = [sql_gen.TypedCol(v, 'text') for v in col_names]

    typed_cols.insert(0, row_num_col_def)

    header.insert(0, row_num_col_def.name)

    reader = csvs.RowNumFilter(reader)

    log('Creating table')

    # Note that this is not rolled back if the import fails. Instead, it is

    # cached, and will not be re-run if the import is retried.

    sql.create_table(db, table, typed_cols, has_pkey=False, col_indexes=False)

    # Free memory used by deleted (rolled back) rows from any failed import.

    # This MUST be run so that the rows will be stored in inserted order, and

    # the row_num added after import will match up with the CSV's row order.

    sql.truncate(db, table)

    # Load the data

    def load(): append_csv(db, table, reader, header)

    sql.with_savepoint(db, load)

    cleanup_table(db, table)

def put(db, table, row, pkey_=None, row_ct_ref=None, on_error=exc.reraise):

    '''Recovers from errors.

    Only works under PostgreSQL (uses INSERT RETURNING).

    '''

    return put_table(db, table, [], row, row_ct_ref, on_error=on_error)

def get(db, table, row, pkey, row_ct_ref=None, create=False):

    '''Recovers from errors'''

    try:

        return sql.value(sql.select(db, table, [pkey], row, limit=1,

            recover=True))

    except StopIteration:

        if not create: raise

        return put(db, table, row, pkey, row_ct_ref) # insert new row

def is_func_result(col):

    return col.table.name.find('(') >= 0 and col.name == 'result'

def into_table_name(out_table, in_tables0, mapping, is_func):

    def in_col_str(in_col):

        in_col = sql_gen.remove_col_rename(in_col)

        if isinstance(in_col, sql_gen.Col):

            table = in_col.table

            if table == in_tables0:

                in_col = sql_gen.to_name_only_col(in_col)

            elif is_func_result(in_col): in_col = table # omit col name

        return strings.ustr(in_col)

    str_ = strings.ustr(out_table)

    if is_func:

        str_ += '('

        try: value_in_col = mapping['value']

        except KeyError:

            str_ += ', '.join((strings.ustr(k)+'='+in_col_str(v)

                for k, v in mapping.iteritems()))

        else: str_ += in_col_str(value_in_col)

        str_ += ')'

    else:

        out_col = 'rank'

        try: in_col = mapping[out_col]

        except KeyError: str_ += '_pkeys'

        else: # has a rank column, so hierarchical

            str_ += '['+strings.ustr(out_col)+'='+in_col_str(in_col)+']'

    return str_

def put_table(db, out_table, in_tables, mapping, row_ct_ref=None, default=None,

    col_defaults={}, on_error=exc.reraise):

    '''Recovers from errors.

    Only works under PostgreSQL (uses INSERT RETURNING).

    Warning: This function's normalizing algorithm does not support database

    triggers that populate fields covered by the unique constraint used to do

    the DISTINCT ON. Such fields must be populated by the mappings instead.

    (Other unique constraints and other non-unique fields are not affected by

    this restriction on triggers. Note that the primary key will normally not be

    the DISTINCT ON constraint, so trigger-populated natural keys are supported

    *unless* the input table contains duplicate rows for some generated keys.)

    Note that much of the complexity of the normalizing algorithm is due to

    PostgreSQL (and other DB systems) not having a native command for

    insert/on duplicate select. This operation is a cross between MySQL's

    INSERT ON DUPLICATE KEY UPDATE (which does not provide SELECT

    functionality), and PostgreSQL's INSERT RETURNING (which throws an error

    on duplicate instead of returning the existing row).

    However, PostgreSQL 9.1+ now provides a way to implement insert/on duplicate

    select just once for each table using the new INSTEAD OF triggers

    (http://www.postgresql.org/docs/9.1/static/plpgsql-trigger.html),

    which even support INSERT RETURNING. INSTEAD OF triggers can also be used to

    implement SQL functions without requiring special support by put_table().

    (INSTEAD OF triggers were not used when put_table() was developed, because

    it was necessary to support PostgreSQL 9.0.)

    @param in_tables The main input table to select from, followed by a list of

        tables to join with it using the main input table's pkey

    @param mapping dict(out_table_col=in_table_col, ...)

        * out_table_col: str (*not* sql_gen.Col)

        * in_table_col: sql_gen.Col|literal-value

    @param default The *output* column to use as the pkey for missing rows.

        If this output column does not exist in the mapping, uses None.

    @param col_defaults Default values for required columns.

    @return sql_gen.Col Where the output pkeys are made available

    '''

    import psycopg2.extensions

    # Special handling for functions with hstore params

    if out_table == '_map':

        import psycopg2.extras

        psycopg2.extras.register_hstore(db.db)

        # Parse args

        try: value = mapping.pop('value')

        except KeyError: return None # value required

        mapping = dict([(k, sql_gen.get_value(v))

            for k, v in mapping.iteritems()]) # unwrap literal value

        mapping = dict(map=mapping, value=value) # non-value params -> hstore

    out_table = sql_gen.as_Table(out_table)

    def log_debug(msg): db.log_debug(msg, level=1.5)

    def col_ustr(str_):

        return strings.repr_no_u(sql_gen.remove_col_rename(str_))

    log_debug('********** New iteration **********')

    log_debug('Inserting these input columns into '+strings.as_tt(

        out_table.to_str(db))+':\n'+strings.as_table(mapping, ustr=col_ustr))

    is_function = sql.function_exists(db, out_table)

    if is_function: row_ct_ref = None # only track inserted rows

    # Warn if inserting empty table rows

    if not mapping and not is_function: # functions with no args OK

        warnings.warn(UserWarning('Inserting empty table row(s)'))

    if is_function: out_pkey = 'result'

    else: out_pkey = sql.pkey_name(db, out_table, recover=True)

    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)

    in_tables_ = copy.copy(in_tables) # don't modify input!

    try: in_tables0 = in_tables_.pop(0) # first table is separate

    except IndexError: in_tables0 = None

    else:

        in_pkey = sql.pkey_name(db, in_tables0, recover=True)

        in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)

    # Determine if can use optimization for only literal values

    is_literals = not reduce(operator.or_, map(sql_gen.is_table_col,

        mapping.values()), False)

    is_literals_or_function = is_literals or is_function

    if in_tables0 == None: errors_table_ = None

    else: errors_table_ = errors_table(db, in_tables0)

    # Create input joins from list of input tables

    input_joins = [in_tables0]+[sql_gen.Join(v,

        {in_pkey: sql_gen.join_same_not_null}) for v in in_tables_]

    orig_mapping = mapping.copy()

    if mapping == {} and not is_function: # need >= one column for INSERT SELECT

        mapping = {out_pkey: None} # ColDict will replace with default value

    if not is_literals:

        into = sql_gen.as_Table(into_table_name(out_table, in_tables0, mapping,

            is_function))

        # Ensure into's out_pkey is different from in_pkey by prepending "out."

        if is_function: into_out_pkey = out_pkey

        else: into_out_pkey = 'out.'+out_pkey

        # Set column sources

        in_cols = filter(sql_gen.is_table_col, mapping.values())

        for col in in_cols:

            if col.table == in_tables0: col.set_srcs(sql_gen.src_self)

        log_debug('Joining together input tables into temp table')

        # Place in new table so don't modify input and for speed

        in_table = sql_gen.Table('in')

        mapping = dicts.join(mapping, sql.flatten(db, in_table, input_joins,

            in_cols, preserve=[in_pkey_col]))

        input_joins = [in_table]

        db.log_debug('Temp table: '+strings.as_tt(in_table.to_str(db)), level=2)

    # Wrap mapping in a sql_gen.ColDict.

    # sql_gen.ColDict sanitizes both keys and values passed into it.

    # Do after applying dicts.join() because that returns a plain dict.

    mapping = sql_gen.ColDict(db, out_table, mapping)

    # Save all rows since in_table may have rows deleted

    if is_literals: pass

    elif is_function: full_in_table = in_table

    else:

        full_in_table = sql_gen.suffixed_table(in_table, '_full')

        sql.copy_table(db, in_table, full_in_table)

    pkeys_table_exists_ref = [False]

    def insert_into_pkeys(query, **kw_args):

        if pkeys_table_exists_ref[0]:

            sql.insert_select(db, into, [in_pkey, into_out_pkey], query,

                **kw_args)

        else:

            kw_args.setdefault('add_pkey_', True)

            sql.run_query_into(db, query, into=into, **kw_args)

            pkeys_table_exists_ref[0] = True

    def mk_main_select(joins, cols): return sql.mk_select(db, joins, cols)

    if is_literals: insert_in_table = None

    else:

        insert_in_table = in_table

        insert_in_tables = [insert_in_table]

    join_cols = sql_gen.ColDict(db, out_table)

    exc_strs = set()

    def log_exc(e):

        e_str = exc.str_(e, first_line_only=True)

        log_debug('Caught exception: '+e_str)

        if e_str in exc_strs: # avoid infinite loops

            log_debug('Exception already seen, handler broken')

            on_error(e)

            remove_all_rows()

            return False

        else: exc_strs.add(e_str)

        return True

    ignore_all_ref = [False]

    def remove_all_rows():

        log_debug('Ignoring all rows')

        ignore_all_ref[0] = True # just return the default value column

    def handle_unknown_exc(e):

        log_debug('No handler for exception')

        on_error(e)

        remove_all_rows()

    def ensure_cond(cond, e, passed=False, failed=False):

        if is_literals: # we know the constraint was applied exactly once

            if passed: pass

            elif failed: remove_all_rows()

            else: raise NotImplementedError()

        else:

            if not is_function:

                out_table_cols = sql_gen.ColDict(db, out_table)

                out_table_cols.update(util.dict_subset_right_join({},

                    sql.table_col_names(db, out_table)))

            in_cols = []

            cond = strings.ustr(cond)

            orig_cond = cond

            cond = sql_gen.map_expr(db, cond, mapping, in_cols)

            if not is_function:

                cond = sql_gen.map_expr(db, cond, out_table_cols)

            log_debug('Ignoring rows that do not satisfy '+strings.as_tt(cond))

            cur = None

            if cond == sql_gen.false_expr:

                assert failed

                remove_all_rows()

            elif cond == sql_gen.true_expr: assert passed

            else:

                while True:

                    not_cond = sql_gen.NotCond(sql_gen.CustomCode(cond))

                    try:

                        cur = sql.delete(db, insert_in_table, not_cond)

                        break

                    except sql.DoesNotExistException, e:

                        if e.type != 'column': raise

                        last_cond = cond

                        cond = sql_gen.map_expr(db, cond, {e.name: None})

                        if cond == last_cond: raise # not fixable

            # If any rows failed cond

            if failed or cur != None and cur.rowcount > 0:

                track_data_error(db, errors_table_,

                    sql_gen.cross_join_srcs(in_cols), None, e.cause.pgcode,

                    strings.ensure_newl(strings.ustr(e.cause.pgerror))

                    +'condition: '+orig_cond+'\ntranslated condition: '+cond)

    not_null_cols = set()

    def ignore(in_col, value, e):

        if sql_gen.is_table_col(in_col):

            in_col = sql_gen.with_table(in_col, insert_in_table)

            track_data_error(db, errors_table_, in_col.srcs, value,

                e.cause.pgcode, e.cause.pgerror)

            sql.add_index(db, in_col, insert_in_table) # enable fast filtering

            if value != None and in_col not in not_null_cols:

                log_debug('Replacing invalid value '

                    +strings.as_tt(strings.urepr(value))+' with NULL in column '

                    +strings.as_tt(in_col.to_str(db)))

                sql.update(db, insert_in_table, [(in_col, None)],

                    sql_gen.ColValueCond(in_col, value))

            else:

                log_debug('Ignoring rows with '+strings.as_tt(in_col.to_str(db))

                    +' = '+strings.as_tt(strings.urepr(value)))

                sql.delete(db, insert_in_table,

                    sql_gen.ColValueCond(in_col, value))

                if value == None: not_null_cols.add(in_col)

        else:

            assert isinstance(in_col, sql_gen.NamedCol)

            in_value = sql_gen.remove_col_rename(in_col)

            assert sql_gen.is_literal(in_value)

            if value == in_value.value:

                if value != None:

                    log_debug('Replacing invalid literal '

                        +strings.as_tt(in_col.to_str(db))+' with NULL')

                    mapping[in_col.name] = None

                else:

                    remove_all_rows()

            # otherwise, all columns were being ignore()d because the specific

            # column couldn't be identified, and this was not the invalid column

    if not is_literals:

        def insert_pkeys_table(which):

            return sql_gen.Table(sql_gen.concat(in_table.name,

                '_insert_'+which+'_pkeys'))

        insert_out_pkeys = insert_pkeys_table('out')

        insert_in_pkeys = insert_pkeys_table('in')

    def mk_func_call():

        args = dict(((k.name, v) for k, v in mapping.iteritems()))

        return sql_gen.FunctionCall(out_table, **args), args

    missing_msg = None

    # Do inserts and selects

    while True:

        has_joins = join_cols != {}

        if ignore_all_ref[0]: break # unrecoverable error, so don't do main case

        # Prepare to insert new rows

        if is_function:

            if is_literals:

                log_debug('Calling function')

                func_call, args = mk_func_call()

        else:

            log_debug('Trying to insert new rows')

            insert_args = dict(recover=True, cacheable=False)

            if has_joins:

                insert_args.update(dict(ignore=True))

            else:

                insert_args.update(dict(returning=out_pkey))

                if not is_literals:

                    insert_args.update(dict(into=insert_out_pkeys))

            main_select = mk_main_select([insert_in_table], [sql_gen.with_table(

                c, insert_in_table) for c in mapping.values()])

        try:

            cur = None

            if is_function:

                if is_literals:

                    cur = sql.select(db, fields=[func_call], recover=True,

                        cacheable=True)

                else:

                    log_debug('Defining wrapper function')

                    func_call, args = mk_func_call()

                    func_call = sql_gen.NamedCol(into_out_pkey, func_call)

                    # Create empty pkeys table so its row type can be used

                    insert_into_pkeys(sql.mk_select(db, input_joins,

                        [in_pkey_col, func_call], limit=0), add_pkey_=False,

                        recover=True)

                    result_type = db.col_info(sql_gen.Col(into_out_pkey,

                        into)).type

                    ## Create error handling wrapper function

                    wrapper = db.TempFunction(sql_gen.concat(into.name,

                        '_wrap'))

                    select_cols = [in_pkey_col]+args.values()

                    row_var = copy.copy(sql_gen.row_var)

                    row_var.set_srcs([in_table])

                    in_pkey_var = sql_gen.Col(in_pkey, row_var)

                    args = dict(((k, sql_gen.with_table(v, row_var))

                        for k, v in args.iteritems()))

                    func_call = sql_gen.FunctionCall(out_table, **args)

                    def mk_return(result):

                        return sql_gen.ReturnQuery(sql.mk_select(db,

                            fields=[in_pkey_var, result], explain=False))

                    exc_handler = func_wrapper_exception_handler(db,

                        mk_return(sql_gen.Cast(result_type, None)),

                        args.values(), errors_table_)

                    sql.define_func(db, sql_gen.FunctionDef(wrapper,

                        sql_gen.SetOf(into),

                        sql_gen.RowExcIgnore(sql_gen.RowType(in_table),

                            sql.mk_select(db, input_joins),

                            mk_return(func_call), exc_handler=exc_handler)

                        ))

                    wrapper_table = sql_gen.FunctionCall(wrapper)

                    log_debug('Calling function')

                    insert_into_pkeys(sql.mk_select(db, wrapper_table,

                        order_by=None), recover=True, cacheable=False)

                    sql.add_pkey_or_index(db, into)

            else:

                cur = sql.insert_select(db, out_table, mapping.keys(),

                    main_select, **insert_args)

            break # insert successful

        except sql.MissingCastException, e:

            if not log_exc(e): break

            type_ = e.type

            if e.col == None: out_cols = mapping.keys()

            else: out_cols = [e.col]

            for out_col in out_cols:

                log_debug('Casting '+strings.as_tt(strings.repr_no_u(out_col))

                    +' input to '+strings.as_tt(type_))

                in_col = mapping[out_col]

                while True:

                    try:

                        mapping[out_col] = cast_temp_col(db, type_, in_col,

                            errors_table_)

                        break # cast successful

                    except sql.InvalidValueException, e:

                        if not log_exc(e): break

                        ignore(in_col, e.value, e)

        except sql.DuplicateKeyException, e:

            if not log_exc(e): break

            # Different rows violating different unique constraints not

            # supported

            assert not join_cols

            if e.cond != None: ensure_cond(e.cond, e, passed=True)

            join_cols.update(util.dict_subset_right_join(mapping, e.cols))

            log_debug('Ignoring existing rows, comparing on these columns:\n'

                +strings.as_inline_table(join_cols, ustr=col_ustr))

            if is_literals:

                return sql.value(sql.select(db, out_table, [out_pkey_col],

                    join_cols, order_by=None))

            # Uniquify and filter input table to avoid (most) duplicate keys

            # (Additional duplicates may be added concurrently and will be

            # filtered out separately upon insert.)

            insert_in_table = sql.distinct_table(db, insert_in_table,

                join_cols.values(), [insert_in_table,

                sql_gen.Join(out_table, join_cols, sql_gen.filter_out)])