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