Project

General

Profile

1 3077 aaronmk
# Database import/export
2
3 3534 aaronmk
import copy
4 4995 aaronmk
import csv
5 3431 aaronmk
import operator
6 3714 aaronmk
import warnings
7 5588 aaronmk
import sys
8 3431 aaronmk
9 4995 aaronmk
import csvs
10 3077 aaronmk
import exc
11
import dicts
12
import sql
13
import sql_gen
14 5588 aaronmk
import streams
15 3077 aaronmk
import strings
16
import util
17
18 3645 aaronmk
##### Exceptions
19
20
# Can't use built-in SyntaxError because it stringifies to only the first line
21
class SyntaxError(Exception): pass
22
23 3081 aaronmk
##### Data cleanup
24
25 10187 aaronmk
def table_nulls_mapped__set(db, table):
26
    assert isinstance(table, sql_gen.Table)
27
    sql.run_query(db, 'SELECT util.table_nulls_mapped__set('
28
        +sql_gen.table2regclass_text(db, table)+')')
29
30
def table_nulls_mapped__get(db, table):
31
    assert isinstance(table, sql_gen.Table)
32
    return sql.value(sql.run_query(db, 'SELECT util.table_nulls_mapped__get('
33
        +sql_gen.table2regclass_text(db, table)+')'))
34
35 4915 aaronmk
null_strs = ['', '-', r'\N', 'NULL', 'UNKNOWN', 'nulo']
36 4209 aaronmk
37 4447 aaronmk
def cleanup_table(db, table):
38 10188 aaronmk
    '''idempotent'''
39 3081 aaronmk
    table = sql_gen.as_Table(table)
40 10195 aaronmk
    assert sql.table_exists(db, table)
41 10188 aaronmk
42
    if table_nulls_mapped__get(db, table): return # already cleaned up
43
44 5384 aaronmk
    cols = filter(lambda c: sql_gen.is_text_col(db, c),
45
        sql.table_cols(db, table))
46 5394 aaronmk
    try: pkey_col = sql.table_pkey_col(db, table)
47
    except sql.DoesNotExistException: pass
48
    else:
49
        try: cols.remove(pkey_col)
50
        except ValueError: pass
51 4928 aaronmk
    if not cols: return
52 3081 aaronmk
53 4993 aaronmk
    db.log_debug('Cleaning up table', level=1.5)
54
55 4209 aaronmk
    expr = 'trim(both from %s)'
56
    for null in null_strs: expr = 'nullif('+expr+', '+db.esc_value(null)+')'
57
    changes = [(v, sql_gen.CustomCode(expr % v.to_str(db))) for v in cols]
58 3081 aaronmk
59 4444 aaronmk
    while True:
60
        try:
61
            sql.update(db, table, changes, in_place=True, recover=True)
62
            break # successful
63
        except sql.NullValueException, e:
64 4457 aaronmk
            db.log_debug('Caught exception: '+exc.str_(e))
65 4444 aaronmk
            col, = e.cols
66
            sql.drop_not_null(db, col)
67 4992 aaronmk
68
    db.log_debug('Vacuuming and reanalyzing table', level=1.5)
69
    sql.vacuum(db, table)
70 10188 aaronmk
71
    table_nulls_mapped__set(db, table)
72 3081 aaronmk
73 3078 aaronmk
##### Error tracking
74
75
def track_data_error(db, errors_table, cols, value, error_code, error):
76
    '''
77
    @param errors_table If None, does nothing.
78
    '''
79 5817 aaronmk
    if errors_table == None: return
80 3078 aaronmk
81 5815 aaronmk
    col_names = [c.name for c in cols]
82
    if not col_names: col_names = [None] # need at least one entry
83
    for col_name in col_names:
84 3078 aaronmk
        try:
85 5815 aaronmk
            sql.insert(db, errors_table, dict(column=col_name, value=value,
86 3078 aaronmk
                error_code=error_code, error=error), recover=True,
87
                cacheable=True, log_level=4)
88
        except sql.DuplicateKeyException: pass
89
90 3506 aaronmk
class ExcToErrorsTable(sql_gen.ExcToWarning):
91
    '''Handles an exception by saving it or converting it to a warning.'''
92 3511 aaronmk
    def __init__(self, return_, srcs, errors_table, value=None):
93 3506 aaronmk
        '''
94
        @param return_ See sql_gen.ExcToWarning
95
        @param srcs The column names for the errors table
96
        @param errors_table None|sql_gen.Table
97 3511 aaronmk
        @param value The value (or an expression for it) that caused the error
98 3506 aaronmk
        @pre The invalid value must be in a local variable "value" of type text.
99
        '''
100
        sql_gen.ExcToWarning.__init__(self, return_)
101
102 3511 aaronmk
        value = sql_gen.as_Code(value)
103
104 3506 aaronmk
        self.srcs = srcs
105
        self.errors_table = errors_table
106 3511 aaronmk
        self.value = value
107 3501 aaronmk
108 3506 aaronmk
    def to_str(self, db):
109
        if not self.srcs or self.errors_table == None:
110
            return sql_gen.ExcToWarning.to_str(self, db)
111
112 3459 aaronmk
        errors_table_cols = map(sql_gen.Col,
113
            ['column', 'value', 'error_code', 'error'])
114 3465 aaronmk
        col_names_query = sql.mk_select(db, sql_gen.NamedValues('c', None,
115 3506 aaronmk
            [[c.name] for c in self.srcs]), order_by=None)
116
        insert_query = sql.mk_insert_select(db, self.errors_table,
117
            errors_table_cols,
118 3465 aaronmk
            sql_gen.Values(errors_table_cols).to_str(db))+';\n'
119 3506 aaronmk
        return '''\
120 3459 aaronmk
-- Save error in errors table.
121
DECLARE
122
    error_code text := SQLSTATE;
123
    error text := SQLERRM;
124 3511 aaronmk
    value text := '''+self.value.to_str(db)+''';
125 3529 aaronmk
    "column" text;
126 3459 aaronmk
BEGIN
127
    -- Insert the value and error for *each* source column.
128 3529 aaronmk
'''+strings.indent(sql_gen.RowExcIgnore(None, col_names_query, insert_query,
129 3467 aaronmk
    row_var=errors_table_cols[0]).to_str(db))+'''
130 3459 aaronmk
END;
131 3501 aaronmk
132 3506 aaronmk
'''+self.return_.to_str(db)
133 3459 aaronmk
134 3507 aaronmk
def data_exception_handler(*args, **kw_args):
135 3506 aaronmk
    '''Handles a data_exception by saving it or converting it to a warning.
136
    For params, see ExcToErrorsTable().
137
    '''
138
    return sql_gen.data_exception_handler(ExcToErrorsTable(*args, **kw_args))
139
140 3078 aaronmk
def cast(db, type_, col, errors_table=None):
141
    '''Casts an (unrenamed) column or value.
142
    If errors_table set and col has srcs, saves errors in errors_table (using
143 3360 aaronmk
    col's srcs attr as source columns). Otherwise, converts errors to warnings.
144 3078 aaronmk
    @param col str|sql_gen.Col|sql_gen.Literal
145
    @param errors_table None|sql_gen.Table|str
146
    '''
147
    col = sql_gen.as_Col(col)
148
149 3112 aaronmk
    # Don't convert exceptions to warnings for user-supplied constants
150
    if isinstance(col, sql_gen.Literal): return sql_gen.Cast(type_, col)
151
152 3078 aaronmk
    assert not isinstance(col, sql_gen.NamedCol)
153
154 3460 aaronmk
    function_name = strings.first_word(type_)
155 3459 aaronmk
    srcs = col.srcs
156 3508 aaronmk
    save_errors = errors_table != None and srcs
157
    if save_errors: # function will be unique for the given srcs
158 3750 aaronmk
        function_name = strings.ustr(sql_gen.FunctionCall(function_name,
159 3508 aaronmk
            *map(sql_gen.to_name_only_col, srcs)))
160 3078 aaronmk
    function = db.TempFunction(function_name)
161
162 3464 aaronmk
    # Create function definition
163
    modifiers = 'STRICT'
164
    if not save_errors: modifiers = 'IMMUTABLE '+modifiers
165 5791 aaronmk
    value_param = sql_gen.FunctionParam('value', 'anyelement')
166 3511 aaronmk
    handler = data_exception_handler('RETURN NULL;\n', srcs, errors_table,
167
        value_param.name)
168 3464 aaronmk
    body = sql_gen.CustomCode(handler.to_str(db, '''\
169 3467 aaronmk
/* The explicit cast to the return type is needed to make the cast happen
170
inside the try block. (Implicit casts to the return type happen at the end
171
of the function, outside any block.) */
172 6300 aaronmk
RETURN '''+sql_gen.Cast(type_, sql_gen.CustomCode('value')).to_str(db)+''';
173 3464 aaronmk
'''))
174
    body.lang='plpgsql'
175 3500 aaronmk
    sql.define_func(db, sql_gen.FunctionDef(function, type_, body,
176 3511 aaronmk
        [value_param], modifiers))
177 3464 aaronmk
178 3078 aaronmk
    return sql_gen.FunctionCall(function, col)
179
180 3538 aaronmk
def func_wrapper_exception_handler(db, return_, args, errors_table):
181 3524 aaronmk
    '''Handles a function call's data_exceptions.
182
    Supports PL/Python functions.
183
    @param return_ See data_exception_handler()
184
    @param args [arg...] Function call's args
185
    @param errors_table See data_exception_handler()
186
    '''
187
    args = filter(sql_gen.has_srcs, args)
188
189
    srcs = sql_gen.cross_join_srcs(args)
190 3538 aaronmk
    value = sql_gen.merge_not_null(db, ',', args)
191 3524 aaronmk
    return sql_gen.NestedExcHandler(
192
        data_exception_handler(return_, srcs, errors_table, value)
193
        , sql_gen.plpythonu_error_handler
194
        )
195
196 3078 aaronmk
def cast_temp_col(db, type_, col, errors_table=None):
197
    '''Like cast(), but creates a new column with the cast values if the input
198
    is a column.
199
    @return The new column or cast value
200
    '''
201
    def cast_(col): return cast(db, type_, col, errors_table)
202
203
    try: col = sql_gen.underlying_col(col)
204
    except sql_gen.NoUnderlyingTableException: return sql_gen.wrap(cast_, col)
205
206
    table = col.table
207 3173 aaronmk
    new_col = sql_gen.suffixed_col(col, '::'+strings.first_word(type_))
208 3078 aaronmk
    expr = cast_(col)
209
210
    # Add column
211
    new_typed_col = sql_gen.TypedCol(new_col.name, type_)
212 3750 aaronmk
    sql.add_col(db, table, new_typed_col, comment=strings.urepr(col)+'::'+type_)
213 3078 aaronmk
    new_col.name = new_typed_col.name # propagate any renaming
214
215 3110 aaronmk
    sql.update(db, table, [(new_col, expr)], in_place=True, recover=True)
216 3078 aaronmk
217
    return new_col
218
219
def errors_table(db, table, if_exists=True):
220
    '''
221
    @param if_exists If set, returns None if the errors table doesn't exist
222
    @return None|sql_gen.Table
223
    '''
224
    table = sql_gen.as_Table(table)
225
    if table.srcs != (): table = table.srcs[0]
226
227
    errors_table = sql_gen.suffixed_table(table, '.errors')
228
    if if_exists and not sql.table_exists(db, errors_table): return None
229
    return errors_table
230
231 4436 aaronmk
def mk_errors_table(db, table):
232
    errors_table_ = errors_table(db, table, if_exists=False)
233 4557 aaronmk
    if sql.table_exists(db, errors_table_, cacheable=False): return
234 4436 aaronmk
235
    typed_cols = [
236 5813 aaronmk
        sql_gen.TypedCol('column', 'text'),
237 4436 aaronmk
        sql_gen.TypedCol('value', 'text'),
238
        sql_gen.TypedCol('error_code', 'character varying(5)', nullable=False),
239
        sql_gen.TypedCol('error', 'text', nullable=False),
240
        ]
241
    sql.create_table(db, errors_table_, typed_cols, has_pkey=False)
242 8077 aaronmk
    index_cols = ['column', sql_gen.CustomCode('md5(value)'), 'error_code',
243
        sql_gen.CustomCode('md5(error)')]
244 4436 aaronmk
    sql.add_index(db, index_cols, errors_table_, unique=True)
245
246 3078 aaronmk
##### Import
247
248 5568 aaronmk
row_num_col_def = copy.copy(sql.row_num_col_def)
249
row_num_col_def.name = 'row_num'
250 5569 aaronmk
row_num_col_def.type = 'integer'
251 5568 aaronmk
252 5590 aaronmk
def append_csv(db, table, reader, header):
253 9508 aaronmk
    def esc_name_(name): return sql.esc_name(db, name)
254 5037 aaronmk
255 5017 aaronmk
    def log(msg, level=1): db.log_debug(msg, level)
256
257 5584 aaronmk
    # Wrap in standardizing stream
258 5590 aaronmk
    cols_ct = len(header)
259 5588 aaronmk
    stream = csvs.InputRewriter(streams.ProgressInputStream(csvs.StreamFilter(
260 5590 aaronmk
        csvs.ColCtFilter(reader, cols_ct)), sys.stderr, msg='Read %d row(s)',
261
        n=1000))
262 5584 aaronmk
    dialect = stream.dialect # use default dialect
263
264
    # Create COPY FROM statement
265 9508 aaronmk
    if header == sql.table_col_names(db, table): cols_str = ''
266
    else: cols_str =' ('+(', '.join(map(esc_name_, header)))+')'
267
    copy_from = ('COPY '+table.to_str(db)+cols_str+' FROM STDIN DELIMITER '
268 5584 aaronmk
        +db.esc_value(dialect.delimiter)+' NULL '+db.esc_value(''))
269
    assert not csvs.is_tsv(dialect)
270
    copy_from += ' CSV'
271
    if dialect.quoting != csv.QUOTE_NONE:
272
        quote_str = db.esc_value(dialect.quotechar)
273
        copy_from += ' QUOTE '+quote_str
274
        if dialect.doublequote: copy_from += ' ESCAPE '+quote_str
275
    copy_from += ';\n'
276
277
    log(copy_from, level=2)
278
    try: db.db.cursor().copy_expert(copy_from, stream)
279
    except Exception, e: sql.parse_exception(db, e, recover=True)
280 5017 aaronmk
281 5591 aaronmk
def import_csv(db, table, reader, header):
282 4995 aaronmk
    def log(msg, level=1): db.log_debug(msg, level)
283
284
    # Get format info
285 5590 aaronmk
    col_names = map(strings.to_unicode, header)
286 4995 aaronmk
    for i, col in enumerate(col_names): # replace empty column names
287
        if col == '': col_names[i] = 'column_'+str(i)
288
289
    # Select schema and escape names
290
    def esc_name(name): return db.esc_name(name)
291
292
    typed_cols = [sql_gen.TypedCol(v, 'text') for v in col_names]
293 5594 aaronmk
    typed_cols.insert(0, row_num_col_def)
294
    header.insert(0, row_num_col_def.name)
295
    reader = csvs.RowNumFilter(reader)
296 4995 aaronmk
297
    log('Creating table')
298 4999 aaronmk
    # Note that this is not rolled back if the import fails. Instead, it is
299
    # cached, and will not be re-run if the import is retried.
300 4995 aaronmk
    sql.create_table(db, table, typed_cols, has_pkey=False, col_indexes=False)
301
302 5001 aaronmk
    # Free memory used by deleted (rolled back) rows from any failed import.
303
    # This MUST be run so that the rows will be stored in inserted order, and
304
    # the row_num added after import will match up with the CSV's row order.
305 4999 aaronmk
    sql.truncate(db, table)
306
307 4995 aaronmk
    # Load the data
308 5590 aaronmk
    def load(): append_csv(db, table, reader, header)
309 5583 aaronmk
    sql.with_savepoint(db, load)
310 4995 aaronmk
311
    cleanup_table(db, table)
312
313 5719 aaronmk
def put(db, table, row, pkey_=None, row_ct_ref=None, on_error=exc.reraise):
314 3077 aaronmk
    '''Recovers from errors.
315
    Only works under PostgreSQL (uses INSERT RETURNING).
316
    '''
317 5719 aaronmk
    return put_table(db, table, [], row, row_ct_ref, on_error=on_error)
318 3077 aaronmk
319
def get(db, table, row, pkey, row_ct_ref=None, create=False):
320
    '''Recovers from errors'''
321
    try:
322
        return sql.value(sql.select(db, table, [pkey], row, limit=1,
323
            recover=True))
324
    except StopIteration:
325
        if not create: raise
326
        return put(db, table, row, pkey, row_ct_ref) # insert new row
327
328
def is_func_result(col):
329
    return col.table.name.find('(') >= 0 and col.name == 'result'
330
331
def into_table_name(out_table, in_tables0, mapping, is_func):
332
    def in_col_str(in_col):
333
        in_col = sql_gen.remove_col_rename(in_col)
334
        if isinstance(in_col, sql_gen.Col):
335
            table = in_col.table
336
            if table == in_tables0:
337
                in_col = sql_gen.to_name_only_col(in_col)
338
            elif is_func_result(in_col): in_col = table # omit col name
339 3750 aaronmk
        return strings.ustr(in_col)
340 3077 aaronmk
341 4491 aaronmk
    str_ = strings.ustr(out_table)
342 3077 aaronmk
    if is_func:
343
        str_ += '('
344
345
        try: value_in_col = mapping['value']
346
        except KeyError:
347 4491 aaronmk
            str_ += ', '.join((strings.ustr(k)+'='+in_col_str(v)
348 3077 aaronmk
                for k, v in mapping.iteritems()))
349
        else: str_ += in_col_str(value_in_col)
350
351
        str_ += ')'
352
    else:
353
        out_col = 'rank'
354
        try: in_col = mapping[out_col]
355
        except KeyError: str_ += '_pkeys'
356
        else: # has a rank column, so hierarchical
357 4491 aaronmk
            str_ += '['+strings.ustr(out_col)+'='+in_col_str(in_col)+']'
358 3077 aaronmk
    return str_
359
360 3628 aaronmk
def put_table(db, out_table, in_tables, mapping, row_ct_ref=None, default=None,
361 3660 aaronmk
    col_defaults={}, on_error=exc.reraise):
362 3077 aaronmk
    '''Recovers from errors.
363
    Only works under PostgreSQL (uses INSERT RETURNING).
364 7392 aaronmk
365 7395 aaronmk
    Warning: This function's normalizing algorithm does not support database
366 10164 aaronmk
    triggers that populate fields covered by the unique constraint used to do
367
    the DISTINCT ON. Such fields must be populated by the mappings instead.
368
    (Other unique constraints and other non-unique fields are not affected by
369
    this restriction on triggers. Note that the primary key will normally not be
370
    the DISTINCT ON constraint, so trigger-populated natural keys are supported
371
    *unless* the input table contains duplicate rows for some generated keys.)
372 7395 aaronmk
373
    Note that much of the complexity of the normalizing algorithm is due to
374
    PostgreSQL (and other DB systems) not having a native command for
375
    insert/on duplicate select. This operation is a cross between MySQL's
376
    INSERT ON DUPLICATE KEY UPDATE (which does not provide SELECT
377
    functionality), and PostgreSQL's INSERT RETURNING (which throws an error
378 10165 aaronmk
    on duplicate instead of returning the existing row).
379 10302 aaronmk
    However, PostgreSQL 9.1+ now provides a way to implement insert/on duplicate
380
    select just once for each table using the new INSTEAD OF triggers
381
    (http://www.postgresql.org/docs/9.1/static/plpgsql-trigger.html),
382
    which even support INSERT RETURNING. INSTEAD OF triggers can also be used to
383
    implement SQL functions without requiring special support by put_table().
384
    (INSTEAD OF triggers were not used when put_table() was developed, because
385
    it was necessary to support PostgreSQL 9.0.)
386 10165 aaronmk
387 3077 aaronmk
    @param in_tables The main input table to select from, followed by a list of
388
        tables to join with it using the main input table's pkey
389
    @param mapping dict(out_table_col=in_table_col, ...)
390
        * out_table_col: str (*not* sql_gen.Col)
391
        * in_table_col: sql_gen.Col|literal-value
392
    @param default The *output* column to use as the pkey for missing rows.
393
        If this output column does not exist in the mapping, uses None.
394 3618 aaronmk
    @param col_defaults Default values for required columns.
395 3077 aaronmk
    @return sql_gen.Col Where the output pkeys are made available
396
    '''
397 3474 aaronmk
    import psycopg2.extensions
398
399 6220 aaronmk
    # Special handling for functions with hstore params
400
    if out_table == '_map':
401
        import psycopg2.extras
402
        psycopg2.extras.register_hstore(db.db)
403
404
        # Parse args
405
        try: value = mapping.pop('value')
406
        except KeyError: return None # value required
407
408 6226 aaronmk
        mapping = dict([(k, sql_gen.get_value(v))
409
            for k, v in mapping.iteritems()]) # unwrap literal value
410 6220 aaronmk
        mapping = dict(map=mapping, value=value) # non-value params -> hstore
411
412 3077 aaronmk
    out_table = sql_gen.as_Table(out_table)
413
414
    def log_debug(msg): db.log_debug(msg, level=1.5)
415
    def col_ustr(str_):
416
        return strings.repr_no_u(sql_gen.remove_col_rename(str_))
417
418
    log_debug('********** New iteration **********')
419
    log_debug('Inserting these input columns into '+strings.as_tt(
420
        out_table.to_str(db))+':\n'+strings.as_table(mapping, ustr=col_ustr))
421
422
    is_function = sql.function_exists(db, out_table)
423
424 5192 aaronmk
    if is_function: row_ct_ref = None # only track inserted rows
425
426 4984 aaronmk
    # Warn if inserting empty table rows
427
    if not mapping and not is_function: # functions with no args OK
428
        warnings.warn(UserWarning('Inserting empty table row(s)'))
429
430 3077 aaronmk
    if is_function: out_pkey = 'result'
431 5388 aaronmk
    else: out_pkey = sql.pkey_name(db, out_table, recover=True)
432 3077 aaronmk
    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)
433
434 5069 aaronmk
    in_tables_ = copy.copy(in_tables) # don't modify input!
435 3432 aaronmk
    try: in_tables0 = in_tables_.pop(0) # first table is separate
436
    except IndexError: in_tables0 = None
437
    else:
438 5388 aaronmk
        in_pkey = sql.pkey_name(db, in_tables0, recover=True)
439 3432 aaronmk
        in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)
440 3431 aaronmk
441
    # Determine if can use optimization for only literal values
442
    is_literals = not reduce(operator.or_, map(sql_gen.is_table_col,
443 3434 aaronmk
        mapping.values()), False)
444 3431 aaronmk
    is_literals_or_function = is_literals or is_function
445
446 3432 aaronmk
    if in_tables0 == None: errors_table_ = None
447
    else: errors_table_ = errors_table(db, in_tables0)
448 3431 aaronmk
449
    # Create input joins from list of input tables
450 3077 aaronmk
    input_joins = [in_tables0]+[sql_gen.Join(v,
451
        {in_pkey: sql_gen.join_same_not_null}) for v in in_tables_]
452
453 5381 aaronmk
    orig_mapping = mapping.copy()
454 3433 aaronmk
    if mapping == {} and not is_function: # need >= one column for INSERT SELECT
455
        mapping = {out_pkey: None} # ColDict will replace with default value
456
457 3431 aaronmk
    if not is_literals:
458 3628 aaronmk
        into = sql_gen.as_Table(into_table_name(out_table, in_tables0, mapping,
459
            is_function))
460 5553 aaronmk
        # Ensure into's out_pkey is different from in_pkey by prepending "out."
461 4495 aaronmk
        if is_function: into_out_pkey = out_pkey
462 5553 aaronmk
        else: into_out_pkey = 'out.'+out_pkey
463 3431 aaronmk
464
        # Set column sources
465
        in_cols = filter(sql_gen.is_table_col, mapping.values())
466
        for col in in_cols:
467
            if col.table == in_tables0: col.set_srcs(sql_gen.src_self)
468
469
        log_debug('Joining together input tables into temp table')
470
        # Place in new table so don't modify input and for speed
471
        in_table = sql_gen.Table('in')
472
        mapping = dicts.join(mapping, sql.flatten(db, in_table, input_joins,
473
            in_cols, preserve=[in_pkey_col]))
474
        input_joins = [in_table]
475
        db.log_debug('Temp table: '+strings.as_tt(in_table.to_str(db)), level=2)
476 3077 aaronmk
477 3692 aaronmk
    # Wrap mapping in a sql_gen.ColDict.
478
    # sql_gen.ColDict sanitizes both keys and values passed into it.
479
    # Do after applying dicts.join() because that returns a plain dict.
480 3077 aaronmk
    mapping = sql_gen.ColDict(db, out_table, mapping)
481
482 5239 aaronmk
    # Save all rows since in_table may have rows deleted
483 3431 aaronmk
    if is_literals: pass
484
    elif is_function: full_in_table = in_table
485 3386 aaronmk
    else:
486
        full_in_table = sql_gen.suffixed_table(in_table, '_full')
487 5530 aaronmk
        sql.copy_table(db, in_table, full_in_table)
488 3287 aaronmk
489 3077 aaronmk
    pkeys_table_exists_ref = [False]
490 5444 aaronmk
    def insert_into_pkeys(query, **kw_args):
491 3077 aaronmk
        if pkeys_table_exists_ref[0]:
492 4484 aaronmk
            sql.insert_select(db, into, [in_pkey, into_out_pkey], query,
493
                **kw_args)
494 3077 aaronmk
        else:
495 6800 aaronmk
            kw_args.setdefault('add_pkey_', True)
496
497
            sql.run_query_into(db, query, into=into, **kw_args)
498 3077 aaronmk
            pkeys_table_exists_ref[0] = True
499
500 5523 aaronmk
    def mk_main_select(joins, cols): return sql.mk_select(db, joins, cols)
501 3418 aaronmk
502 3431 aaronmk
    if is_literals: insert_in_table = None
503
    else:
504
        insert_in_table = in_table
505
        insert_in_tables = [insert_in_table]
506 3352 aaronmk
    join_cols = sql_gen.ColDict(db, out_table)
507 3077 aaronmk
508
    exc_strs = set()
509
    def log_exc(e):
510
        e_str = exc.str_(e, first_line_only=True)
511
        log_debug('Caught exception: '+e_str)
512 3552 aaronmk
        if e_str in exc_strs: # avoid infinite loops
513
            log_debug('Exception already seen, handler broken')
514
            on_error(e)
515
            remove_all_rows()
516 5718 aaronmk
            return False
517 3552 aaronmk
        else: exc_strs.add(e_str)
518 5718 aaronmk
        return True
519 3077 aaronmk
520 5377 aaronmk
    ignore_all_ref = [False]
521 3077 aaronmk
    def remove_all_rows():
522
        log_debug('Ignoring all rows')
523 5377 aaronmk
        ignore_all_ref[0] = True # just return the default value column
524 3077 aaronmk
525 5889 aaronmk
    def handle_unknown_exc(e):
526
        log_debug('No handler for exception')
527
        on_error(e)
528
        remove_all_rows()
529
530 5504 aaronmk
    def ensure_cond(cond, e, passed=False, failed=False):
531 5449 aaronmk
        if is_literals: # we know the constraint was applied exactly once
532
            if passed: pass
533
            elif failed: remove_all_rows()
534
            else: raise NotImplementedError()
535 3704 aaronmk
        else:
536 5888 aaronmk
            if not is_function:
537
                out_table_cols = sql_gen.ColDict(db, out_table)
538
                out_table_cols.update(util.dict_subset_right_join({},
539
                    sql.table_col_names(db, out_table)))
540 3704 aaronmk
541
            in_cols = []
542 7117 aaronmk
            cond = strings.ustr(cond)
543 5818 aaronmk
            orig_cond = cond
544 5367 aaronmk
            cond = sql_gen.map_expr(db, cond, mapping, in_cols)
545 5888 aaronmk
            if not is_function:
546
                cond = sql_gen.map_expr(db, cond, out_table_cols)
547 3704 aaronmk
548 5442 aaronmk
            log_debug('Ignoring rows that do not satisfy '+strings.as_tt(cond))
549 5449 aaronmk
            cur = None
550
            if cond == sql_gen.false_expr:
551
                assert failed
552
                remove_all_rows()
553
            elif cond == sql_gen.true_expr: assert passed
554 5895 aaronmk
            else:
555
                while True:
556
                    not_cond = sql_gen.NotCond(sql_gen.CustomCode(cond))
557
                    try:
558
                        cur = sql.delete(db, insert_in_table, not_cond)
559
                        break
560
                    except sql.DoesNotExistException, e:
561
                        if e.type != 'column': raise
562
563
                        last_cond = cond
564
                        cond = sql_gen.map_expr(db, cond, {e.name: None})
565
                        if cond == last_cond: raise # not fixable
566 5449 aaronmk
567 5827 aaronmk
            # If any rows failed cond
568
            if failed or cur != None and cur.rowcount > 0:
569 5505 aaronmk
                track_data_error(db, errors_table_,
570
                    sql_gen.cross_join_srcs(in_cols), None, e.cause.pgcode,
571 7117 aaronmk
                    strings.ensure_newl(strings.ustr(e.cause.pgerror))
572
                    +'condition: '+orig_cond+'\ntranslated condition: '+cond)
573 3352 aaronmk
574 3294 aaronmk
    not_null_cols = set()
575 3077 aaronmk
    def ignore(in_col, value, e):
576 3630 aaronmk
        if sql_gen.is_table_col(in_col):
577
            in_col = sql_gen.with_table(in_col, insert_in_table)
578
579
            track_data_error(db, errors_table_, in_col.srcs, value,
580
                e.cause.pgcode, e.cause.pgerror)
581
582
            sql.add_index(db, in_col, insert_in_table) # enable fast filtering
583
            if value != None and in_col not in not_null_cols:
584 4492 aaronmk
                log_debug('Replacing invalid value '
585
                    +strings.as_tt(strings.urepr(value))+' with NULL in column '
586
                    +strings.as_tt(in_col.to_str(db)))
587 3630 aaronmk
                sql.update(db, insert_in_table, [(in_col, None)],
588
                    sql_gen.ColValueCond(in_col, value))
589
            else:
590 3637 aaronmk
                log_debug('Ignoring rows with '+strings.as_tt(in_col.to_str(db))
591 4492 aaronmk
                    +' = '+strings.as_tt(strings.urepr(value)))
592 3630 aaronmk
                sql.delete(db, insert_in_table,
593
                    sql_gen.ColValueCond(in_col, value))
594
                if value == None: not_null_cols.add(in_col)
595 3293 aaronmk
        else:
596 3630 aaronmk
            assert isinstance(in_col, sql_gen.NamedCol)
597 3684 aaronmk
            in_value = sql_gen.remove_col_rename(in_col)
598
            assert sql_gen.is_literal(in_value)
599
            if value == in_value.value:
600
                if value != None:
601
                    log_debug('Replacing invalid literal '
602
                        +strings.as_tt(in_col.to_str(db))+' with NULL')
603
                    mapping[in_col.name] = None
604
                else:
605
                    remove_all_rows()
606
            # otherwise, all columns were being ignore()d because the specific
607
            # column couldn't be identified, and this was not the invalid column
608 3077 aaronmk
609 3431 aaronmk
    if not is_literals:
610
        def insert_pkeys_table(which):
611
            return sql_gen.Table(sql_gen.concat(in_table.name,
612
                '_insert_'+which+'_pkeys'))
613
        insert_out_pkeys = insert_pkeys_table('out')
614
        insert_in_pkeys = insert_pkeys_table('in')
615 3077 aaronmk
616 3918 aaronmk
    def mk_func_call():
617 3550 aaronmk
        args = dict(((k.name, v) for k, v in mapping.iteritems()))
618 4484 aaronmk
        return sql_gen.FunctionCall(out_table, **args), args
619 3550 aaronmk
620 5239 aaronmk
    missing_msg = None
621
622 3077 aaronmk
    # Do inserts and selects
623
    while True:
624 3473 aaronmk
        has_joins = join_cols != {}
625
626 5377 aaronmk
        if ignore_all_ref[0]: break # unrecoverable error, so don't do main case
627 3077 aaronmk
628
        # Prepare to insert new rows
629 3918 aaronmk
        if is_function:
630 5726 aaronmk
            if is_literals:
631
                log_debug('Calling function')
632
                func_call, args = mk_func_call()
633 3077 aaronmk
        else:
634 3550 aaronmk
            log_debug('Trying to insert new rows')
635 3291 aaronmk
            insert_args = dict(recover=True, cacheable=False)
636
            if has_joins:
637
                insert_args.update(dict(ignore=True))
638
            else:
639 3431 aaronmk
                insert_args.update(dict(returning=out_pkey))
640
                if not is_literals:
641
                    insert_args.update(dict(into=insert_out_pkeys))
642 3291 aaronmk
            main_select = mk_main_select([insert_in_table], [sql_gen.with_table(
643
                c, insert_in_table) for c in mapping.values()])
644 3077 aaronmk
645 3292 aaronmk
        try:
646
            cur = None
647 3077 aaronmk
            if is_function:
648 3917 aaronmk
                if is_literals:
649
                    cur = sql.select(db, fields=[func_call], recover=True,
650
                        cacheable=True)
651 5444 aaronmk
                else:
652 5726 aaronmk
                    log_debug('Defining wrapper function')
653
654
                    func_call, args = mk_func_call()
655
                    func_call = sql_gen.NamedCol(into_out_pkey, func_call)
656
657
                    # Create empty pkeys table so its row type can be used
658
                    insert_into_pkeys(sql.mk_select(db, input_joins,
659 6801 aaronmk
                        [in_pkey_col, func_call], limit=0), add_pkey_=False,
660
                        recover=True)
661 5726 aaronmk
                    result_type = db.col_info(sql_gen.Col(into_out_pkey,
662
                        into)).type
663
664
                    ## Create error handling wrapper function
665
666
                    wrapper = db.TempFunction(sql_gen.concat(into.name,
667
                        '_wrap'))
668
669
                    select_cols = [in_pkey_col]+args.values()
670
                    row_var = copy.copy(sql_gen.row_var)
671
                    row_var.set_srcs([in_table])
672
                    in_pkey_var = sql_gen.Col(in_pkey, row_var)
673
674
                    args = dict(((k, sql_gen.with_table(v, row_var))
675
                        for k, v in args.iteritems()))
676
                    func_call = sql_gen.FunctionCall(out_table, **args)
677
678
                    def mk_return(result):
679
                        return sql_gen.ReturnQuery(sql.mk_select(db,
680
                            fields=[in_pkey_var, result], explain=False))
681
                    exc_handler = func_wrapper_exception_handler(db,
682
                        mk_return(sql_gen.Cast(result_type, None)),
683
                        args.values(), errors_table_)
684
685
                    sql.define_func(db, sql_gen.FunctionDef(wrapper,
686
                        sql_gen.SetOf(into),
687
                        sql_gen.RowExcIgnore(sql_gen.RowType(in_table),
688
                            sql.mk_select(db, input_joins),
689
                            mk_return(func_call), exc_handler=exc_handler)
690
                        ))
691
                    wrapper_table = sql_gen.FunctionCall(wrapper)
692
693
                    log_debug('Calling function')
694 5444 aaronmk
                    insert_into_pkeys(sql.mk_select(db, wrapper_table,
695 5894 aaronmk
                        order_by=None), recover=True, cacheable=False)
696 8820 aaronmk
                    sql.add_pkey_or_index(db, into)
697 3077 aaronmk
            else:
698 3292 aaronmk
                cur = sql.insert_select(db, out_table, mapping.keys(),
699 3077 aaronmk
                    main_select, **insert_args)
700
            break # insert successful
701
        except sql.MissingCastException, e:
702 5718 aaronmk
            if not log_exc(e): break
703 3077 aaronmk
704
            type_ = e.type
705 4140 aaronmk
            if e.col == None: out_cols = mapping.keys()
706
            else: out_cols = [e.col]
707 3077 aaronmk
708 4140 aaronmk
            for out_col in out_cols:
709 4171 aaronmk
                log_debug('Casting '+strings.as_tt(strings.repr_no_u(out_col))
710 4140 aaronmk
                    +' input to '+strings.as_tt(type_))
711
                in_col = mapping[out_col]
712
                while True:
713
                    try:
714
                        mapping[out_col] = cast_temp_col(db, type_, in_col,
715
                            errors_table_)
716
                        break # cast successful
717
                    except sql.InvalidValueException, e:
718 5718 aaronmk
                        if not log_exc(e): break
719 4140 aaronmk
720
                        ignore(in_col, e.value, e)
721 3077 aaronmk
        except sql.DuplicateKeyException, e:
722 5718 aaronmk
            if not log_exc(e): break
723 3077 aaronmk
724 3274 aaronmk
            # Different rows violating different unique constraints not
725
            # supported
726
            assert not join_cols
727
728 5504 aaronmk
            if e.cond != None: ensure_cond(e.cond, e, passed=True)
729 5450 aaronmk
730 3077 aaronmk
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
731
            log_debug('Ignoring existing rows, comparing on these columns:\n'
732
                +strings.as_inline_table(join_cols, ustr=col_ustr))
733 3102 aaronmk
734 3431 aaronmk
            if is_literals:
735
                return sql.value(sql.select(db, out_table, [out_pkey_col],
736 4025 aaronmk
                    join_cols, order_by=None))
737 3431 aaronmk
738 7180 aaronmk
            # Uniquify and filter input table to avoid (most) duplicate keys
739
            # (Additional duplicates may be added concurrently and will be
740
            # filtered out separately upon insert.)
741 3102 aaronmk
            insert_in_table = sql.distinct_table(db, insert_in_table,
742 7180 aaronmk
                join_cols.values(), [insert_in_table,
743
                sql_gen.Join(out_table, join_cols, sql_gen.filter_out)])
744 3144 aaronmk
            insert_in_tables.append(insert_in_table)
745 3077 aaronmk
        except sql.NullValueException, e:
746 5718 aaronmk
            if not log_exc(e): break
747 3077 aaronmk
748
            out_col, = e.cols
749
            try: in_col = mapping[out_col]
750 3618 aaronmk
            except KeyError, e:
751
                try: in_col = mapping[out_col] = col_defaults[out_col]
752
                except KeyError:
753 5239 aaronmk
                    missing_msg = 'Missing mapping for NOT NULL column '+out_col
754
                    log_debug(missing_msg)
755 3618 aaronmk
                    remove_all_rows()
756 3294 aaronmk
            else: ignore(in_col, None, e)
757 3352 aaronmk
        except sql.CheckException, e:
758 5718 aaronmk
            if not log_exc(e): break
759 3352 aaronmk
760 5504 aaronmk
            ensure_cond(e.cond, e, failed=True)
761 3413 aaronmk
        except sql.InvalidValueException, e:
762 5718 aaronmk
            if not log_exc(e): break
763 3413 aaronmk
764
            for in_col in mapping.values(): ignore(in_col, e.value, e)
765 3474 aaronmk
        except psycopg2.extensions.TransactionRollbackError, e:
766 5718 aaronmk
            if not log_exc(e): break
767 3474 aaronmk
            # retry
768 3077 aaronmk
        except sql.DatabaseErrors, e:
769 5718 aaronmk
            if not log_exc(e): break
770 3077 aaronmk
771 5889 aaronmk
            handle_unknown_exc(e)
772 3077 aaronmk
        # after exception handled, rerun loop with additional constraints
773
774 5239 aaronmk
    # Resolve default value column
775
    if default != None:
776 5381 aaronmk
        if ignore_all_ref[0]: mapping.update(orig_mapping) # use input cols
777 5239 aaronmk
        try: default = mapping[default]
778
        except KeyError:
779
            db.log_debug('Default value column '
780
                +strings.as_tt(strings.repr_no_u(default))
781
                +' does not exist in mapping, falling back to None', level=2.1)
782
            default = None
783 5380 aaronmk
        else: default = sql_gen.remove_col_rename(default)
784 5239 aaronmk
785
    if missing_msg != None and default == None:
786
        warnings.warn(UserWarning(missing_msg))
787
        # not an error because sometimes the mappings include
788
        # extra tables which aren't used by the dataset
789
790
    # Handle unrecoverable errors
791 5377 aaronmk
    if ignore_all_ref[0]:
792 5373 aaronmk
        log_debug('Returning default: '+strings.as_tt(strings.urepr(default)))
793
        return default
794 5239 aaronmk
795 3077 aaronmk
    if cur != None and row_ct_ref != None and cur.rowcount >= 0:
796
        row_ct_ref[0] += cur.rowcount
797
798 3530 aaronmk
    if is_literals: return sql.value(cur)
799
800
    if is_function: pass # pkeys table already created
801 3077 aaronmk
    elif has_joins:
802
        select_joins = input_joins+[sql_gen.Join(out_table, join_cols)]
803
        log_debug('Getting output table pkeys of existing/inserted rows')
804 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, select_joins, [in_pkey_col,
805
            sql_gen.NamedCol(into_out_pkey, out_pkey_col)], order_by=None))
806 3077 aaronmk
    else:
807
        sql.add_row_num(db, insert_out_pkeys) # for joining with input pkeys
808
809
        log_debug('Getting input table pkeys of inserted rows')
810 3285 aaronmk
        # Note that mk_main_select() does not use ORDER BY. Instead, assume that
811
        # since the SELECT query is identical to the one used in INSERT SELECT,
812
        # its rows will be retrieved in the same order.
813 3077 aaronmk
        sql.run_query_into(db, mk_main_select(input_joins, [in_pkey]),
814
            into=insert_in_pkeys)
815
        sql.add_row_num(db, insert_in_pkeys) # for joining with output pkeys
816
817
        assert sql.table_row_count(db, insert_out_pkeys) == sql.table_row_count(
818
            db, insert_in_pkeys)
819
820
        log_debug('Combining output and input pkeys in inserted order')
821
        pkey_joins = [insert_in_pkeys, sql_gen.Join(insert_out_pkeys,
822
            {sql.row_num_col: sql_gen.join_same_not_null})]
823 4484 aaronmk
        in_col = sql_gen.Col(in_pkey, insert_in_pkeys)
824
        out_col = sql_gen.NamedCol(into_out_pkey,
825
            sql_gen.Col(out_pkey, insert_out_pkeys))
826 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, pkey_joins, [in_col, out_col],
827
            order_by=None))
828 3077 aaronmk
829
        sql.empty_temp(db, [insert_out_pkeys, insert_in_pkeys])
830
831 5374 aaronmk
    if not is_function: # is_function doesn't leave holes
832 3187 aaronmk
        log_debug('Setting pkeys of missing rows to '
833 4492 aaronmk
            +strings.as_tt(strings.urepr(default)))
834 5993 aaronmk
835
        full_in_pkey_col = sql_gen.Col(in_pkey, full_in_table)
836 5380 aaronmk
        if sql_gen.is_table_col(default):
837
            default = sql_gen.with_table(default, full_in_table)
838 3287 aaronmk
        missing_rows_joins = [full_in_table, sql_gen.Join(into,
839 3187 aaronmk
            {in_pkey: sql_gen.join_same_not_null}, sql_gen.filter_out)]
840
            # must use join_same_not_null or query will take forever
841 5993 aaronmk
842
        insert_args = dict(order_by=None)
843
        if not sql.table_has_pkey(db, full_in_table): # in_table has duplicates
844
            insert_args.update(dict(distinct_on=[full_in_pkey_col]))
845
846 5444 aaronmk
        insert_into_pkeys(sql.mk_select(db, missing_rows_joins,
847 5993 aaronmk
            [full_in_pkey_col, sql_gen.NamedCol(into_out_pkey, default)],
848
            **insert_args))
849 3187 aaronmk
    # otherwise, there is already an entry for every row
850 3077 aaronmk
851 3530 aaronmk
    sql.empty_temp(db, insert_in_tables+[full_in_table])
852
853
    srcs = []
854 3619 aaronmk
    if is_function: srcs = sql_gen.cols_srcs(in_cols)
855 4484 aaronmk
    return sql_gen.Col(into_out_pkey, into, srcs)