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
    def esc_name_(name): return sql.esc_name(db, name)
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
    if header == sql.table_col_names(db, table): cols_str = ''
249
    else: cols_str =' ('+(', '.join(map(esc_name_, header)))+')'
250
    copy_from = ('COPY '+table.to_str(db)+cols_str+' FROM STDIN DELIMITER '
251
        +db.esc_value(dialect.delimiter)+' NULL '+db.esc_value(''))
252
    assert not csvs.is_tsv(dialect)
253
    copy_from += ' CSV'
254
    if dialect.quoting != csv.QUOTE_NONE:
255
        quote_str = db.esc_value(dialect.quotechar)
256
        copy_from += ' QUOTE '+quote_str
257
        if dialect.doublequote: copy_from += ' ESCAPE '+quote_str
258
    copy_from += ';\n'
259
    
260
    log(copy_from, level=2)
261
    try: db.db.cursor().copy_expert(copy_from, stream)
262
    except Exception, e: sql.parse_exception(db, e, recover=True)
263

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

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

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

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

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

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