Project

General

Profile

1
# Database import/export
2

    
3
import copy
4
import operator
5
import warnings
6

    
7
import exc
8
import dicts
9
import sql
10
import sql_gen
11
import strings
12
import util
13

    
14
##### Exceptions
15

    
16
# Can't use built-in SyntaxError because it stringifies to only the first line
17
class SyntaxError(Exception): pass
18

    
19
##### Data cleanup
20

    
21
null_strs = ['', r'\N', 'NULL', 'UNKNOWN']
22

    
23
def cleanup_table(db, table):
24
    table = sql_gen.as_Table(table)
25
    cols = [sql_gen.as_Col(c, table) for c in sql.table_cols(db, table)]
26
    cols = filter(lambda c: sql_gen.is_text_col(db, c), cols)
27
    
28
    expr = 'trim(both from %s)'
29
    for null in null_strs: expr = 'nullif('+expr+', '+db.esc_value(null)+')'
30
    changes = [(v, sql_gen.CustomCode(expr % v.to_str(db))) for v in cols]
31
    
32
    while True:
33
        try:
34
            sql.update(db, table, changes, in_place=True, recover=True)
35
            break # successful
36
        except sql.NullValueException, e:
37
            db.log_debug('Caught exception: '+exc.str_(e))
38
            col, = e.cols
39
            sql.drop_not_null(db, col)
40

    
41
##### Error tracking
42

    
43
def track_data_error(db, errors_table, cols, value, error_code, error):
44
    '''
45
    @param errors_table If None, does nothing.
46
    '''
47
    if errors_table == None or cols == (): return
48
    
49
    for col in cols:
50
        try:
51
            sql.insert(db, errors_table, dict(column=col.name, value=value,
52
                error_code=error_code, error=error), recover=True,
53
                cacheable=True, log_level=4)
54
        except sql.DuplicateKeyException: pass
55

    
56
class ExcToErrorsTable(sql_gen.ExcToWarning):
57
    '''Handles an exception by saving it or converting it to a warning.'''
58
    def __init__(self, return_, srcs, errors_table, value=None):
59
        '''
60
        @param return_ See sql_gen.ExcToWarning
61
        @param srcs The column names for the errors table
62
        @param errors_table None|sql_gen.Table
63
        @param value The value (or an expression for it) that caused the error
64
        @pre The invalid value must be in a local variable "value" of type text.
65
        '''
66
        sql_gen.ExcToWarning.__init__(self, return_)
67
        
68
        value = sql_gen.as_Code(value)
69
        
70
        self.srcs = srcs
71
        self.errors_table = errors_table
72
        self.value = value
73
    
74
    def to_str(self, db):
75
        if not self.srcs or self.errors_table == None:
76
            return sql_gen.ExcToWarning.to_str(self, db)
77
        
78
        errors_table_cols = map(sql_gen.Col,
79
            ['column', 'value', 'error_code', 'error'])
80
        col_names_query = sql.mk_select(db, sql_gen.NamedValues('c', None,
81
            [[c.name] for c in self.srcs]), order_by=None)
82
        insert_query = sql.mk_insert_select(db, self.errors_table,
83
            errors_table_cols,
84
            sql_gen.Values(errors_table_cols).to_str(db))+';\n'
85
        return '''\
86
-- Save error in errors table.
87
DECLARE
88
    error_code text := SQLSTATE;
89
    error text := SQLERRM;
90
    value text := '''+self.value.to_str(db)+''';
91
    "column" text;
92
BEGIN
93
    -- Insert the value and error for *each* source column.
94
'''+strings.indent(sql_gen.RowExcIgnore(None, col_names_query, insert_query,
95
    row_var=errors_table_cols[0]).to_str(db))+'''
96
END;
97

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

    
100
def data_exception_handler(*args, **kw_args):
101
    '''Handles a data_exception by saving it or converting it to a warning.
102
    For params, see ExcToErrorsTable().
103
    '''
104
    return sql_gen.data_exception_handler(ExcToErrorsTable(*args, **kw_args))
105

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

    
146
def func_wrapper_exception_handler(db, return_, args, errors_table):
147
    '''Handles a function call's data_exceptions.
148
    Supports PL/Python functions.
149
    @param return_ See data_exception_handler()
150
    @param args [arg...] Function call's args
151
    @param errors_table See data_exception_handler()
152
    '''
153
    args = filter(sql_gen.has_srcs, args)
154
    
155
    srcs = sql_gen.cross_join_srcs(args)
156
    value = sql_gen.merge_not_null(db, ',', args)
157
    return sql_gen.NestedExcHandler(
158
        data_exception_handler(return_, srcs, errors_table, value)
159
        , sql_gen.plpythonu_error_handler
160
        )
161

    
162
def cast_temp_col(db, type_, col, errors_table=None):
163
    '''Like cast(), but creates a new column with the cast values if the input
164
    is a column.
165
    @return The new column or cast value
166
    '''
167
    def cast_(col): return cast(db, type_, col, errors_table)
168
    
169
    try: col = sql_gen.underlying_col(col)
170
    except sql_gen.NoUnderlyingTableException: return sql_gen.wrap(cast_, col)
171
    
172
    table = col.table
173
    new_col = sql_gen.suffixed_col(col, '::'+strings.first_word(type_))
174
    expr = cast_(col)
175
    
176
    # Add column
177
    new_typed_col = sql_gen.TypedCol(new_col.name, type_)
178
    sql.add_col(db, table, new_typed_col, comment=strings.urepr(col)+'::'+type_)
179
    new_col.name = new_typed_col.name # propagate any renaming
180
    
181
    sql.update(db, table, [(new_col, expr)], in_place=True, recover=True)
182
    
183
    return new_col
184

    
185
def errors_table(db, table, if_exists=True):
186
    '''
187
    @param if_exists If set, returns None if the errors table doesn't exist
188
    @return None|sql_gen.Table
189
    '''
190
    table = sql_gen.as_Table(table)
191
    if table.srcs != (): table = table.srcs[0]
192
    
193
    errors_table = sql_gen.suffixed_table(table, '.errors')
194
    if if_exists and not sql.table_exists(db, errors_table): return None
195
    return errors_table
196

    
197
def mk_errors_table(db, table):
198
    errors_table_ = errors_table(db, table, if_exists=False)
199
    if sql.table_exists(db, errors_table_): return
200
    
201
    typed_cols = [
202
        sql_gen.TypedCol('column', 'text', nullable=False),
203
        sql_gen.TypedCol('value', 'text'),
204
        sql_gen.TypedCol('error_code', 'character varying(5)', nullable=False),
205
        sql_gen.TypedCol('error', 'text', nullable=False),
206
        ]
207
    sql.create_table(db, errors_table_, typed_cols, has_pkey=False)
208
    index_cols = ['column', 'value', 'error_code', 'error']
209
    sql.add_index(db, index_cols, errors_table_, unique=True)
210

    
211
##### Import
212

    
213
def put(db, table, row, pkey_=None, row_ct_ref=None):
214
    '''Recovers from errors.
215
    Only works under PostgreSQL (uses INSERT RETURNING).
216
    '''
217
    return put_table(db, table, [], row, row_ct_ref)
218

    
219
def get(db, table, row, pkey, row_ct_ref=None, create=False):
220
    '''Recovers from errors'''
221
    try:
222
        return sql.value(sql.select(db, table, [pkey], row, limit=1,
223
            recover=True))
224
    except StopIteration:
225
        if not create: raise
226
        return put(db, table, row, pkey, row_ct_ref) # insert new row
227

    
228
def is_func_result(col):
229
    return col.table.name.find('(') >= 0 and col.name == 'result'
230

    
231
def into_table_name(out_table, in_tables0, mapping, is_func):
232
    def in_col_str(in_col):
233
        in_col = sql_gen.remove_col_rename(in_col)
234
        if isinstance(in_col, sql_gen.Col):
235
            table = in_col.table
236
            if table == in_tables0:
237
                in_col = sql_gen.to_name_only_col(in_col)
238
            elif is_func_result(in_col): in_col = table # omit col name
239
        return strings.ustr(in_col)
240
    
241
    str_ = str(out_table)
242
    if is_func:
243
        str_ += '('
244
        
245
        try: value_in_col = mapping['value']
246
        except KeyError:
247
            str_ += ', '.join((str(k)+'='+in_col_str(v)
248
                for k, v in mapping.iteritems()))
249
        else: str_ += in_col_str(value_in_col)
250
        
251
        str_ += ')'
252
    else:
253
        out_col = 'rank'
254
        try: in_col = mapping[out_col]
255
        except KeyError: str_ += '_pkeys'
256
        else: # has a rank column, so hierarchical
257
            str_ += '['+str(out_col)+'='+in_col_str(in_col)+']'
258
    return str_
259

    
260
def put_table(db, out_table, in_tables, mapping, row_ct_ref=None, default=None,
261
    col_defaults={}, on_error=exc.reraise):
262
    '''Recovers from errors.
263
    Only works under PostgreSQL (uses INSERT RETURNING).
264
    IMPORTANT: Must be run at the *beginning* of a transaction.
265
    @param in_tables The main input table to select from, followed by a list of
266
        tables to join with it using the main input table's pkey
267
    @param mapping dict(out_table_col=in_table_col, ...)
268
        * out_table_col: str (*not* sql_gen.Col)
269
        * in_table_col: sql_gen.Col|literal-value
270
    @param default The *output* column to use as the pkey for missing rows.
271
        If this output column does not exist in the mapping, uses None.
272
    @param col_defaults Default values for required columns.
273
    @return sql_gen.Col Where the output pkeys are made available
274
    '''
275
    import psycopg2.extensions
276
    
277
    out_table = sql_gen.as_Table(out_table)
278
    
279
    def log_debug(msg): db.log_debug(msg, level=1.5)
280
    def col_ustr(str_):
281
        return strings.repr_no_u(sql_gen.remove_col_rename(str_))
282
    
283
    log_debug('********** New iteration **********')
284
    log_debug('Inserting these input columns into '+strings.as_tt(
285
        out_table.to_str(db))+':\n'+strings.as_table(mapping, ustr=col_ustr))
286
    
287
    is_function = sql.function_exists(db, out_table)
288
    
289
    if is_function: out_pkey = 'result'
290
    else: out_pkey = sql.pkey(db, out_table, recover=True)
291
    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)
292
    
293
    in_tables_ = in_tables[:] # don't modify input!
294
    try: in_tables0 = in_tables_.pop(0) # first table is separate
295
    except IndexError: in_tables0 = None
296
    else:
297
        in_pkey = sql.pkey(db, in_tables0, recover=True)
298
        in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)
299
    
300
    # Determine if can use optimization for only literal values
301
    is_literals = not reduce(operator.or_, map(sql_gen.is_table_col,
302
        mapping.values()), False)
303
    is_literals_or_function = is_literals or is_function
304
    
305
    if in_tables0 == None: errors_table_ = None
306
    else: errors_table_ = errors_table(db, in_tables0)
307
    
308
    # Create input joins from list of input tables
309
    input_joins = [in_tables0]+[sql_gen.Join(v,
310
        {in_pkey: sql_gen.join_same_not_null}) for v in in_tables_]
311
    
312
    if mapping == {} and not is_function: # need >= one column for INSERT SELECT
313
        mapping = {out_pkey: None} # ColDict will replace with default value
314
    
315
    if not is_literals:
316
        into = sql_gen.as_Table(into_table_name(out_table, in_tables0, mapping,
317
            is_function))
318
        
319
        # Set column sources
320
        in_cols = filter(sql_gen.is_table_col, mapping.values())
321
        for col in in_cols:
322
            if col.table == in_tables0: col.set_srcs(sql_gen.src_self)
323
        
324
        log_debug('Joining together input tables into temp table')
325
        # Place in new table so don't modify input and for speed
326
        in_table = sql_gen.Table('in')
327
        mapping = dicts.join(mapping, sql.flatten(db, in_table, input_joins,
328
            in_cols, preserve=[in_pkey_col]))
329
        input_joins = [in_table]
330
        db.log_debug('Temp table: '+strings.as_tt(in_table.to_str(db)), level=2)
331
    
332
    # Wrap mapping in a sql_gen.ColDict.
333
    # sql_gen.ColDict sanitizes both keys and values passed into it.
334
    # Do after applying dicts.join() because that returns a plain dict.
335
    mapping = sql_gen.ColDict(db, out_table, mapping)
336
    
337
    # Resolve default value column
338
    if default != None:
339
        try: default = mapping[default]
340
        except KeyError:
341
            db.log_debug('Default value column '
342
                +strings.as_tt(strings.repr_no_u(default))
343
                +' does not exist in mapping, falling back to None', level=2.1)
344
            default = None
345
    
346
    # Save default values for all rows since in_table may have rows deleted
347
    if is_literals: pass
348
    elif is_function: full_in_table = in_table
349
    else:
350
        full_in_table = sql_gen.suffixed_table(in_table, '_full')
351
        full_in_table_cols = [in_pkey_col]
352
        if default != None:
353
            full_in_table_cols.append(default)
354
            default = sql_gen.with_table(default, full_in_table)
355
        sql.run_query_into(db, sql.mk_select(db, in_table, full_in_table_cols,
356
            order_by=None), into=full_in_table, add_pkey_=True)
357
    
358
    if not is_literals:
359
        pkeys_names = [in_pkey, out_pkey]
360
        pkeys_cols = [in_pkey_col, out_pkey_col]
361
    
362
    pkeys_table_exists_ref = [False]
363
    def insert_into_pkeys(joins, cols=None, limit=None, **kw_args):
364
        query = sql.mk_select(db, joins, cols, order_by=None, limit=limit)
365
        if pkeys_table_exists_ref[0]:
366
            sql.insert_select(db, into, pkeys_names, query, **kw_args)
367
        else:
368
            sql.run_query_into(db, query, into=into, add_pkey_=True, **kw_args)
369
            pkeys_table_exists_ref[0] = True
370
    
371
    limit_ref = [None]
372
    def mk_main_select(joins, cols):
373
        return sql.mk_select(db, joins, cols, limit=limit_ref[0], order_by=None)
374
    
375
    if is_literals: insert_in_table = None
376
    else:
377
        insert_in_table = in_table
378
        insert_in_tables = [insert_in_table]
379
    join_cols = sql_gen.ColDict(db, out_table)
380
    
381
    exc_strs = set()
382
    def log_exc(e):
383
        e_str = exc.str_(e, first_line_only=True)
384
        log_debug('Caught exception: '+e_str)
385
        if e_str in exc_strs: # avoid infinite loops
386
            log_debug('Exception already seen, handler broken')
387
            on_error(e)
388
            remove_all_rows()
389
        else: exc_strs.add(e_str)
390
    
391
    def remove_all_rows():
392
        log_debug('Ignoring all rows')
393
        limit_ref[0] = 0 # just create an empty pkeys table
394
    
395
    def ignore_cond(cond, e):
396
        if is_literals: remove_all_rows()
397
        else:
398
            out_table_cols = sql_gen.ColDict(db, out_table)
399
            out_table_cols.update(util.dict_subset_right_join({},
400
                sql.table_cols(db, out_table)))
401
            
402
            in_cols = []
403
            cond = sql.map_expr(db, cond, mapping, in_cols)
404
            cond = sql.map_expr(db, cond, out_table_cols)
405
            
406
            track_data_error(db, errors_table_, sql_gen.cols_srcs(in_cols),
407
                None, e.cause.pgcode,
408
                strings.ensure_newl(e.cause.pgerror)+'condition: '+cond)
409
            
410
            not_cond = sql_gen.NotCond(sql_gen.CustomCode(cond))
411
            log_debug('Ignoring rows where '+strings.as_tt(not_cond.to_str(db)))
412
            sql.delete(db, insert_in_table, not_cond)
413
    
414
    not_null_cols = set()
415
    def ignore(in_col, value, e):
416
        if sql_gen.is_table_col(in_col):
417
            in_col = sql_gen.with_table(in_col, insert_in_table)
418
            
419
            track_data_error(db, errors_table_, in_col.srcs, value,
420
                e.cause.pgcode, e.cause.pgerror)
421
            
422
            sql.add_index(db, in_col, insert_in_table) # enable fast filtering
423
            if value != None and in_col not in not_null_cols:
424
                log_debug('Replacing invalid value '+strings.as_tt(repr(value))
425
                    +' with NULL in column '+strings.as_tt(in_col.to_str(db)))
426
                sql.update(db, insert_in_table, [(in_col, None)],
427
                    sql_gen.ColValueCond(in_col, value))
428
            else:
429
                log_debug('Ignoring rows with '+strings.as_tt(in_col.to_str(db))
430
                    +' = '+strings.as_tt(repr(value)))
431
                sql.delete(db, insert_in_table,
432
                    sql_gen.ColValueCond(in_col, value))
433
                if value == None: not_null_cols.add(in_col)
434
        else:
435
            assert isinstance(in_col, sql_gen.NamedCol)
436
            in_value = sql_gen.remove_col_rename(in_col)
437
            assert sql_gen.is_literal(in_value)
438
            if value == in_value.value:
439
                if value != None:
440
                    log_debug('Replacing invalid literal '
441
                        +strings.as_tt(in_col.to_str(db))+' with NULL')
442
                    mapping[in_col.name] = None
443
                else:
444
                    remove_all_rows()
445
            # otherwise, all columns were being ignore()d because the specific
446
            # column couldn't be identified, and this was not the invalid column
447
    
448
    if not is_literals:
449
        def insert_pkeys_table(which):
450
            return sql_gen.Table(sql_gen.concat(in_table.name,
451
                '_insert_'+which+'_pkeys'))
452
        insert_out_pkeys = insert_pkeys_table('out')
453
        insert_in_pkeys = insert_pkeys_table('in')
454
    
455
    def mk_func_call():
456
        args = dict(((k.name, v) for k, v in mapping.iteritems()))
457
        func_call = sql_gen.NamedCol(out_pkey,
458
            sql_gen.FunctionCall(out_table, **args))
459
        return func_call, args
460
        
461
    if is_function and not is_literals:
462
        log_debug('Defining wrapper function')
463
        
464
        func_call, args = mk_func_call()
465
        
466
        # Create empty pkeys table so its row type can be used
467
        insert_into_pkeys(input_joins, [in_pkey_col, func_call], limit=0,
468
            recover=True)
469
        result_type = db.col_info(sql_gen.Col(out_pkey, into)).type
470
        
471
        ## Create error handling wrapper function
472
        
473
        wrapper = db.TempFunction(sql_gen.concat(into.name, '_wrap'))
474
        
475
        select_cols = [in_pkey_col]+args.values()
476
        row_var = copy.copy(sql_gen.row_var)
477
        row_var.set_srcs([in_table])
478
        in_pkey_var = sql_gen.Col(in_pkey, row_var)
479
        
480
        args = dict(((k, sql_gen.with_table(v, row_var))
481
            for k, v in args.iteritems()))
482
        func_call = sql_gen.FunctionCall(out_table, **args)
483
        
484
        def mk_return(result):
485
            return sql_gen.ReturnQuery(sql.mk_select(db,
486
                fields=[in_pkey_var, result], explain=False))
487
        exc_handler = func_wrapper_exception_handler(db,
488
            mk_return(sql_gen.Cast(result_type, None)), args.values(),
489
            errors_table_)
490
        
491
        sql.define_func(db, sql_gen.FunctionDef(wrapper, sql_gen.SetOf(into),
492
            sql_gen.RowExcIgnore(sql_gen.RowType(in_table),
493
                sql.mk_select(db, input_joins, order_by=None),
494
                mk_return(func_call), exc_handler=exc_handler)
495
            ))
496
        wrapper_table = sql_gen.FunctionCall(wrapper)
497
    
498
    # Do inserts and selects
499
    while True:
500
        has_joins = join_cols != {}
501
        
502
        # Handle unrecoverable errors in a special case
503
        if limit_ref[0] == 0:
504
            if is_literals or default == None:
505
                default = sql_gen.remove_col_rename(default)
506
                log_debug('Returning default: '+strings.as_tt(repr(default)))
507
                return default
508
            elif is_function: pass # empty pkeys table already created
509
            else:
510
                log_debug('Creating an empty output pkeys table')
511
                has_joins = False # use the no-joins case
512
                cur = sql.run_query_into(db, sql.mk_select(db, out_table,
513
                    [out_pkey], order_by=None, limit=0), into=insert_out_pkeys)
514
            
515
            break # don't do main case
516
        
517
        # Prepare to insert new rows
518
        if is_function:
519
            log_debug('Calling function on input rows')
520
            if is_literals: func_call, args = mk_func_call()
521
        else:
522
            log_debug('Trying to insert new rows')
523
            insert_args = dict(recover=True, cacheable=False)
524
            if has_joins:
525
                insert_args.update(dict(ignore=True))
526
            else:
527
                insert_args.update(dict(returning=out_pkey))
528
                if not is_literals:
529
                    insert_args.update(dict(into=insert_out_pkeys))
530
            main_select = mk_main_select([insert_in_table], [sql_gen.with_table(
531
                c, insert_in_table) for c in mapping.values()])
532
        
533
        try:
534
            cur = None
535
            if is_function:
536
                if is_literals:
537
                    cur = sql.select(db, fields=[func_call], recover=True,
538
                        cacheable=True)
539
                else: insert_into_pkeys(wrapper_table, recover=True)
540
            else:
541
                cur = sql.insert_select(db, out_table, mapping.keys(),
542
                    main_select, **insert_args)
543
            break # insert successful
544
        except sql.MissingCastException, e:
545
            log_exc(e)
546
            
547
            type_ = e.type
548
            if e.col == None: out_cols = mapping.keys()
549
            else: out_cols = [e.col]
550
            
551
            for out_col in out_cols:
552
                log_debug('Casting '+strings.as_tt(strings.repr_no_u(out_col))
553
                    +' input to '+strings.as_tt(type_))
554
                in_col = mapping[out_col]
555
                while True:
556
                    try:
557
                        mapping[out_col] = cast_temp_col(db, type_, in_col,
558
                            errors_table_)
559
                        break # cast successful
560
                    except sql.InvalidValueException, e:
561
                        log_exc(e)
562
                        
563
                        ignore(in_col, e.value, e)
564
        except sql.DuplicateKeyException, e:
565
            log_exc(e)
566
            
567
            # Different rows violating different unique constraints not
568
            # supported
569
            assert not join_cols
570
            
571
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
572
            log_debug('Ignoring existing rows, comparing on these columns:\n'
573
                +strings.as_inline_table(join_cols, ustr=col_ustr))
574
            
575
            if is_literals:
576
                return sql.value(sql.select(db, out_table, [out_pkey_col],
577
                    join_cols, order_by=None))
578
            
579
            # Uniquify input table to avoid internal duplicate keys
580
            insert_in_table = sql.distinct_table(db, insert_in_table,
581
                join_cols.values())
582
            insert_in_tables.append(insert_in_table)
583
        except sql.NullValueException, e:
584
            log_exc(e)
585
            
586
            out_col, = e.cols
587
            try: in_col = mapping[out_col]
588
            except KeyError, e:
589
                try: in_col = mapping[out_col] = col_defaults[out_col]
590
                except KeyError:
591
                    msg = 'Missing mapping for NOT NULL column '+out_col
592
                    log_debug(msg)
593
                    if default == None: warnings.warn(UserWarning(msg))
594
                        # not an error because sometimes the mappings include
595
                        # extra tables which aren't used by the dataset
596
                    remove_all_rows()
597
            else: ignore(in_col, None, e)
598
        except sql.CheckException, e:
599
            log_exc(e)
600
            
601
            ignore_cond(e.cond, e)
602
        except sql.InvalidValueException, e:
603
            log_exc(e)
604
            
605
            for in_col in mapping.values(): ignore(in_col, e.value, e)
606
        except psycopg2.extensions.TransactionRollbackError, e:
607
            log_exc(e)
608
            # retry
609
        except sql.DatabaseErrors, e:
610
            log_exc(e)
611
            
612
            log_debug('No handler for exception')
613
            on_error(e)
614
            remove_all_rows()
615
        # after exception handled, rerun loop with additional constraints
616
    
617
    if cur != None and row_ct_ref != None and cur.rowcount >= 0:
618
        row_ct_ref[0] += cur.rowcount
619
    
620
    if is_literals: return sql.value(cur)
621
    
622
    if is_function: pass # pkeys table already created
623
    elif has_joins:
624
        select_joins = input_joins+[sql_gen.Join(out_table, join_cols)]
625
        log_debug('Getting output table pkeys of existing/inserted rows')
626
        insert_into_pkeys(select_joins, pkeys_cols)
627
    else:
628
        sql.add_row_num(db, insert_out_pkeys) # for joining with input pkeys
629
        
630
        log_debug('Getting input table pkeys of inserted rows')
631
        # Note that mk_main_select() does not use ORDER BY. Instead, assume that
632
        # since the SELECT query is identical to the one used in INSERT SELECT,
633
        # its rows will be retrieved in the same order.
634
        sql.run_query_into(db, mk_main_select(input_joins, [in_pkey]),
635
            into=insert_in_pkeys)
636
        sql.add_row_num(db, insert_in_pkeys) # for joining with output pkeys
637
        
638
        assert sql.table_row_count(db, insert_out_pkeys) == sql.table_row_count(
639
            db, insert_in_pkeys)
640
        
641
        log_debug('Combining output and input pkeys in inserted order')
642
        pkey_joins = [insert_in_pkeys, sql_gen.Join(insert_out_pkeys,
643
            {sql.row_num_col: sql_gen.join_same_not_null})]
644
        insert_into_pkeys(pkey_joins, pkeys_names)
645
        
646
        sql.empty_temp(db, [insert_out_pkeys, insert_in_pkeys])
647
    
648
    if limit_ref[0] == 0 or not is_function: # is_function doesn't leave holes
649
        log_debug('Setting pkeys of missing rows to '
650
            +strings.as_tt(repr(default)))
651
        missing_rows_joins = [full_in_table, sql_gen.Join(into,
652
            {in_pkey: sql_gen.join_same_not_null}, sql_gen.filter_out)]
653
            # must use join_same_not_null or query will take forever
654
        insert_into_pkeys(missing_rows_joins,
655
            [sql_gen.Col(in_pkey, full_in_table),
656
            sql_gen.NamedCol(out_pkey, default)])
657
    # otherwise, there is already an entry for every row
658
    
659
    assert (sql.table_row_count(db, into)
660
        == sql.table_row_count(db, full_in_table))
661
    
662
    sql.empty_temp(db, insert_in_tables+[full_in_table])
663
    
664
    srcs = []
665
    if is_function: srcs = sql_gen.cols_srcs(in_cols)
666
    return sql_gen.Col(out_pkey, into, srcs)
(29-29/40)