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_ = strings.ustr(out_table)
242
    if is_func:
243
        str_ += '('
244
        
245
        try: value_in_col = mapping['value']
246
        except KeyError:
247
            str_ += ', '.join((strings.ustr(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_ += '['+strings.ustr(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 '
424
                    +strings.as_tt(strings.urepr(value))+' with NULL in column '
425
                    +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(strings.urepr(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
        return sql_gen.FunctionCall(out_table, **args), args
458
        
459
    if is_function and not is_literals:
460
        log_debug('Defining wrapper function')
461
        
462
        func_call, args = mk_func_call()
463
        func_call = sql_gen.NamedCol(into_out_pkey, func_call)
464
        
465
        # Create empty pkeys table so its row type can be used
466
        insert_into_pkeys(input_joins, [in_pkey_col, func_call], limit=0,
467
            recover=True)
468
        result_type = db.col_info(sql_gen.Col(into_out_pkey, into)).type
469
        
470
        ## Create error handling wrapper function
471
        
472
        wrapper = db.TempFunction(sql_gen.concat(into.name, '_wrap'))
473
        
474
        select_cols = [in_pkey_col]+args.values()
475
        row_var = copy.copy(sql_gen.row_var)
476
        row_var.set_srcs([in_table])
477
        in_pkey_var = sql_gen.Col(in_pkey, row_var)
478
        
479
        args = dict(((k, sql_gen.with_table(v, row_var))
480
            for k, v in args.iteritems()))
481
        func_call = sql_gen.FunctionCall(out_table, **args)
482
        
483
        def mk_return(result):
484
            return sql_gen.ReturnQuery(sql.mk_select(db,
485
                fields=[in_pkey_var, result], explain=False))
486
        exc_handler = func_wrapper_exception_handler(db,
487
            mk_return(sql_gen.Cast(result_type, None)), args.values(),
488
            errors_table_)
489
        
490
        sql.define_func(db, sql_gen.FunctionDef(wrapper, sql_gen.SetOf(into),
491
            sql_gen.RowExcIgnore(sql_gen.RowType(in_table),
492
                sql.mk_select(db, input_joins, order_by=None),
493
                mk_return(func_call), exc_handler=exc_handler)
494
            ))
495
        wrapper_table = sql_gen.FunctionCall(wrapper)
496
    
497
    # Do inserts and selects
498
    while True:
499
        has_joins = join_cols != {}
500
        
501
        # Handle unrecoverable errors in a special case
502
        if limit_ref[0] == 0:
503
            if is_literals or default == None:
504
                default = sql_gen.remove_col_rename(default)
505
                log_debug('Returning default: '
506
                    +strings.as_tt(strings.urepr(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, [in_pkey_col,
627
            sql_gen.NamedCol(into_out_pkey, out_pkey_col)])
628
    else:
629
        sql.add_row_num(db, insert_out_pkeys) # for joining with input pkeys
630
        
631
        log_debug('Getting input table pkeys of inserted rows')
632
        # Note that mk_main_select() does not use ORDER BY. Instead, assume that
633
        # since the SELECT query is identical to the one used in INSERT SELECT,
634
        # its rows will be retrieved in the same order.
635
        sql.run_query_into(db, mk_main_select(input_joins, [in_pkey]),
636
            into=insert_in_pkeys)
637
        sql.add_row_num(db, insert_in_pkeys) # for joining with output pkeys
638
        
639
        assert sql.table_row_count(db, insert_out_pkeys) == sql.table_row_count(
640
            db, insert_in_pkeys)
641
        
642
        log_debug('Combining output and input pkeys in inserted order')
643
        pkey_joins = [insert_in_pkeys, sql_gen.Join(insert_out_pkeys,
644
            {sql.row_num_col: sql_gen.join_same_not_null})]
645
        in_col = sql_gen.Col(in_pkey, insert_in_pkeys)
646
        out_col = sql_gen.NamedCol(into_out_pkey,
647
            sql_gen.Col(out_pkey, insert_out_pkeys))
648
        insert_into_pkeys(pkey_joins, [in_col, out_col])
649
        
650
        sql.empty_temp(db, [insert_out_pkeys, insert_in_pkeys])
651
    
652
    if limit_ref[0] == 0 or not is_function: # is_function doesn't leave holes
653
        log_debug('Setting pkeys of missing rows to '
654
            +strings.as_tt(strings.urepr(default)))
655
        missing_rows_joins = [full_in_table, sql_gen.Join(into,
656
            {in_pkey: sql_gen.join_same_not_null}, sql_gen.filter_out)]
657
            # must use join_same_not_null or query will take forever
658
        insert_into_pkeys(missing_rows_joins,
659
            [sql_gen.Col(in_pkey, full_in_table),
660
            sql_gen.NamedCol(into_out_pkey, default)])
661
    # otherwise, there is already an entry for every row
662
    
663
    assert (sql.table_row_count(db, into)
664
        == sql.table_row_count(db, full_in_table))
665
    
666
    sql.empty_temp(db, insert_in_tables+[full_in_table])
667
    
668
    srcs = []
669
    if is_function: srcs = sql_gen.cols_srcs(in_cols)
670
    return sql_gen.Col(into_out_pkey, into, srcs)
(29-29/40)