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