Project

General

Profile

1 3077 aaronmk
# Database import/export
2
3 3534 aaronmk
import copy
4 4995 aaronmk
import csv
5 3431 aaronmk
import operator
6 3714 aaronmk
import warnings
7 5588 aaronmk
import sys
8 3431 aaronmk
9 4995 aaronmk
import csvs
10 3077 aaronmk
import exc
11
import dicts
12
import sql
13
import sql_gen
14 5588 aaronmk
import streams
15 3077 aaronmk
import strings
16
import util
17
18 3645 aaronmk
##### Exceptions
19
20
# Can't use built-in SyntaxError because it stringifies to only the first line
21
class SyntaxError(Exception): pass
22
23 3081 aaronmk
##### Data cleanup
24
25 4915 aaronmk
null_strs = ['', '-', r'\N', 'NULL', 'UNKNOWN', 'nulo']
26 4209 aaronmk
27 4447 aaronmk
def cleanup_table(db, table):
28 3081 aaronmk
    table = sql_gen.as_Table(table)
29 5384 aaronmk
    cols = filter(lambda c: sql_gen.is_text_col(db, c),
30
        sql.table_cols(db, table))
31 5394 aaronmk
    try: pkey_col = sql.table_pkey_col(db, table)
32
    except sql.DoesNotExistException: pass
33
    else:
34
        try: cols.remove(pkey_col)
35
        except ValueError: pass
36 4928 aaronmk
    if not cols: return
37 3081 aaronmk
38 4993 aaronmk
    db.log_debug('Cleaning up table', level=1.5)
39
40 4209 aaronmk
    expr = 'trim(both from %s)'
41
    for null in null_strs: expr = 'nullif('+expr+', '+db.esc_value(null)+')'
42
    changes = [(v, sql_gen.CustomCode(expr % v.to_str(db))) for v in cols]
43 3081 aaronmk
44 4444 aaronmk
    while True:
45
        try:
46
            sql.update(db, table, changes, in_place=True, recover=True)
47
            break # successful
48
        except sql.NullValueException, e:
49 4457 aaronmk
            db.log_debug('Caught exception: '+exc.str_(e))
50 4444 aaronmk
            col, = e.cols
51
            sql.drop_not_null(db, col)
52 4992 aaronmk
53
    db.log_debug('Vacuuming and reanalyzing table', level=1.5)
54
    sql.vacuum(db, table)
55 3081 aaronmk
56 3078 aaronmk
##### Error tracking
57
58
def track_data_error(db, errors_table, cols, value, error_code, error):
59
    '''
60
    @param errors_table If None, does nothing.
61
    '''
62
    if errors_table == None or cols == (): return
63
64
    for col in cols:
65
        try:
66
            sql.insert(db, errors_table, dict(column=col.name, value=value,
67
                error_code=error_code, error=error), recover=True,
68
                cacheable=True, log_level=4)
69
        except sql.DuplicateKeyException: pass
70
71 3506 aaronmk
class ExcToErrorsTable(sql_gen.ExcToWarning):
72
    '''Handles an exception by saving it or converting it to a warning.'''
73 3511 aaronmk
    def __init__(self, return_, srcs, errors_table, value=None):
74 3506 aaronmk
        '''
75
        @param return_ See sql_gen.ExcToWarning
76
        @param srcs The column names for the errors table
77
        @param errors_table None|sql_gen.Table
78 3511 aaronmk
        @param value The value (or an expression for it) that caused the error
79 3506 aaronmk
        @pre The invalid value must be in a local variable "value" of type text.
80
        '''
81
        sql_gen.ExcToWarning.__init__(self, return_)
82
83 3511 aaronmk
        value = sql_gen.as_Code(value)
84
85 3506 aaronmk
        self.srcs = srcs
86
        self.errors_table = errors_table
87 3511 aaronmk
        self.value = value
88 3501 aaronmk
89 3506 aaronmk
    def to_str(self, db):
90
        if not self.srcs or self.errors_table == None:
91
            return sql_gen.ExcToWarning.to_str(self, db)
92
93 3459 aaronmk
        errors_table_cols = map(sql_gen.Col,
94
            ['column', 'value', 'error_code', 'error'])
95 3465 aaronmk
        col_names_query = sql.mk_select(db, sql_gen.NamedValues('c', None,
96 3506 aaronmk
            [[c.name] for c in self.srcs]), order_by=None)
97
        insert_query = sql.mk_insert_select(db, self.errors_table,
98
            errors_table_cols,
99 3465 aaronmk
            sql_gen.Values(errors_table_cols).to_str(db))+';\n'
100 3506 aaronmk
        return '''\
101 3459 aaronmk
-- Save error in errors table.
102
DECLARE
103
    error_code text := SQLSTATE;
104
    error text := SQLERRM;
105 3511 aaronmk
    value text := '''+self.value.to_str(db)+''';
106 3529 aaronmk
    "column" text;
107 3459 aaronmk
BEGIN
108
    -- Insert the value and error for *each* source column.
109 3529 aaronmk
'''+strings.indent(sql_gen.RowExcIgnore(None, col_names_query, insert_query,
110 3467 aaronmk
    row_var=errors_table_cols[0]).to_str(db))+'''
111 3459 aaronmk
END;
112 3501 aaronmk
113 3506 aaronmk
'''+self.return_.to_str(db)
114 3459 aaronmk
115 3507 aaronmk
def data_exception_handler(*args, **kw_args):
116 3506 aaronmk
    '''Handles a data_exception by saving it or converting it to a warning.
117
    For params, see ExcToErrorsTable().
118
    '''
119
    return sql_gen.data_exception_handler(ExcToErrorsTable(*args, **kw_args))
120
121 3078 aaronmk
def cast(db, type_, col, errors_table=None):
122
    '''Casts an (unrenamed) column or value.
123
    If errors_table set and col has srcs, saves errors in errors_table (using
124 3360 aaronmk
    col's srcs attr as source columns). Otherwise, converts errors to warnings.
125 3078 aaronmk
    @param col str|sql_gen.Col|sql_gen.Literal
126
    @param errors_table None|sql_gen.Table|str
127
    '''
128
    col = sql_gen.as_Col(col)
129
130 3112 aaronmk
    # Don't convert exceptions to warnings for user-supplied constants
131
    if isinstance(col, sql_gen.Literal): return sql_gen.Cast(type_, col)
132
133 3078 aaronmk
    assert not isinstance(col, sql_gen.NamedCol)
134
135 3460 aaronmk
    function_name = strings.first_word(type_)
136 3459 aaronmk
    srcs = col.srcs
137 3508 aaronmk
    save_errors = errors_table != None and srcs
138
    if save_errors: # function will be unique for the given srcs
139 3750 aaronmk
        function_name = strings.ustr(sql_gen.FunctionCall(function_name,
140 3508 aaronmk
            *map(sql_gen.to_name_only_col, srcs)))
141 3078 aaronmk
    function = db.TempFunction(function_name)
142
143 3464 aaronmk
    # Create function definition
144
    modifiers = 'STRICT'
145
    if not save_errors: modifiers = 'IMMUTABLE '+modifiers
146 3511 aaronmk
    value_param = sql_gen.FunctionParam('value', 'text')
147
    handler = data_exception_handler('RETURN NULL;\n', srcs, errors_table,
148
        value_param.name)
149 3464 aaronmk
    body = sql_gen.CustomCode(handler.to_str(db, '''\
150 3467 aaronmk
/* The explicit cast to the return type is needed to make the cast happen
151
inside the try block. (Implicit casts to the return type happen at the end
152
of the function, outside any block.) */
153
RETURN value::'''+type_+''';
154 3464 aaronmk
'''))
155
    body.lang='plpgsql'
156 3500 aaronmk
    sql.define_func(db, sql_gen.FunctionDef(function, type_, body,
157 3511 aaronmk
        [value_param], modifiers))
158 3464 aaronmk
159 3078 aaronmk
    return sql_gen.FunctionCall(function, col)
160
161 3538 aaronmk
def func_wrapper_exception_handler(db, return_, args, errors_table):
162 3524 aaronmk
    '''Handles a function call's data_exceptions.
163
    Supports PL/Python functions.
164
    @param return_ See data_exception_handler()
165
    @param args [arg...] Function call's args
166
    @param errors_table See data_exception_handler()
167
    '''
168
    args = filter(sql_gen.has_srcs, args)
169
170
    srcs = sql_gen.cross_join_srcs(args)
171 3538 aaronmk
    value = sql_gen.merge_not_null(db, ',', args)
172 3524 aaronmk
    return sql_gen.NestedExcHandler(
173
        data_exception_handler(return_, srcs, errors_table, value)
174
        , sql_gen.plpythonu_error_handler
175
        )
176
177 3078 aaronmk
def cast_temp_col(db, type_, col, errors_table=None):
178
    '''Like cast(), but creates a new column with the cast values if the input
179
    is a column.
180
    @return The new column or cast value
181
    '''
182
    def cast_(col): return cast(db, type_, col, errors_table)
183
184
    try: col = sql_gen.underlying_col(col)
185
    except sql_gen.NoUnderlyingTableException: return sql_gen.wrap(cast_, col)
186
187
    table = col.table
188 3173 aaronmk
    new_col = sql_gen.suffixed_col(col, '::'+strings.first_word(type_))
189 3078 aaronmk
    expr = cast_(col)
190
191
    # Add column
192
    new_typed_col = sql_gen.TypedCol(new_col.name, type_)
193 3750 aaronmk
    sql.add_col(db, table, new_typed_col, comment=strings.urepr(col)+'::'+type_)
194 3078 aaronmk
    new_col.name = new_typed_col.name # propagate any renaming
195
196 3110 aaronmk
    sql.update(db, table, [(new_col, expr)], in_place=True, recover=True)
197 3078 aaronmk
198
    return new_col
199
200
def errors_table(db, table, if_exists=True):
201
    '''
202
    @param if_exists If set, returns None if the errors table doesn't exist
203
    @return None|sql_gen.Table
204
    '''
205
    table = sql_gen.as_Table(table)
206
    if table.srcs != (): table = table.srcs[0]
207
208
    errors_table = sql_gen.suffixed_table(table, '.errors')
209
    if if_exists and not sql.table_exists(db, errors_table): return None
210
    return errors_table
211
212 4436 aaronmk
def mk_errors_table(db, table):
213
    errors_table_ = errors_table(db, table, if_exists=False)
214 4557 aaronmk
    if sql.table_exists(db, errors_table_, cacheable=False): return
215 4436 aaronmk
216
    typed_cols = [
217
        sql_gen.TypedCol('column', 'text', nullable=False),
218
        sql_gen.TypedCol('value', 'text'),
219
        sql_gen.TypedCol('error_code', 'character varying(5)', nullable=False),
220
        sql_gen.TypedCol('error', 'text', nullable=False),
221
        ]
222
    sql.create_table(db, errors_table_, typed_cols, has_pkey=False)
223
    index_cols = ['column', 'value', 'error_code', 'error']
224
    sql.add_index(db, index_cols, errors_table_, unique=True)
225
226 3078 aaronmk
##### Import
227
228 5568 aaronmk
row_num_col_def = copy.copy(sql.row_num_col_def)
229
row_num_col_def.name = 'row_num'
230 5569 aaronmk
row_num_col_def.type = 'integer'
231 5568 aaronmk
232 5590 aaronmk
def append_csv(db, table, reader, header):
233
    assert sql.table_col_names(db, table) == header
234 5037 aaronmk
235 5017 aaronmk
    def log(msg, level=1): db.log_debug(msg, level)
236
237 5584 aaronmk
    # Wrap in standardizing stream
238 5590 aaronmk
    cols_ct = len(header)
239 5588 aaronmk
    stream = csvs.InputRewriter(streams.ProgressInputStream(csvs.StreamFilter(
240 5590 aaronmk
        csvs.ColCtFilter(reader, cols_ct)), sys.stderr, msg='Read %d row(s)',
241
        n=1000))
242 5584 aaronmk
    dialect = stream.dialect # use default dialect
243
244
    # Create COPY FROM statement
245
    copy_from = ('COPY '+table.to_str(db)+' FROM STDIN DELIMITER '
246
        +db.esc_value(dialect.delimiter)+' NULL '+db.esc_value(''))
247
    assert not csvs.is_tsv(dialect)
248
    copy_from += ' CSV'
249
    if dialect.quoting != csv.QUOTE_NONE:
250
        quote_str = db.esc_value(dialect.quotechar)
251
        copy_from += ' QUOTE '+quote_str
252
        if dialect.doublequote: copy_from += ' ESCAPE '+quote_str
253
    copy_from += ';\n'
254
255
    log(copy_from, level=2)
256
    try: db.db.cursor().copy_expert(copy_from, stream)
257
    except Exception, e: sql.parse_exception(db, e, recover=True)
258 5017 aaronmk
259 5591 aaronmk
def import_csv(db, table, reader, header):
260 4995 aaronmk
    def log(msg, level=1): db.log_debug(msg, level)
261
262
    # Get format info
263 5590 aaronmk
    col_names = map(strings.to_unicode, header)
264 4995 aaronmk
    for i, col in enumerate(col_names): # replace empty column names
265
        if col == '': col_names[i] = 'column_'+str(i)
266
267
    # Select schema and escape names
268
    def esc_name(name): return db.esc_name(name)
269
270
    typed_cols = [sql_gen.TypedCol(v, 'text') for v in col_names]
271
272
    log('Creating table')
273 4999 aaronmk
    # Note that this is not rolled back if the import fails. Instead, it is
274
    # cached, and will not be re-run if the import is retried.
275 4995 aaronmk
    sql.create_table(db, table, typed_cols, has_pkey=False, col_indexes=False)
276
277 5001 aaronmk
    # Free memory used by deleted (rolled back) rows from any failed import.
278
    # This MUST be run so that the rows will be stored in inserted order, and
279
    # the row_num added after import will match up with the CSV's row order.
280 4999 aaronmk
    sql.truncate(db, table)
281
282 4995 aaronmk
    # Load the data
283 5590 aaronmk
    def load(): append_csv(db, table, reader, header)
284 5583 aaronmk
    sql.with_savepoint(db, load)
285 4995 aaronmk
286
    cleanup_table(db, table)
287
288 3077 aaronmk
def put(db, table, row, pkey_=None, row_ct_ref=None):
289
    '''Recovers from errors.
290
    Only works under PostgreSQL (uses INSERT RETURNING).
291
    '''
292 3633 aaronmk
    return put_table(db, table, [], row, row_ct_ref)
293 3077 aaronmk
294
def get(db, table, row, pkey, row_ct_ref=None, create=False):
295
    '''Recovers from errors'''
296
    try:
297
        return sql.value(sql.select(db, table, [pkey], row, limit=1,
298
            recover=True))
299
    except StopIteration:
300
        if not create: raise
301
        return put(db, table, row, pkey, row_ct_ref) # insert new row
302
303
def is_func_result(col):
304
    return col.table.name.find('(') >= 0 and col.name == 'result'
305
306
def into_table_name(out_table, in_tables0, mapping, is_func):
307
    def in_col_str(in_col):
308
        in_col = sql_gen.remove_col_rename(in_col)
309
        if isinstance(in_col, sql_gen.Col):
310
            table = in_col.table
311
            if table == in_tables0:
312
                in_col = sql_gen.to_name_only_col(in_col)
313
            elif is_func_result(in_col): in_col = table # omit col name
314 3750 aaronmk
        return strings.ustr(in_col)
315 3077 aaronmk
316 4491 aaronmk
    str_ = strings.ustr(out_table)
317 3077 aaronmk
    if is_func:
318
        str_ += '('
319
320
        try: value_in_col = mapping['value']
321
        except KeyError:
322 4491 aaronmk
            str_ += ', '.join((strings.ustr(k)+'='+in_col_str(v)
323 3077 aaronmk
                for k, v in mapping.iteritems()))
324
        else: str_ += in_col_str(value_in_col)
325
326
        str_ += ')'
327
    else:
328
        out_col = 'rank'
329
        try: in_col = mapping[out_col]
330
        except KeyError: str_ += '_pkeys'
331
        else: # has a rank column, so hierarchical
332 4491 aaronmk
            str_ += '['+strings.ustr(out_col)+'='+in_col_str(in_col)+']'
333 3077 aaronmk
    return str_
334
335 3628 aaronmk
def put_table(db, out_table, in_tables, mapping, row_ct_ref=None, default=None,
336 3660 aaronmk
    col_defaults={}, on_error=exc.reraise):
337 3077 aaronmk
    '''Recovers from errors.
338
    Only works under PostgreSQL (uses INSERT RETURNING).
339
    IMPORTANT: Must be run at the *beginning* of a transaction.
340
    @param in_tables The main input table to select from, followed by a list of
341
        tables to join with it using the main input table's pkey
342
    @param mapping dict(out_table_col=in_table_col, ...)
343
        * out_table_col: str (*not* sql_gen.Col)
344
        * in_table_col: sql_gen.Col|literal-value
345
    @param default The *output* column to use as the pkey for missing rows.
346
        If this output column does not exist in the mapping, uses None.
347 3618 aaronmk
    @param col_defaults Default values for required columns.
348 3077 aaronmk
    @return sql_gen.Col Where the output pkeys are made available
349
    '''
350 3474 aaronmk
    import psycopg2.extensions
351
352 3077 aaronmk
    out_table = sql_gen.as_Table(out_table)
353
354
    def log_debug(msg): db.log_debug(msg, level=1.5)
355
    def col_ustr(str_):
356
        return strings.repr_no_u(sql_gen.remove_col_rename(str_))
357
358
    log_debug('********** New iteration **********')
359
    log_debug('Inserting these input columns into '+strings.as_tt(
360
        out_table.to_str(db))+':\n'+strings.as_table(mapping, ustr=col_ustr))
361
362
    is_function = sql.function_exists(db, out_table)
363
364 5192 aaronmk
    if is_function: row_ct_ref = None # only track inserted rows
365
366 4984 aaronmk
    # Warn if inserting empty table rows
367
    if not mapping and not is_function: # functions with no args OK
368
        warnings.warn(UserWarning('Inserting empty table row(s)'))
369
370 3077 aaronmk
    if is_function: out_pkey = 'result'
371 5388 aaronmk
    else: out_pkey = sql.pkey_name(db, out_table, recover=True)
372 3077 aaronmk
    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)
373
374 5069 aaronmk
    in_tables_ = copy.copy(in_tables) # don't modify input!
375 3432 aaronmk
    try: in_tables0 = in_tables_.pop(0) # first table is separate
376
    except IndexError: in_tables0 = None
377
    else:
378 5388 aaronmk
        in_pkey = sql.pkey_name(db, in_tables0, recover=True)
379 3432 aaronmk
        in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)
380 3431 aaronmk
381
    # Determine if can use optimization for only literal values
382
    is_literals = not reduce(operator.or_, map(sql_gen.is_table_col,
383 3434 aaronmk
        mapping.values()), False)
384 3431 aaronmk
    is_literals_or_function = is_literals or is_function
385
386 3432 aaronmk
    if in_tables0 == None: errors_table_ = None
387
    else: errors_table_ = errors_table(db, in_tables0)
388 3431 aaronmk
389
    # Create input joins from list of input tables
390 3077 aaronmk
    input_joins = [in_tables0]+[sql_gen.Join(v,
391
        {in_pkey: sql_gen.join_same_not_null}) for v in in_tables_]
392
393 5381 aaronmk
    orig_mapping = mapping.copy()
394 3433 aaronmk
    if mapping == {} and not is_function: # need >= one column for INSERT SELECT
395
        mapping = {out_pkey: None} # ColDict will replace with default value
396
397 3431 aaronmk
    if not is_literals:
398 3628 aaronmk
        into = sql_gen.as_Table(into_table_name(out_table, in_tables0, mapping,
399
            is_function))
400 5553 aaronmk
        # Ensure into's out_pkey is different from in_pkey by prepending "out."
401 4495 aaronmk
        if is_function: into_out_pkey = out_pkey
402 5553 aaronmk
        else: into_out_pkey = 'out.'+out_pkey
403 3431 aaronmk
404
        # Set column sources
405
        in_cols = filter(sql_gen.is_table_col, mapping.values())
406
        for col in in_cols:
407
            if col.table == in_tables0: col.set_srcs(sql_gen.src_self)
408
409
        log_debug('Joining together input tables into temp table')
410
        # Place in new table so don't modify input and for speed
411
        in_table = sql_gen.Table('in')
412
        mapping = dicts.join(mapping, sql.flatten(db, in_table, input_joins,
413
            in_cols, preserve=[in_pkey_col]))
414
        input_joins = [in_table]
415
        db.log_debug('Temp table: '+strings.as_tt(in_table.to_str(db)), level=2)
416 3077 aaronmk
417 3692 aaronmk
    # Wrap mapping in a sql_gen.ColDict.
418
    # sql_gen.ColDict sanitizes both keys and values passed into it.
419
    # Do after applying dicts.join() because that returns a plain dict.
420 3077 aaronmk
    mapping = sql_gen.ColDict(db, out_table, mapping)
421
422 5239 aaronmk
    # Save all rows since in_table may have rows deleted
423 3431 aaronmk
    if is_literals: pass
424
    elif is_function: full_in_table = in_table
425 3386 aaronmk
    else:
426
        full_in_table = sql_gen.suffixed_table(in_table, '_full')
427 5530 aaronmk
        sql.copy_table(db, in_table, full_in_table)
428 3287 aaronmk
429 3077 aaronmk
    pkeys_table_exists_ref = [False]
430 5444 aaronmk
    def insert_into_pkeys(query, **kw_args):
431 3077 aaronmk
        if pkeys_table_exists_ref[0]:
432 4484 aaronmk
            sql.insert_select(db, into, [in_pkey, into_out_pkey], query,
433
                **kw_args)
434 3077 aaronmk
        else:
435 3304 aaronmk
            sql.run_query_into(db, query, into=into, add_pkey_=True, **kw_args)
436 3077 aaronmk
            pkeys_table_exists_ref[0] = True
437
438 5523 aaronmk
    def mk_main_select(joins, cols): return sql.mk_select(db, joins, cols)
439 3418 aaronmk
440 3431 aaronmk
    if is_literals: insert_in_table = None
441
    else:
442
        insert_in_table = in_table
443
        insert_in_tables = [insert_in_table]
444 3352 aaronmk
    join_cols = sql_gen.ColDict(db, out_table)
445 3077 aaronmk
446
    exc_strs = set()
447
    def log_exc(e):
448
        e_str = exc.str_(e, first_line_only=True)
449
        log_debug('Caught exception: '+e_str)
450 3552 aaronmk
        if e_str in exc_strs: # avoid infinite loops
451
            log_debug('Exception already seen, handler broken')
452
            on_error(e)
453
            remove_all_rows()
454
        else: exc_strs.add(e_str)
455 3077 aaronmk
456 5377 aaronmk
    ignore_all_ref = [False]
457 3077 aaronmk
    def remove_all_rows():
458
        log_debug('Ignoring all rows')
459 5377 aaronmk
        ignore_all_ref[0] = True # just return the default value column
460 3077 aaronmk
461 5504 aaronmk
    def ensure_cond(cond, e, passed=False, failed=False):
462 5449 aaronmk
        if is_literals: # we know the constraint was applied exactly once
463
            if passed: pass
464
            elif failed: remove_all_rows()
465
            else: raise NotImplementedError()
466 3704 aaronmk
        else:
467
            out_table_cols = sql_gen.ColDict(db, out_table)
468
            out_table_cols.update(util.dict_subset_right_join({},
469 5337 aaronmk
                sql.table_col_names(db, out_table)))
470 3704 aaronmk
471
            in_cols = []
472 5367 aaronmk
            cond = sql_gen.map_expr(db, cond, mapping, in_cols)
473
            cond = sql_gen.map_expr(db, cond, out_table_cols)
474 3704 aaronmk
475
            not_cond = sql_gen.NotCond(sql_gen.CustomCode(cond))
476 5442 aaronmk
            log_debug('Ignoring rows that do not satisfy '+strings.as_tt(cond))
477 5449 aaronmk
            cur = None
478
            if cond == sql_gen.false_expr:
479
                assert failed
480
                remove_all_rows()
481
            elif cond == sql_gen.true_expr: assert passed
482
            else: cur = sql.delete(db, insert_in_table, not_cond)
483
484
            if failed or cur.rowcount > 0: # only if any rows failed cond
485 5505 aaronmk
                track_data_error(db, errors_table_,
486
                    sql_gen.cross_join_srcs(in_cols), None, e.cause.pgcode,
487 5449 aaronmk
                    strings.ensure_newl(e.cause.pgerror)+'condition: '+cond)
488 3352 aaronmk
489 3294 aaronmk
    not_null_cols = set()
490 3077 aaronmk
    def ignore(in_col, value, e):
491 3630 aaronmk
        if sql_gen.is_table_col(in_col):
492
            in_col = sql_gen.with_table(in_col, insert_in_table)
493
494
            track_data_error(db, errors_table_, in_col.srcs, value,
495
                e.cause.pgcode, e.cause.pgerror)
496
497
            sql.add_index(db, in_col, insert_in_table) # enable fast filtering
498
            if value != None and in_col not in not_null_cols:
499 4492 aaronmk
                log_debug('Replacing invalid value '
500
                    +strings.as_tt(strings.urepr(value))+' with NULL in column '
501
                    +strings.as_tt(in_col.to_str(db)))
502 3630 aaronmk
                sql.update(db, insert_in_table, [(in_col, None)],
503
                    sql_gen.ColValueCond(in_col, value))
504
            else:
505 3637 aaronmk
                log_debug('Ignoring rows with '+strings.as_tt(in_col.to_str(db))
506 4492 aaronmk
                    +' = '+strings.as_tt(strings.urepr(value)))
507 3630 aaronmk
                sql.delete(db, insert_in_table,
508
                    sql_gen.ColValueCond(in_col, value))
509
                if value == None: not_null_cols.add(in_col)
510 3293 aaronmk
        else:
511 3630 aaronmk
            assert isinstance(in_col, sql_gen.NamedCol)
512 3684 aaronmk
            in_value = sql_gen.remove_col_rename(in_col)
513
            assert sql_gen.is_literal(in_value)
514
            if value == in_value.value:
515
                if value != None:
516
                    log_debug('Replacing invalid literal '
517
                        +strings.as_tt(in_col.to_str(db))+' with NULL')
518
                    mapping[in_col.name] = None
519
                else:
520
                    remove_all_rows()
521
            # otherwise, all columns were being ignore()d because the specific
522
            # column couldn't be identified, and this was not the invalid column
523 3077 aaronmk
524 3431 aaronmk
    if not is_literals:
525
        def insert_pkeys_table(which):
526
            return sql_gen.Table(sql_gen.concat(in_table.name,
527
                '_insert_'+which+'_pkeys'))
528
        insert_out_pkeys = insert_pkeys_table('out')
529
        insert_in_pkeys = insert_pkeys_table('in')
530 3077 aaronmk
531 3918 aaronmk
    def mk_func_call():
532 3550 aaronmk
        args = dict(((k.name, v) for k, v in mapping.iteritems()))
533 4484 aaronmk
        return sql_gen.FunctionCall(out_table, **args), args
534 3550 aaronmk
535 3918 aaronmk
    if is_function and not is_literals:
536
        log_debug('Defining wrapper function')
537
538
        func_call, args = mk_func_call()
539 4484 aaronmk
        func_call = sql_gen.NamedCol(into_out_pkey, func_call)
540 3918 aaronmk
541
        # Create empty pkeys table so its row type can be used
542 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, input_joins,
543 5528 aaronmk
            [in_pkey_col, func_call], limit=0), recover=True)
544 4484 aaronmk
        result_type = db.col_info(sql_gen.Col(into_out_pkey, into)).type
545 3918 aaronmk
546
        ## Create error handling wrapper function
547
548
        wrapper = db.TempFunction(sql_gen.concat(into.name, '_wrap'))
549
550
        select_cols = [in_pkey_col]+args.values()
551
        row_var = copy.copy(sql_gen.row_var)
552
        row_var.set_srcs([in_table])
553
        in_pkey_var = sql_gen.Col(in_pkey, row_var)
554
555
        args = dict(((k, sql_gen.with_table(v, row_var))
556
            for k, v in args.iteritems()))
557
        func_call = sql_gen.FunctionCall(out_table, **args)
558
559
        def mk_return(result):
560
            return sql_gen.ReturnQuery(sql.mk_select(db,
561
                fields=[in_pkey_var, result], explain=False))
562
        exc_handler = func_wrapper_exception_handler(db,
563
            mk_return(sql_gen.Cast(result_type, None)), args.values(),
564
            errors_table_)
565
566
        sql.define_func(db, sql_gen.FunctionDef(wrapper, sql_gen.SetOf(into),
567
            sql_gen.RowExcIgnore(sql_gen.RowType(in_table),
568 5523 aaronmk
                sql.mk_select(db, input_joins), mk_return(func_call),
569
                exc_handler=exc_handler)
570 3918 aaronmk
            ))
571
        wrapper_table = sql_gen.FunctionCall(wrapper)
572 3550 aaronmk
573 5239 aaronmk
    missing_msg = None
574
575 3077 aaronmk
    # Do inserts and selects
576
    while True:
577 3473 aaronmk
        has_joins = join_cols != {}
578
579 5377 aaronmk
        if ignore_all_ref[0]: break # unrecoverable error, so don't do main case
580 3077 aaronmk
581
        # Prepare to insert new rows
582 3918 aaronmk
        if is_function:
583
            log_debug('Calling function on input rows')
584
            if is_literals: func_call, args = mk_func_call()
585 3077 aaronmk
        else:
586 3550 aaronmk
            log_debug('Trying to insert new rows')
587 3291 aaronmk
            insert_args = dict(recover=True, cacheable=False)
588
            if has_joins:
589
                insert_args.update(dict(ignore=True))
590
            else:
591 3431 aaronmk
                insert_args.update(dict(returning=out_pkey))
592
                if not is_literals:
593
                    insert_args.update(dict(into=insert_out_pkeys))
594 3291 aaronmk
            main_select = mk_main_select([insert_in_table], [sql_gen.with_table(
595
                c, insert_in_table) for c in mapping.values()])
596 3077 aaronmk
597 3292 aaronmk
        try:
598
            cur = None
599 3077 aaronmk
            if is_function:
600 3917 aaronmk
                if is_literals:
601
                    cur = sql.select(db, fields=[func_call], recover=True,
602
                        cacheable=True)
603 5444 aaronmk
                else:
604
                    insert_into_pkeys(sql.mk_select(db, wrapper_table,
605
                        order_by=None), recover=True)
606 3077 aaronmk
            else:
607 3292 aaronmk
                cur = sql.insert_select(db, out_table, mapping.keys(),
608 3077 aaronmk
                    main_select, **insert_args)
609
            break # insert successful
610
        except sql.MissingCastException, e:
611
            log_exc(e)
612
613
            type_ = e.type
614 4140 aaronmk
            if e.col == None: out_cols = mapping.keys()
615
            else: out_cols = [e.col]
616 3077 aaronmk
617 4140 aaronmk
            for out_col in out_cols:
618 4171 aaronmk
                log_debug('Casting '+strings.as_tt(strings.repr_no_u(out_col))
619 4140 aaronmk
                    +' input to '+strings.as_tt(type_))
620
                in_col = mapping[out_col]
621
                while True:
622
                    try:
623
                        mapping[out_col] = cast_temp_col(db, type_, in_col,
624
                            errors_table_)
625
                        break # cast successful
626
                    except sql.InvalidValueException, e:
627
                        log_exc(e)
628
629
                        ignore(in_col, e.value, e)
630 3077 aaronmk
        except sql.DuplicateKeyException, e:
631
            log_exc(e)
632
633 3274 aaronmk
            # Different rows violating different unique constraints not
634
            # supported
635
            assert not join_cols
636
637 5504 aaronmk
            if e.cond != None: ensure_cond(e.cond, e, passed=True)
638 5450 aaronmk
639 3077 aaronmk
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
640
            log_debug('Ignoring existing rows, comparing on these columns:\n'
641
                +strings.as_inline_table(join_cols, ustr=col_ustr))
642 3102 aaronmk
643 3431 aaronmk
            if is_literals:
644
                return sql.value(sql.select(db, out_table, [out_pkey_col],
645 4025 aaronmk
                    join_cols, order_by=None))
646 3431 aaronmk
647 3102 aaronmk
            # Uniquify input table to avoid internal duplicate keys
648
            insert_in_table = sql.distinct_table(db, insert_in_table,
649 3358 aaronmk
                join_cols.values())
650 3144 aaronmk
            insert_in_tables.append(insert_in_table)
651 3077 aaronmk
        except sql.NullValueException, e:
652
            log_exc(e)
653
654
            out_col, = e.cols
655
            try: in_col = mapping[out_col]
656 3618 aaronmk
            except KeyError, e:
657
                try: in_col = mapping[out_col] = col_defaults[out_col]
658
                except KeyError:
659 5239 aaronmk
                    missing_msg = 'Missing mapping for NOT NULL column '+out_col
660
                    log_debug(missing_msg)
661 3618 aaronmk
                    remove_all_rows()
662 3294 aaronmk
            else: ignore(in_col, None, e)
663 3352 aaronmk
        except sql.CheckException, e:
664
            log_exc(e)
665
666 5504 aaronmk
            ensure_cond(e.cond, e, failed=True)
667 3413 aaronmk
        except sql.InvalidValueException, e:
668
            log_exc(e)
669
670
            for in_col in mapping.values(): ignore(in_col, e.value, e)
671 3474 aaronmk
        except psycopg2.extensions.TransactionRollbackError, e:
672
            log_exc(e)
673
            # retry
674 3077 aaronmk
        except sql.DatabaseErrors, e:
675
            log_exc(e)
676
677
            log_debug('No handler for exception')
678
            on_error(e)
679
            remove_all_rows()
680
        # after exception handled, rerun loop with additional constraints
681
682 5239 aaronmk
    # Resolve default value column
683
    if default != None:
684 5381 aaronmk
        if ignore_all_ref[0]: mapping.update(orig_mapping) # use input cols
685 5239 aaronmk
        try: default = mapping[default]
686
        except KeyError:
687
            db.log_debug('Default value column '
688
                +strings.as_tt(strings.repr_no_u(default))
689
                +' does not exist in mapping, falling back to None', level=2.1)
690
            default = None
691 5380 aaronmk
        else: default = sql_gen.remove_col_rename(default)
692 5239 aaronmk
693
    if missing_msg != None and default == None:
694
        warnings.warn(UserWarning(missing_msg))
695
        # not an error because sometimes the mappings include
696
        # extra tables which aren't used by the dataset
697
698
    # Handle unrecoverable errors
699 5377 aaronmk
    if ignore_all_ref[0]:
700 5373 aaronmk
        log_debug('Returning default: '+strings.as_tt(strings.urepr(default)))
701
        return default
702 5239 aaronmk
703 3077 aaronmk
    if cur != None and row_ct_ref != None and cur.rowcount >= 0:
704
        row_ct_ref[0] += cur.rowcount
705
706 3530 aaronmk
    if is_literals: return sql.value(cur)
707
708
    if is_function: pass # pkeys table already created
709 3077 aaronmk
    elif has_joins:
710
        select_joins = input_joins+[sql_gen.Join(out_table, join_cols)]
711
        log_debug('Getting output table pkeys of existing/inserted rows')
712 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, select_joins, [in_pkey_col,
713
            sql_gen.NamedCol(into_out_pkey, out_pkey_col)], order_by=None))
714 3077 aaronmk
    else:
715
        sql.add_row_num(db, insert_out_pkeys) # for joining with input pkeys
716
717
        log_debug('Getting input table pkeys of inserted rows')
718 3285 aaronmk
        # Note that mk_main_select() does not use ORDER BY. Instead, assume that
719
        # since the SELECT query is identical to the one used in INSERT SELECT,
720
        # its rows will be retrieved in the same order.
721 3077 aaronmk
        sql.run_query_into(db, mk_main_select(input_joins, [in_pkey]),
722
            into=insert_in_pkeys)
723
        sql.add_row_num(db, insert_in_pkeys) # for joining with output pkeys
724
725
        assert sql.table_row_count(db, insert_out_pkeys) == sql.table_row_count(
726
            db, insert_in_pkeys)
727
728
        log_debug('Combining output and input pkeys in inserted order')
729
        pkey_joins = [insert_in_pkeys, sql_gen.Join(insert_out_pkeys,
730
            {sql.row_num_col: sql_gen.join_same_not_null})]
731 4484 aaronmk
        in_col = sql_gen.Col(in_pkey, insert_in_pkeys)
732
        out_col = sql_gen.NamedCol(into_out_pkey,
733
            sql_gen.Col(out_pkey, insert_out_pkeys))
734 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, pkey_joins, [in_col, out_col],
735
            order_by=None))
736 3077 aaronmk
737
        sql.empty_temp(db, [insert_out_pkeys, insert_in_pkeys])
738
739 5374 aaronmk
    if not is_function: # is_function doesn't leave holes
740 3187 aaronmk
        log_debug('Setting pkeys of missing rows to '
741 4492 aaronmk
            +strings.as_tt(strings.urepr(default)))
742 5380 aaronmk
        if sql_gen.is_table_col(default):
743
            default = sql_gen.with_table(default, full_in_table)
744 3287 aaronmk
        missing_rows_joins = [full_in_table, sql_gen.Join(into,
745 3187 aaronmk
            {in_pkey: sql_gen.join_same_not_null}, sql_gen.filter_out)]
746
            # must use join_same_not_null or query will take forever
747 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, missing_rows_joins,
748 3287 aaronmk
            [sql_gen.Col(in_pkey, full_in_table),
749 5444 aaronmk
            sql_gen.NamedCol(into_out_pkey, default)], order_by=None))
750 3187 aaronmk
    # otherwise, there is already an entry for every row
751 3077 aaronmk
752 3530 aaronmk
    assert (sql.table_row_count(db, into)
753
        == sql.table_row_count(db, full_in_table))
754
755
    sql.empty_temp(db, insert_in_tables+[full_in_table])
756
757
    srcs = []
758 3619 aaronmk
    if is_function: srcs = sql_gen.cols_srcs(in_cols)
759 4484 aaronmk
    return sql_gen.Col(into_out_pkey, into, srcs)