1 |
13
|
aaronmk
|
# XML-database conversion
|
2 |
|
|
|
3 |
2418
|
aaronmk
|
import copy
|
4 |
16
|
aaronmk
|
import re
|
5 |
13
|
aaronmk
|
from xml.dom import Node
|
6 |
|
|
|
7 |
446
|
aaronmk
|
import exc
|
8 |
2436
|
aaronmk
|
import Parser
|
9 |
46
|
aaronmk
|
import sql
|
10 |
3077
|
aaronmk
|
import sql_io
|
11 |
2268
|
aaronmk
|
import sql_gen
|
12 |
84
|
aaronmk
|
import strings
|
13 |
133
|
aaronmk
|
import util
|
14 |
46
|
aaronmk
|
import xml_dom
|
15 |
2113
|
aaronmk
|
import xml_func
|
16 |
2436
|
aaronmk
|
import xpath
|
17 |
13
|
aaronmk
|
|
18 |
139
|
aaronmk
|
def name_of(node): return re.sub(r'^.*\.', r'', node.tagName)
|
19 |
16
|
aaronmk
|
|
20 |
13
|
aaronmk
|
ptr_suffix = '_id'
|
21 |
|
|
|
22 |
138
|
aaronmk
|
def is_ptr(node_name): return node_name.lower().endswith(ptr_suffix)
|
23 |
13
|
aaronmk
|
|
24 |
172
|
aaronmk
|
def ptr_type_guess(node_name):
|
25 |
13
|
aaronmk
|
assert is_ptr(node_name)
|
26 |
|
|
return node_name[:-len(ptr_suffix)]
|
27 |
|
|
|
28 |
|
|
def ptr_target(node):
|
29 |
|
|
assert is_ptr(name_of(node))
|
30 |
2113
|
aaronmk
|
return xml_dom.value_node(node)
|
31 |
13
|
aaronmk
|
|
32 |
|
|
def find_by_name(node, name):
|
33 |
46
|
aaronmk
|
for parent in xml_dom.NodeParentIter(node):
|
34 |
13
|
aaronmk
|
if name_of(parent) == name: return parent
|
35 |
|
|
else:
|
36 |
46
|
aaronmk
|
for child in xml_dom.NodeElemIter(parent):
|
37 |
16
|
aaronmk
|
child_name = name_of(child)
|
38 |
13
|
aaronmk
|
if is_ptr(child_name):
|
39 |
172
|
aaronmk
|
target = ptr_target(child)
|
40 |
|
|
if target.tagName == name: return target
|
41 |
13
|
aaronmk
|
elif child_name == name: return child
|
42 |
|
|
return None
|
43 |
|
|
|
44 |
1850
|
aaronmk
|
def get(db, node, limit=None, start=None):
|
45 |
|
|
def pkey(table): return sql.pkey(db, table)
|
46 |
126
|
aaronmk
|
|
47 |
141
|
aaronmk
|
node = node.firstChild
|
48 |
135
|
aaronmk
|
table = name_of(node)
|
49 |
|
|
pkey_ = pkey(table)
|
50 |
|
|
|
51 |
|
|
fields = []
|
52 |
|
|
conds = {}
|
53 |
133
|
aaronmk
|
for child in xml_dom.NodeElemIter(node):
|
54 |
135
|
aaronmk
|
child_name = name_of(child)
|
55 |
|
|
if xml_dom.is_empty(child): fields.append(child_name)
|
56 |
|
|
elif xml_dom.is_text(child): conds[child_name] = xml_dom.value(child)
|
57 |
|
|
else: raise Exception('Joins not supported yet')
|
58 |
|
|
id_ = xml_dom.get_id(node)
|
59 |
1836
|
aaronmk
|
if id_ != None: conds[pkey(table)] = id_ # replace any existing pkey value
|
60 |
135
|
aaronmk
|
if fields == []: fields.append(pkey_)
|
61 |
133
|
aaronmk
|
|
62 |
864
|
aaronmk
|
return sql.select(db, table, fields, conds, limit, start)
|
63 |
126
|
aaronmk
|
|
64 |
1864
|
aaronmk
|
def put(db, node, row_ct_ref=None, on_error=exc.raise_, pool=None,
|
65 |
|
|
store_ids=False, parent_id=None):
|
66 |
138
|
aaronmk
|
'''store_ids enables searching the tree for missing fields'''
|
67 |
1850
|
aaronmk
|
def pkey(table): return sql.pkey(db, table, True)
|
68 |
15
|
aaronmk
|
|
69 |
461
|
aaronmk
|
def put_(node, parent_id=None):
|
70 |
1874
|
aaronmk
|
args = (db, node, row_ct_ref, on_error, pool, store_ids, parent_id)
|
71 |
|
|
if parent_id != None and pool != None: pool.apply_async(put, args)
|
72 |
|
|
else: return put(*args)
|
73 |
461
|
aaronmk
|
|
74 |
|
|
def on_error_(e):
|
75 |
446
|
aaronmk
|
exc.add_msg(e, 'node:\n'+str(node))
|
76 |
461
|
aaronmk
|
on_error(e)
|
77 |
446
|
aaronmk
|
|
78 |
48
|
aaronmk
|
table = name_of(node)
|
79 |
446
|
aaronmk
|
try: pkey_ = pkey(table)
|
80 |
461
|
aaronmk
|
except sql.DatabaseErrors, e: on_error_(e); return None
|
81 |
48
|
aaronmk
|
row = {}
|
82 |
|
|
children = []
|
83 |
13
|
aaronmk
|
|
84 |
48
|
aaronmk
|
# Divide children into fields and children with fkeys to parent
|
85 |
|
|
for child in xml_dom.NodeElemIter(node):
|
86 |
|
|
child_name = name_of(child)
|
87 |
463
|
aaronmk
|
if xml_dom.is_empty(child): row[child_name] = None
|
88 |
454
|
aaronmk
|
elif xml_dom.is_text(child):
|
89 |
84
|
aaronmk
|
row[child_name] = strings.to_unicode(xml_dom.value(child))
|
90 |
461
|
aaronmk
|
elif is_ptr(child_name): row[child_name] = put_(ptr_target(child))
|
91 |
48
|
aaronmk
|
else: children.append(child)
|
92 |
|
|
|
93 |
|
|
# Add fkey to parent
|
94 |
59
|
aaronmk
|
if parent_id != None:
|
95 |
|
|
parent_ptr = node.getAttribute('fkey')
|
96 |
|
|
if parent_ptr == '': parent_ptr = pkey(name_of(node.parentNode))
|
97 |
|
|
row[parent_ptr] = parent_id
|
98 |
48
|
aaronmk
|
|
99 |
|
|
# Insert node
|
100 |
446
|
aaronmk
|
try:
|
101 |
|
|
for try_num in xrange(2):
|
102 |
|
|
try:
|
103 |
3077
|
aaronmk
|
id_ = sql_io.put(db, table, row, pkey_, row_ct_ref)
|
104 |
446
|
aaronmk
|
if store_ids: xml_dom.set_id(node, id_)
|
105 |
|
|
break
|
106 |
|
|
except sql.NullValueException, e:
|
107 |
468
|
aaronmk
|
col = e.cols[0]
|
108 |
446
|
aaronmk
|
if try_num > 0: raise # exception still raised after retry
|
109 |
468
|
aaronmk
|
if store_ids and is_ptr(col):
|
110 |
446
|
aaronmk
|
# Search for required column in ancestors and their children
|
111 |
468
|
aaronmk
|
target = find_by_name(node, ptr_type_guess(col))
|
112 |
446
|
aaronmk
|
if target == None: raise
|
113 |
468
|
aaronmk
|
row[col] = xml_dom.get_id(target)
|
114 |
446
|
aaronmk
|
else: raise
|
115 |
461
|
aaronmk
|
except sql.DatabaseErrors, e: on_error_(e); return None
|
116 |
48
|
aaronmk
|
|
117 |
|
|
# Insert children with fkeys to parent
|
118 |
1874
|
aaronmk
|
for child in children: put_(child, id_)
|
119 |
48
|
aaronmk
|
|
120 |
|
|
return id_
|
121 |
1996
|
aaronmk
|
|
122 |
2039
|
aaronmk
|
class ColRef:
|
123 |
|
|
'''A reference to a table column'''
|
124 |
|
|
def __init__(self, name, idx):
|
125 |
|
|
self.name = name
|
126 |
|
|
self.idx = idx
|
127 |
|
|
|
128 |
|
|
def __str__(self): return self.name
|
129 |
|
|
|
130 |
3114
|
aaronmk
|
# Controls when and how put_table() will partition the input table
|
131 |
3145
|
aaronmk
|
partition_size = 500000 # rows
|
132 |
3114
|
aaronmk
|
|
133 |
2177
|
aaronmk
|
input_col_prefix = '$'
|
134 |
|
|
|
135 |
2549
|
aaronmk
|
put_table_special_funcs = set(['_simplifyPath'])
|
136 |
|
|
|
137 |
2928
|
aaronmk
|
def put_table(db, node, in_table, in_row_ct_ref=None,
|
138 |
2806
|
aaronmk
|
row_ins_ct_ref=None, limit=None, start=0, on_error=exc.raise_,
|
139 |
3194
|
aaronmk
|
parent_ids_loc=None, next=None, top_call=True):
|
140 |
1996
|
aaronmk
|
'''
|
141 |
1998
|
aaronmk
|
@param node The XML tree that transforms the input to the output. Similar to
|
142 |
2177
|
aaronmk
|
put()'s node param, but with the input column name prefixed by
|
143 |
|
|
input_col_prefix in place of the column value.
|
144 |
3194
|
aaronmk
|
@return sql_gen.Col Where the pkeys (from INSERT RETURNING) are made
|
145 |
2067
|
aaronmk
|
available
|
146 |
1998
|
aaronmk
|
'''
|
147 |
2418
|
aaronmk
|
in_table = sql_gen.as_Table(in_table)
|
148 |
3113
|
aaronmk
|
in_table.set_srcs([in_table], overwrite=False)
|
149 |
3184
|
aaronmk
|
db.src = str(in_table)
|
150 |
2418
|
aaronmk
|
|
151 |
3123
|
aaronmk
|
def put_table_(node, in_row_ct_ref=None):
|
152 |
3107
|
aaronmk
|
return put_table(db, node, in_table, in_row_ct_ref, row_ins_ct_ref,
|
153 |
3194
|
aaronmk
|
None, 0, on_error, parent_ids_loc, next, False)
|
154 |
3107
|
aaronmk
|
|
155 |
3122
|
aaronmk
|
# Subset and/or partition in_table if needed
|
156 |
3194
|
aaronmk
|
if top_call:
|
157 |
3268
|
aaronmk
|
db.autoanalyze = True # but don't do this in row-based import
|
158 |
3269
|
aaronmk
|
db.autoexplain = True # but don't do this in row-based import
|
159 |
3268
|
aaronmk
|
|
160 |
3122
|
aaronmk
|
full_in_table = in_table
|
161 |
|
|
|
162 |
3194
|
aaronmk
|
total = 0
|
163 |
|
|
while limit == None or total < limit:
|
164 |
|
|
# Adjust partition size if last partition
|
165 |
|
|
this_limit = partition_size
|
166 |
|
|
if limit != None: this_limit = min(this_limit, limit - total)
|
167 |
3122
|
aaronmk
|
|
168 |
|
|
# Row # is interally 0-based, but 1-based to the user
|
169 |
3194
|
aaronmk
|
db.log_debug('********** Partition: rows '+str(start+1)+'-'
|
170 |
|
|
+str(start+this_limit)+' **********', level=1.2)
|
171 |
3122
|
aaronmk
|
|
172 |
|
|
# Subset in_table
|
173 |
|
|
in_table = copy.copy(full_in_table) # don't modify input!
|
174 |
3175
|
aaronmk
|
in_table.name = str(in_table) # prepend schema
|
175 |
3194
|
aaronmk
|
cur = sql.run_query_into(db, sql.mk_select(db, full_in_table,
|
176 |
3304
|
aaronmk
|
limit=this_limit, start=start), into=in_table, add_pkey_=True)
|
177 |
3122
|
aaronmk
|
# full_in_table will be shadowed (hidden) by created temp table
|
178 |
3194
|
aaronmk
|
|
179 |
|
|
this_ct = cur.rowcount
|
180 |
|
|
total += this_ct
|
181 |
|
|
start += this_ct # advance start to fetch next set
|
182 |
3195
|
aaronmk
|
if this_ct == 0: break # in_table size is multiple of partition_size
|
183 |
3194
|
aaronmk
|
|
184 |
3142
|
aaronmk
|
# Recurse
|
185 |
3122
|
aaronmk
|
pkeys_loc = put_table_(node, in_row_ct_ref)
|
186 |
3195
|
aaronmk
|
if in_row_ct_ref != None: in_row_ct_ref[0] += this_ct
|
187 |
3130
|
aaronmk
|
|
188 |
3142
|
aaronmk
|
sql.empty_temp(db, in_table)
|
189 |
|
|
|
190 |
3194
|
aaronmk
|
if this_ct < partition_size: break # partial partition = last
|
191 |
|
|
|
192 |
3130
|
aaronmk
|
# Work around PostgreSQL's temp table disk space leak
|
193 |
|
|
db.reconnect()
|
194 |
3121
|
aaronmk
|
|
195 |
3114
|
aaronmk
|
return pkeys_loc
|
196 |
|
|
|
197 |
2551
|
aaronmk
|
is_func = xml_func.is_func(node)
|
198 |
2005
|
aaronmk
|
out_table = name_of(node)
|
199 |
2432
|
aaronmk
|
|
200 |
|
|
# Divide children into fields and children with fkeys to parent
|
201 |
2005
|
aaronmk
|
row = {}
|
202 |
|
|
children = []
|
203 |
|
|
for child in xml_dom.NodeElemIter(node):
|
204 |
|
|
child_name = name_of(child)
|
205 |
|
|
if xml_dom.is_empty(child): row[child_name] = None
|
206 |
|
|
elif xml_dom.is_text(child):
|
207 |
|
|
row[child_name] = strings.to_unicode(xml_dom.value(child))
|
208 |
2113
|
aaronmk
|
else:
|
209 |
|
|
child_value = xml_dom.value_node(child)
|
210 |
2434
|
aaronmk
|
if is_func or is_ptr(child_name) or xml_func.is_func(child_value):
|
211 |
2432
|
aaronmk
|
row[child_name] = child_value
|
212 |
2113
|
aaronmk
|
else: children.append(child)
|
213 |
2434
|
aaronmk
|
|
214 |
|
|
# Special handling for structural XML functions
|
215 |
|
|
if out_table == '_simplifyPath':
|
216 |
2436
|
aaronmk
|
# Parse args
|
217 |
|
|
def wrap_e(e): raise xml_func.SyntaxError(e)
|
218 |
2434
|
aaronmk
|
try:
|
219 |
2506
|
aaronmk
|
next = row['next'] # modifies outer next var used by put_table_()
|
220 |
2434
|
aaronmk
|
require = row['require']
|
221 |
|
|
path = row['path']
|
222 |
2436
|
aaronmk
|
except KeyError, e: wrap_e(e)
|
223 |
|
|
try: next = xpath.parse(next)
|
224 |
|
|
except Parser.SyntaxError, e: wrap_e(e)
|
225 |
|
|
try: next = next[0].name
|
226 |
|
|
except IndexError, e: wrap_e(e)
|
227 |
2434
|
aaronmk
|
|
228 |
2506
|
aaronmk
|
return put_table_(path)
|
229 |
2434
|
aaronmk
|
|
230 |
2966
|
aaronmk
|
no_empty = set([in_table])
|
231 |
|
|
|
232 |
3107
|
aaronmk
|
def pkey(table): return sql.pkey(db, table, True)
|
233 |
|
|
|
234 |
2177
|
aaronmk
|
# Add fkey to parent
|
235 |
|
|
if parent_ids_loc != None:
|
236 |
2966
|
aaronmk
|
no_empty.add(parent_ids_loc.table)
|
237 |
2177
|
aaronmk
|
parent_ptr = node.getAttribute('fkey')
|
238 |
|
|
if parent_ptr == '': parent_ptr = pkey(name_of(node.parentNode))
|
239 |
|
|
row[parent_ptr] = parent_ids_loc
|
240 |
|
|
|
241 |
2060
|
aaronmk
|
# Divide fields into input columns and literal values
|
242 |
2506
|
aaronmk
|
parent_ids_loc = None # applies to this section
|
243 |
2313
|
aaronmk
|
in_tables = [in_table]
|
244 |
2060
|
aaronmk
|
for out_col, value in row.iteritems():
|
245 |
2432
|
aaronmk
|
# Handle forward pointers
|
246 |
|
|
if xml_dom.is_node(value): row[out_col] = value = put_table_(value)
|
247 |
|
|
|
248 |
|
|
# Translate values
|
249 |
2268
|
aaronmk
|
if isinstance(value, sql_gen.Col): # value is temp table column
|
250 |
|
|
in_tables.append(value.table)
|
251 |
2177
|
aaronmk
|
elif util.is_str(value) and value.startswith(input_col_prefix):
|
252 |
|
|
# value is input column
|
253 |
2272
|
aaronmk
|
row[out_col] = sql_gen.Col(strings.remove_prefix(input_col_prefix,
|
254 |
2313
|
aaronmk
|
value), in_table)
|
255 |
3428
|
aaronmk
|
else: # value is literal value
|
256 |
2323
|
aaronmk
|
row[out_col] = sql_gen.NamedCol(out_col, value)
|
257 |
2060
|
aaronmk
|
|
258 |
|
|
# Insert node
|
259 |
3077
|
aaronmk
|
pkeys_loc = sql_io.put_table(db, out_table, in_tables, row, row_ins_ct_ref,
|
260 |
2813
|
aaronmk
|
None, next, is_func, on_error)
|
261 |
2060
|
aaronmk
|
|
262 |
2975
|
aaronmk
|
sql.empty_temp(db, set(in_tables) - no_empty)
|
263 |
2966
|
aaronmk
|
|
264 |
2177
|
aaronmk
|
# Insert children with fkeys to parent
|
265 |
2506
|
aaronmk
|
parent_ids_loc = pkeys_loc # applies to this section
|
266 |
|
|
for child in children: put_table_(child)
|
267 |
2177
|
aaronmk
|
|
268 |
2133
|
aaronmk
|
return pkeys_loc
|