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', sql_gen.CustomCode('md5(value)'), 'error_code',
226
        sql_gen.CustomCode('md5(error)')]
227
    sql.add_index(db, index_cols, errors_table_, unique=True)
228

    
229
##### Import
230

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

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

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

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

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

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

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

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