Project

General

Profile

1
# Database import/export
2

    
3
import copy
4
import csv
5
import operator
6
import warnings
7
import sys
8

    
9
import csvs
10
import exc
11
import dicts
12
import sql
13
import sql_gen
14
import streams
15
import strings
16
import util
17

    
18
##### Exceptions
19

    
20
# Can't use built-in SyntaxError because it stringifies to only the first line
21
class SyntaxError(Exception): pass
22

    
23
##### Data cleanup
24

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

    
27
def cleanup_table(db, table):
28
    table = sql_gen.as_Table(table)
29
    cols = filter(lambda c: sql_gen.is_text_col(db, c),
30
        sql.table_cols(db, table))
31
    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
    if not cols: return
37
    
38
    db.log_debug('Cleaning up table', level=1.5)
39
    
40
    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
    
44
    while True:
45
        try:
46
            sql.update(db, table, changes, in_place=True, recover=True)
47
            break # successful
48
        except sql.NullValueException, e:
49
            db.log_debug('Caught exception: '+exc.str_(e))
50
            col, = e.cols
51
            sql.drop_not_null(db, col)
52
    
53
    db.log_debug('Vacuuming and reanalyzing table', level=1.5)
54
    sql.vacuum(db, table)
55

    
56
##### 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: return
63
    
64
    col_names = [c.name for c in cols]
65
    if not col_names: col_names = [None] # need at least one entry
66
    for col_name in col_names:
67
        try:
68
            sql.insert(db, errors_table, dict(column=col_name, value=value,
69
                error_code=error_code, error=error), recover=True,
70
                cacheable=True, log_level=4)
71
        except sql.DuplicateKeyException: pass
72

    
73
class ExcToErrorsTable(sql_gen.ExcToWarning):
74
    '''Handles an exception by saving it or converting it to a warning.'''
75
    def __init__(self, return_, srcs, errors_table, value=None):
76
        '''
77
        @param return_ See sql_gen.ExcToWarning
78
        @param srcs The column names for the errors table
79
        @param errors_table None|sql_gen.Table
80
        @param value The value (or an expression for it) that caused the error
81
        @pre The invalid value must be in a local variable "value" of type text.
82
        '''
83
        sql_gen.ExcToWarning.__init__(self, return_)
84
        
85
        value = sql_gen.as_Code(value)
86
        
87
        self.srcs = srcs
88
        self.errors_table = errors_table
89
        self.value = value
90
    
91
    def to_str(self, db):
92
        if not self.srcs or self.errors_table == None:
93
            return sql_gen.ExcToWarning.to_str(self, db)
94
        
95
        errors_table_cols = map(sql_gen.Col,
96
            ['column', 'value', 'error_code', 'error'])
97
        col_names_query = sql.mk_select(db, sql_gen.NamedValues('c', None,
98
            [[c.name] for c in self.srcs]), order_by=None)
99
        insert_query = sql.mk_insert_select(db, self.errors_table,
100
            errors_table_cols,
101
            sql_gen.Values(errors_table_cols).to_str(db))+';\n'
102
        return '''\
103
-- Save error in errors table.
104
DECLARE
105
    error_code text := SQLSTATE;
106
    error text := SQLERRM;
107
    value text := '''+self.value.to_str(db)+''';
108
    "column" text;
109
BEGIN
110
    -- Insert the value and error for *each* source column.
111
'''+strings.indent(sql_gen.RowExcIgnore(None, col_names_query, insert_query,
112
    row_var=errors_table_cols[0]).to_str(db))+'''
113
END;
114

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

    
117
def data_exception_handler(*args, **kw_args):
118
    '''Handles a data_exception by saving it or converting it to a warning.
119
    For params, see ExcToErrorsTable().
120
    '''
121
    return sql_gen.data_exception_handler(ExcToErrorsTable(*args, **kw_args))
122

    
123
def cast(db, type_, col, errors_table=None):
124
    '''Casts an (unrenamed) column or value.
125
    If errors_table set and col has srcs, saves errors in errors_table (using
126
    col's srcs attr as source columns). Otherwise, converts errors to warnings.
127
    @param col str|sql_gen.Col|sql_gen.Literal
128
    @param errors_table None|sql_gen.Table|str
129
    '''
130
    col = sql_gen.as_Col(col)
131
    
132
    # Don't convert exceptions to warnings for user-supplied constants
133
    if isinstance(col, sql_gen.Literal): return sql_gen.Cast(type_, col)
134
    
135
    assert not isinstance(col, sql_gen.NamedCol)
136
    
137
    function_name = strings.first_word(type_)
138
    srcs = col.srcs
139
    save_errors = errors_table != None and srcs
140
    if save_errors: # function will be unique for the given srcs
141
        function_name = strings.ustr(sql_gen.FunctionCall(function_name,
142
            *map(sql_gen.to_name_only_col, srcs)))
143
    function = db.TempFunction(function_name)
144
    
145
    # Create function definition
146
    modifiers = 'STRICT'
147
    if not save_errors: modifiers = 'IMMUTABLE '+modifiers
148
    value_param = sql_gen.FunctionParam('value', 'anyelement')
149
    handler = data_exception_handler('RETURN NULL;\n', srcs, errors_table,
150
        value_param.name)
151
    body = sql_gen.CustomCode(handler.to_str(db, '''\
152
/* The explicit cast to the return type is needed to make the cast happen
153
inside the try block. (Implicit casts to the return type happen at the end
154
of the function, outside any block.) */
155
RETURN '''+sql_gen.Cast(type_, sql_gen.CustomCode('value')).to_str(db)+''';
156
'''))
157
    body.lang='plpgsql'
158
    sql.define_func(db, sql_gen.FunctionDef(function, type_, body,
159
        [value_param], modifiers))
160
    
161
    return sql_gen.FunctionCall(function, col)
162

    
163
def func_wrapper_exception_handler(db, return_, args, errors_table):
164
    '''Handles a function call's data_exceptions.
165
    Supports PL/Python functions.
166
    @param return_ See data_exception_handler()
167
    @param args [arg...] Function call's args
168
    @param errors_table See data_exception_handler()
169
    '''
170
    args = filter(sql_gen.has_srcs, args)
171
    
172
    srcs = sql_gen.cross_join_srcs(args)
173
    value = sql_gen.merge_not_null(db, ',', args)
174
    return sql_gen.NestedExcHandler(
175
        data_exception_handler(return_, srcs, errors_table, value)
176
        , sql_gen.plpythonu_error_handler
177
        )
178

    
179
def cast_temp_col(db, type_, col, errors_table=None):
180
    '''Like cast(), but creates a new column with the cast values if the input
181
    is a column.
182
    @return The new column or cast value
183
    '''
184
    def cast_(col): return cast(db, type_, col, errors_table)
185
    
186
    try: col = sql_gen.underlying_col(col)
187
    except sql_gen.NoUnderlyingTableException: return sql_gen.wrap(cast_, col)
188
    
189
    table = col.table
190
    new_col = sql_gen.suffixed_col(col, '::'+strings.first_word(type_))
191
    expr = cast_(col)
192
    
193
    # Add column
194
    new_typed_col = sql_gen.TypedCol(new_col.name, type_)
195
    sql.add_col(db, table, new_typed_col, comment=strings.urepr(col)+'::'+type_)
196
    new_col.name = new_typed_col.name # propagate any renaming
197
    
198
    sql.update(db, table, [(new_col, expr)], in_place=True, recover=True)
199
    
200
    return new_col
201

    
202
def errors_table(db, table, if_exists=True):
203
    '''
204
    @param if_exists If set, returns None if the errors table doesn't exist
205
    @return None|sql_gen.Table
206
    '''
207
    table = sql_gen.as_Table(table)
208
    if table.srcs != (): table = table.srcs[0]
209
    
210
    errors_table = sql_gen.suffixed_table(table, '.errors')
211
    if if_exists and not sql.table_exists(db, errors_table): return None
212
    return errors_table
213

    
214
def mk_errors_table(db, table):
215
    errors_table_ = errors_table(db, table, if_exists=False)
216
    if sql.table_exists(db, errors_table_, cacheable=False): return
217
    
218
    typed_cols = [
219
        sql_gen.TypedCol('column', 'text'),
220
        sql_gen.TypedCol('value', 'text'),
221
        sql_gen.TypedCol('error_code', 'character varying(5)', nullable=False),
222
        sql_gen.TypedCol('error', 'text', nullable=False),
223
        ]
224
    sql.create_table(db, errors_table_, typed_cols, has_pkey=False)
225
    index_cols = ['column', 'value', 'error_code', 'error']
226
    sql.add_index(db, index_cols, errors_table_, unique=True)
227

    
228
##### Import
229

    
230
row_num_col_def = copy.copy(sql.row_num_col_def)
231
row_num_col_def.name = 'row_num'
232
row_num_col_def.type = 'integer'
233

    
234
def append_csv(db, table, reader, header):
235
    assert sql.table_col_names(db, table) == header
236
    
237
    def log(msg, level=1): db.log_debug(msg, level)
238
    
239
    # Wrap in standardizing stream
240
    cols_ct = len(header)
241
    stream = csvs.InputRewriter(streams.ProgressInputStream(csvs.StreamFilter(
242
        csvs.ColCtFilter(reader, cols_ct)), sys.stderr, msg='Read %d row(s)',
243
        n=1000))
244
    dialect = stream.dialect # use default dialect
245
    
246
    # Create COPY FROM statement
247
    copy_from = ('COPY '+table.to_str(db)+' FROM STDIN DELIMITER '
248
        +db.esc_value(dialect.delimiter)+' NULL '+db.esc_value(''))
249
    assert not csvs.is_tsv(dialect)
250
    copy_from += ' CSV'
251
    if dialect.quoting != csv.QUOTE_NONE:
252
        quote_str = db.esc_value(dialect.quotechar)
253
        copy_from += ' QUOTE '+quote_str
254
        if dialect.doublequote: copy_from += ' ESCAPE '+quote_str
255
    copy_from += ';\n'
256
    
257
    log(copy_from, level=2)
258
    try: db.db.cursor().copy_expert(copy_from, stream)
259
    except Exception, e: sql.parse_exception(db, e, recover=True)
260

    
261
def import_csv(db, table, reader, header):
262
    def log(msg, level=1): db.log_debug(msg, level)
263
    
264
    # Get format info
265
    col_names = map(strings.to_unicode, header)
266
    for i, col in enumerate(col_names): # replace empty column names
267
        if col == '': col_names[i] = 'column_'+str(i)
268
    
269
    # Select schema and escape names
270
    def esc_name(name): return db.esc_name(name)
271
    
272
    typed_cols = [sql_gen.TypedCol(v, 'text') for v in col_names]
273
    typed_cols.insert(0, row_num_col_def)
274
    header.insert(0, row_num_col_def.name)
275
    reader = csvs.RowNumFilter(reader)
276
    
277
    log('Creating table')
278
    # Note that this is not rolled back if the import fails. Instead, it is
279
    # cached, and will not be re-run if the import is retried.
280
    sql.create_table(db, table, typed_cols, has_pkey=False, col_indexes=False)
281
    
282
    # Free memory used by deleted (rolled back) rows from any failed import.
283
    # This MUST be run so that the rows will be stored in inserted order, and
284
    # the row_num added after import will match up with the CSV's row order.
285
    sql.truncate(db, table)
286
    
287
    # Load the data
288
    def load(): append_csv(db, table, reader, header)
289
    sql.with_savepoint(db, load)
290
    
291
    cleanup_table(db, table)
292

    
293
def put(db, table, row, pkey_=None, row_ct_ref=None, on_error=exc.reraise):
294
    '''Recovers from errors.
295
    Only works under PostgreSQL (uses INSERT RETURNING).
296
    '''
297
    return put_table(db, table, [], row, row_ct_ref, on_error=on_error)
298

    
299
def get(db, table, row, pkey, row_ct_ref=None, create=False):
300
    '''Recovers from errors'''
301
    try:
302
        return sql.value(sql.select(db, table, [pkey], row, limit=1,
303
            recover=True))
304
    except StopIteration:
305
        if not create: raise
306
        return put(db, table, row, pkey, row_ct_ref) # insert new row
307

    
308
def is_func_result(col):
309
    return col.table.name.find('(') >= 0 and col.name == 'result'
310

    
311
def into_table_name(out_table, in_tables0, mapping, is_func):
312
    def in_col_str(in_col):
313
        in_col = sql_gen.remove_col_rename(in_col)
314
        if isinstance(in_col, sql_gen.Col):
315
            table = in_col.table
316
            if table == in_tables0:
317
                in_col = sql_gen.to_name_only_col(in_col)
318
            elif is_func_result(in_col): in_col = table # omit col name
319
        return strings.ustr(in_col)
320
    
321
    str_ = strings.ustr(out_table)
322
    if is_func:
323
        str_ += '('
324
        
325
        try: value_in_col = mapping['value']
326
        except KeyError:
327
            str_ += ', '.join((strings.ustr(k)+'='+in_col_str(v)
328
                for k, v in mapping.iteritems()))
329
        else: str_ += in_col_str(value_in_col)
330
        
331
        str_ += ')'
332
    else:
333
        out_col = 'rank'
334
        try: in_col = mapping[out_col]
335
        except KeyError: str_ += '_pkeys'
336
        else: # has a rank column, so hierarchical
337
            str_ += '['+strings.ustr(out_col)+'='+in_col_str(in_col)+']'
338
    return str_
339

    
340
def put_table(db, out_table, in_tables, mapping, row_ct_ref=None, default=None,
341
    col_defaults={}, on_error=exc.reraise):
342
    '''Recovers from errors.
343
    Only works under PostgreSQL (uses INSERT RETURNING).
344
    IMPORTANT: Must be run at the *beginning* of a transaction.
345
    @param in_tables The main input table to select from, followed by a list of
346
        tables to join with it using the main input table's pkey
347
    @param mapping dict(out_table_col=in_table_col, ...)
348
        * out_table_col: str (*not* sql_gen.Col)
349
        * in_table_col: sql_gen.Col|literal-value
350
    @param default The *output* column to use as the pkey for missing rows.
351
        If this output column does not exist in the mapping, uses None.
352
    @param col_defaults Default values for required columns.
353
    @return sql_gen.Col Where the output pkeys are made available
354
    '''
355
    import psycopg2.extensions
356
    
357
    # Special handling for functions with hstore params
358
    if out_table == '_map':
359
        import psycopg2.extras
360
        psycopg2.extras.register_hstore(db.db)
361
        
362
        # Parse args
363
        try: value = mapping.pop('value')
364
        except KeyError: return None # value required
365
        
366
        mapping = dict([(k, sql_gen.get_value(v))
367
            for k, v in mapping.iteritems()]) # unwrap literal value
368
        mapping = dict(map=mapping, value=value) # non-value params -> hstore
369
    
370
    out_table = sql_gen.as_Table(out_table)
371
    
372
    def log_debug(msg): db.log_debug(msg, level=1.5)
373
    def col_ustr(str_):
374
        return strings.repr_no_u(sql_gen.remove_col_rename(str_))
375
    
376
    log_debug('********** New iteration **********')
377
    log_debug('Inserting these input columns into '+strings.as_tt(
378
        out_table.to_str(db))+':\n'+strings.as_table(mapping, ustr=col_ustr))
379
    
380
    is_function = sql.function_exists(db, out_table)
381
    
382
    if is_function: row_ct_ref = None # only track inserted rows
383
    
384
    # Warn if inserting empty table rows
385
    if not mapping and not is_function: # functions with no args OK
386
        warnings.warn(UserWarning('Inserting empty table row(s)'))
387
    
388
    if is_function: out_pkey = 'result'
389
    else: out_pkey = sql.pkey_name(db, out_table, recover=True)
390
    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)
391
    
392
    in_tables_ = copy.copy(in_tables) # don't modify input!
393
    try: in_tables0 = in_tables_.pop(0) # first table is separate
394
    except IndexError: in_tables0 = None
395
    else:
396
        in_pkey = sql.pkey_name(db, in_tables0, recover=True)
397
        in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)
398
    
399
    # Determine if can use optimization for only literal values
400
    is_literals = not reduce(operator.or_, map(sql_gen.is_table_col,
401
        mapping.values()), False)
402
    is_literals_or_function = is_literals or is_function
403
    
404
    if in_tables0 == None: errors_table_ = None
405
    else: errors_table_ = errors_table(db, in_tables0)
406
    
407
    # Create input joins from list of input tables
408
    input_joins = [in_tables0]+[sql_gen.Join(v,
409
        {in_pkey: sql_gen.join_same_not_null}) for v in in_tables_]
410
    
411
    orig_mapping = mapping.copy()
412
    if mapping == {} and not is_function: # need >= one column for INSERT SELECT
413
        mapping = {out_pkey: None} # ColDict will replace with default value
414
    
415
    if not is_literals:
416
        into = sql_gen.as_Table(into_table_name(out_table, in_tables0, mapping,
417
            is_function))
418
        # Ensure into's out_pkey is different from in_pkey by prepending "out."
419
        if is_function: into_out_pkey = out_pkey
420
        else: into_out_pkey = 'out.'+out_pkey
421
        
422
        # Set column sources
423
        in_cols = filter(sql_gen.is_table_col, mapping.values())
424
        for col in in_cols:
425
            if col.table == in_tables0: col.set_srcs(sql_gen.src_self)
426
        
427
        log_debug('Joining together input tables into temp table')
428
        # Place in new table so don't modify input and for speed
429
        in_table = sql_gen.Table('in')
430
        mapping = dicts.join(mapping, sql.flatten(db, in_table, input_joins,
431
            in_cols, preserve=[in_pkey_col]))
432
        input_joins = [in_table]
433
        db.log_debug('Temp table: '+strings.as_tt(in_table.to_str(db)), level=2)
434
    
435
    # Wrap mapping in a sql_gen.ColDict.
436
    # sql_gen.ColDict sanitizes both keys and values passed into it.
437
    # Do after applying dicts.join() because that returns a plain dict.
438
    mapping = sql_gen.ColDict(db, out_table, mapping)
439
    
440
    # Save all rows since in_table may have rows deleted
441
    if is_literals: pass
442
    elif is_function: full_in_table = in_table
443
    else:
444
        full_in_table = sql_gen.suffixed_table(in_table, '_full')
445
        sql.copy_table(db, in_table, full_in_table)
446
    
447
    pkeys_table_exists_ref = [False]
448
    def insert_into_pkeys(query, **kw_args):
449
        if pkeys_table_exists_ref[0]:
450
            sql.insert_select(db, into, [in_pkey, into_out_pkey], query,
451
                **kw_args)
452
        else:
453
            kw_args.setdefault('add_pkey_', True)
454
            
455
            sql.run_query_into(db, query, into=into, **kw_args)
456
            pkeys_table_exists_ref[0] = True
457
    
458
    def mk_main_select(joins, cols): return sql.mk_select(db, joins, cols)
459
    
460
    if is_literals: insert_in_table = None
461
    else:
462
        insert_in_table = in_table
463
        insert_in_tables = [insert_in_table]
464
    join_cols = sql_gen.ColDict(db, out_table)
465
    
466
    exc_strs = set()
467
    def log_exc(e):
468
        e_str = exc.str_(e, first_line_only=True)
469
        log_debug('Caught exception: '+e_str)
470
        if e_str in exc_strs: # avoid infinite loops
471
            log_debug('Exception already seen, handler broken')
472
            on_error(e)
473
            remove_all_rows()
474
            return False
475
        else: exc_strs.add(e_str)
476
        return True
477
    
478
    ignore_all_ref = [False]
479
    def remove_all_rows():
480
        log_debug('Ignoring all rows')
481
        ignore_all_ref[0] = True # just return the default value column
482
    
483
    def handle_unknown_exc(e):
484
        log_debug('No handler for exception')
485
        on_error(e)
486
        remove_all_rows()
487
    
488
    def ensure_cond(cond, e, passed=False, failed=False):
489
        if is_literals: # we know the constraint was applied exactly once
490
            if passed: pass
491
            elif failed: remove_all_rows()
492
            else: raise NotImplementedError()
493
        else:
494
            if not is_function:
495
                out_table_cols = sql_gen.ColDict(db, out_table)
496
                out_table_cols.update(util.dict_subset_right_join({},
497
                    sql.table_col_names(db, out_table)))
498
            
499
            in_cols = []
500
            orig_cond = cond
501
            cond = sql_gen.map_expr(db, cond, mapping, in_cols)
502
            if not is_function:
503
                cond = sql_gen.map_expr(db, cond, out_table_cols)
504
            
505
            log_debug('Ignoring rows that do not satisfy '+strings.as_tt(cond))
506
            cur = None
507
            if cond == sql_gen.false_expr:
508
                assert failed
509
                remove_all_rows()
510
            elif cond == sql_gen.true_expr: assert passed
511
            else:
512
                while True:
513
                    not_cond = sql_gen.NotCond(sql_gen.CustomCode(cond))
514
                    try:
515
                        cur = sql.delete(db, insert_in_table, not_cond)
516
                        break
517
                    except sql.DoesNotExistException, e:
518
                        if e.type != 'column': raise
519
                        
520
                        last_cond = cond
521
                        cond = sql_gen.map_expr(db, cond, {e.name: None})
522
                        if cond == last_cond: raise # not fixable
523
            
524
            # If any rows failed cond
525
            if failed or cur != None and cur.rowcount > 0:
526
                track_data_error(db, errors_table_,
527
                    sql_gen.cross_join_srcs(in_cols), None, e.cause.pgcode,
528
                    strings.ensure_newl(e.cause.pgerror)+'condition: '+orig_cond
529
                    +'\ntranslated condition: '+cond)
530
    
531
    not_null_cols = set()
532
    def ignore(in_col, value, e):
533
        if sql_gen.is_table_col(in_col):
534
            in_col = sql_gen.with_table(in_col, insert_in_table)
535
            
536
            track_data_error(db, errors_table_, in_col.srcs, value,
537
                e.cause.pgcode, e.cause.pgerror)
538
            
539
            sql.add_index(db, in_col, insert_in_table) # enable fast filtering
540
            if value != None and in_col not in not_null_cols:
541
                log_debug('Replacing invalid value '
542
                    +strings.as_tt(strings.urepr(value))+' with NULL in column '
543
                    +strings.as_tt(in_col.to_str(db)))
544
                sql.update(db, insert_in_table, [(in_col, None)],
545
                    sql_gen.ColValueCond(in_col, value))
546
            else:
547
                log_debug('Ignoring rows with '+strings.as_tt(in_col.to_str(db))
548
                    +' = '+strings.as_tt(strings.urepr(value)))
549
                sql.delete(db, insert_in_table,
550
                    sql_gen.ColValueCond(in_col, value))
551
                if value == None: not_null_cols.add(in_col)
552
        else:
553
            assert isinstance(in_col, sql_gen.NamedCol)
554
            in_value = sql_gen.remove_col_rename(in_col)
555
            assert sql_gen.is_literal(in_value)
556
            if value == in_value.value:
557
                if value != None:
558
                    log_debug('Replacing invalid literal '
559
                        +strings.as_tt(in_col.to_str(db))+' with NULL')
560
                    mapping[in_col.name] = None
561
                else:
562
                    remove_all_rows()
563
            # otherwise, all columns were being ignore()d because the specific
564
            # column couldn't be identified, and this was not the invalid column
565
    
566
    if not is_literals:
567
        def insert_pkeys_table(which):
568
            return sql_gen.Table(sql_gen.concat(in_table.name,
569
                '_insert_'+which+'_pkeys'))
570
        insert_out_pkeys = insert_pkeys_table('out')
571
        insert_in_pkeys = insert_pkeys_table('in')
572
    
573
    def mk_func_call():
574
        args = dict(((k.name, v) for k, v in mapping.iteritems()))
575
        return sql_gen.FunctionCall(out_table, **args), args
576
    
577
    missing_msg = None
578
    
579
    # Do inserts and selects
580
    while True:
581
        has_joins = join_cols != {}
582
        
583
        if ignore_all_ref[0]: break # unrecoverable error, so don't do main case
584
        
585
        # Prepare to insert new rows
586
        if is_function:
587
            if is_literals:
588
                log_debug('Calling function')
589
                func_call, args = mk_func_call()
590
        else:
591
            log_debug('Trying to insert new rows')
592
            insert_args = dict(recover=True, cacheable=False)
593
            if has_joins:
594
                insert_args.update(dict(ignore=True))
595
            else:
596
                insert_args.update(dict(returning=out_pkey))
597
                if not is_literals:
598
                    insert_args.update(dict(into=insert_out_pkeys))
599
            main_select = mk_main_select([insert_in_table], [sql_gen.with_table(
600
                c, insert_in_table) for c in mapping.values()])
601
        
602
        try:
603
            cur = None
604
            if is_function:
605
                if is_literals:
606
                    cur = sql.select(db, fields=[func_call], recover=True,
607
                        cacheable=True)
608
                else:
609
                    log_debug('Defining wrapper function')
610
                    
611
                    func_call, args = mk_func_call()
612
                    func_call = sql_gen.NamedCol(into_out_pkey, func_call)
613
                    
614
                    # Create empty pkeys table so its row type can be used
615
                    insert_into_pkeys(sql.mk_select(db, input_joins,
616
                        [in_pkey_col, func_call], limit=0), add_pkey_=False,
617
                        recover=True)
618
                    result_type = db.col_info(sql_gen.Col(into_out_pkey,
619
                        into)).type
620
                    
621
                    ## Create error handling wrapper function
622
                    
623
                    wrapper = db.TempFunction(sql_gen.concat(into.name,
624
                        '_wrap'))
625
                    
626
                    select_cols = [in_pkey_col]+args.values()
627
                    row_var = copy.copy(sql_gen.row_var)
628
                    row_var.set_srcs([in_table])
629
                    in_pkey_var = sql_gen.Col(in_pkey, row_var)
630
                    
631
                    args = dict(((k, sql_gen.with_table(v, row_var))
632
                        for k, v in args.iteritems()))
633
                    func_call = sql_gen.FunctionCall(out_table, **args)
634
                    
635
                    def mk_return(result):
636
                        return sql_gen.ReturnQuery(sql.mk_select(db,
637
                            fields=[in_pkey_var, result], explain=False))
638
                    exc_handler = func_wrapper_exception_handler(db,
639
                        mk_return(sql_gen.Cast(result_type, None)),
640
                        args.values(), errors_table_)
641
                    
642
                    sql.define_func(db, sql_gen.FunctionDef(wrapper,
643
                        sql_gen.SetOf(into),
644
                        sql_gen.RowExcIgnore(sql_gen.RowType(in_table),
645
                            sql.mk_select(db, input_joins),
646
                            mk_return(func_call), exc_handler=exc_handler)
647
                        ))
648
                    wrapper_table = sql_gen.FunctionCall(wrapper)
649
                    
650
                    log_debug('Calling function')
651
                    insert_into_pkeys(sql.mk_select(db, wrapper_table,
652
                        order_by=None), recover=True, cacheable=False)
653
                    sql.add_pkey_or_index(db, into, warn=True)
654
            else:
655
                cur = sql.insert_select(db, out_table, mapping.keys(),
656
                    main_select, **insert_args)
657
            break # insert successful
658
        except sql.MissingCastException, e:
659
            if not log_exc(e): break
660
            
661
            type_ = e.type
662
            if e.col == None: out_cols = mapping.keys()
663
            else: out_cols = [e.col]
664
            
665
            for out_col in out_cols:
666
                log_debug('Casting '+strings.as_tt(strings.repr_no_u(out_col))
667
                    +' input to '+strings.as_tt(type_))
668
                in_col = mapping[out_col]
669
                while True:
670
                    try:
671
                        mapping[out_col] = cast_temp_col(db, type_, in_col,
672
                            errors_table_)
673
                        break # cast successful
674
                    except sql.InvalidValueException, e:
675
                        if not log_exc(e): break
676
                        
677
                        ignore(in_col, e.value, e)
678
        except sql.DuplicateKeyException, e:
679
            if not log_exc(e): break
680
            
681
            # Different rows violating different unique constraints not
682
            # supported
683
            assert not join_cols
684
            
685
            if e.cond != None: ensure_cond(e.cond, e, passed=True)
686
            
687
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
688
            log_debug('Ignoring existing rows, comparing on these columns:\n'
689
                +strings.as_inline_table(join_cols, ustr=col_ustr))
690
            
691
            if is_literals:
692
                return sql.value(sql.select(db, out_table, [out_pkey_col],
693
                    join_cols, order_by=None))
694
            
695
            # Uniquify input table to avoid internal duplicate keys
696
            insert_in_table = sql.distinct_table(db, insert_in_table,
697
                join_cols.values())
698
            insert_in_tables.append(insert_in_table)
699
        except sql.NullValueException, e:
700
            if not log_exc(e): break
701
            
702
            out_col, = e.cols
703
            try: in_col = mapping[out_col]
704
            except KeyError, e:
705
                try: in_col = mapping[out_col] = col_defaults[out_col]
706
                except KeyError:
707
                    missing_msg = 'Missing mapping for NOT NULL column '+out_col
708
                    log_debug(missing_msg)
709
                    remove_all_rows()
710
            else: ignore(in_col, None, e)
711
        except sql.CheckException, e:
712
            if not log_exc(e): break
713
            
714
            ensure_cond(e.cond, e, failed=True)
715
        except sql.InvalidValueException, e:
716
            if not log_exc(e): break
717
            
718
            for in_col in mapping.values(): ignore(in_col, e.value, e)
719
        except psycopg2.extensions.TransactionRollbackError, e:
720
            if not log_exc(e): break
721
            # retry
722
        except sql.DatabaseErrors, e:
723
            if not log_exc(e): break
724
            
725
            handle_unknown_exc(e)
726
        # after exception handled, rerun loop with additional constraints
727
    
728
    # Resolve default value column
729
    if default != None:
730
        if ignore_all_ref[0]: mapping.update(orig_mapping) # use input cols
731
        try: default = mapping[default]
732
        except KeyError:
733
            db.log_debug('Default value column '
734
                +strings.as_tt(strings.repr_no_u(default))
735
                +' does not exist in mapping, falling back to None', level=2.1)
736
            default = None
737
        else: default = sql_gen.remove_col_rename(default)
738
    
739
    if missing_msg != None and default == None:
740
        warnings.warn(UserWarning(missing_msg))
741
        # not an error because sometimes the mappings include
742
        # extra tables which aren't used by the dataset
743
    
744
    # Handle unrecoverable errors
745
    if ignore_all_ref[0]:
746
        log_debug('Returning default: '+strings.as_tt(strings.urepr(default)))
747
        return default
748
    
749
    if cur != None and row_ct_ref != None and cur.rowcount >= 0:
750
        row_ct_ref[0] += cur.rowcount
751
    
752
    if is_literals: return sql.value(cur)
753
    
754
    if is_function: pass # pkeys table already created
755
    elif has_joins:
756
        select_joins = input_joins+[sql_gen.Join(out_table, join_cols)]
757
        log_debug('Getting output table pkeys of existing/inserted rows')
758
        insert_into_pkeys(sql.mk_select(db, select_joins, [in_pkey_col,
759
            sql_gen.NamedCol(into_out_pkey, out_pkey_col)], order_by=None))
760
    else:
761
        sql.add_row_num(db, insert_out_pkeys) # for joining with input pkeys
762
        
763
        log_debug('Getting input table pkeys of inserted rows')
764
        # Note that mk_main_select() does not use ORDER BY. Instead, assume that
765
        # since the SELECT query is identical to the one used in INSERT SELECT,
766
        # its rows will be retrieved in the same order.
767
        sql.run_query_into(db, mk_main_select(input_joins, [in_pkey]),
768
            into=insert_in_pkeys)
769
        sql.add_row_num(db, insert_in_pkeys) # for joining with output pkeys
770
        
771
        assert sql.table_row_count(db, insert_out_pkeys) == sql.table_row_count(
772
            db, insert_in_pkeys)
773
        
774
        log_debug('Combining output and input pkeys in inserted order')
775
        pkey_joins = [insert_in_pkeys, sql_gen.Join(insert_out_pkeys,
776
            {sql.row_num_col: sql_gen.join_same_not_null})]
777
        in_col = sql_gen.Col(in_pkey, insert_in_pkeys)
778
        out_col = sql_gen.NamedCol(into_out_pkey,
779
            sql_gen.Col(out_pkey, insert_out_pkeys))
780
        insert_into_pkeys(sql.mk_select(db, pkey_joins, [in_col, out_col],
781
            order_by=None))
782
        
783
        sql.empty_temp(db, [insert_out_pkeys, insert_in_pkeys])
784
    
785
    if not is_function: # is_function doesn't leave holes
786
        log_debug('Setting pkeys of missing rows to '
787
            +strings.as_tt(strings.urepr(default)))
788
        
789
        full_in_pkey_col = sql_gen.Col(in_pkey, full_in_table)
790
        if sql_gen.is_table_col(default):
791
            default = sql_gen.with_table(default, full_in_table)
792
        missing_rows_joins = [full_in_table, sql_gen.Join(into,
793
            {in_pkey: sql_gen.join_same_not_null}, sql_gen.filter_out)]
794
            # must use join_same_not_null or query will take forever
795
        
796
        insert_args = dict(order_by=None)
797
        if not sql.table_has_pkey(db, full_in_table): # in_table has duplicates
798
            insert_args.update(dict(distinct_on=[full_in_pkey_col]))
799
        
800
        insert_into_pkeys(sql.mk_select(db, missing_rows_joins,
801
            [full_in_pkey_col, sql_gen.NamedCol(into_out_pkey, default)],
802
            **insert_args))
803
    # otherwise, there is already an entry for every row
804
    
805
    sql.empty_temp(db, insert_in_tables+[full_in_table])
806
    
807
    srcs = []
808
    if is_function: srcs = sql_gen.cols_srcs(in_cols)
809
    return sql_gen.Col(into_out_pkey, into, srcs)
(35-35/47)