254 # Drop temporary relation table and record from cwmassive_initialized |
254 # Drop temporary relation table and record from cwmassive_initialized |
255 self.sql('DROP TABLE %(tmp_table)s' % {'tmp_table': tmp_tablename}) |
255 self.sql('DROP TABLE %(tmp_table)s' % {'tmp_table': tmp_tablename}) |
256 self.sql('DELETE FROM cwmassive_initialized ' |
256 self.sql('DELETE FROM cwmassive_initialized ' |
257 'WHERE retype = %(rtype)s AND uuid = %(uuid)s', |
257 'WHERE retype = %(rtype)s AND uuid = %(uuid)s', |
258 {'rtype': retype, 'uuid': uuid}) |
258 {'rtype': retype, 'uuid': uuid}) |
|
259 # restore all deleted indexes and constraints |
259 self._dbh.restore_indexes_and_constraints() |
260 self._dbh.restore_indexes_and_constraints() |
260 # Delete the meta data table |
261 # delete the meta data table |
261 self.sql('DROP TABLE IF EXISTS cwmassive_initialized') |
262 self.sql('DROP TABLE IF EXISTS cwmassive_initialized') |
262 self.commit() |
263 self.commit() |
263 |
264 |
264 # FLUSH ################################################################# |
265 # FLUSH ################################################################# |
265 |
266 |
273 self._cnx.rollback() |
274 self._cnx.rollback() |
274 else: |
275 else: |
275 raise exc |
276 raise exc |
276 |
277 |
277 def flush_relations(self): |
278 def flush_relations(self): |
278 """Flush the relations data.""" |
279 """Flush the relations data from in-memory structures to a temporary table.""" |
279 for rtype, data in self._data_relations.items(): |
280 for rtype, data in self._data_relations.items(): |
280 if not data: |
281 if not data: |
281 # There is no data for these etype for this flush round. |
282 # There is no data for these etype for this flush round. |
282 continue |
283 continue |
283 buf = pgstore._create_copyfrom_buffer(data, ('eid_from', 'eid_to')) |
284 buf = pgstore._create_copyfrom_buffer(data, ('eid_from', 'eid_to')) |
284 if not buf: |
285 if not buf: |
285 # The buffer is empty. This is probably due to error in _create_copyfrom_buffer |
286 # The buffer is empty. This is probably due to error in _create_copyfrom_buffer |
286 raise ValueError |
287 raise ValueError |
287 cursor = self._cnx.cnxset.cu |
288 cursor = self._cnx.cnxset.cu |
288 # Push into the tmp table |
|
289 tablename = '%s_relation' % rtype.lower() |
289 tablename = '%s_relation' % rtype.lower() |
290 tmp_tablename = '%s_%s' % (tablename, self.uuid) |
290 tmp_tablename = '%s_%s' % (tablename, self.uuid) |
291 cursor.copy_from(buf, tmp_tablename, null='NULL', columns=('eid_from', 'eid_to')) |
291 cursor.copy_from(buf, tmp_tablename, null='NULL', columns=('eid_from', 'eid_to')) |
292 # Clear data cache |
292 # Clear data cache |
293 self._data_relations[rtype] = [] |
293 self._data_relations[rtype] = [] |
294 |
294 |
295 def flush_entities(self): |
295 def flush_entities(self): |
296 """Flush the entities data.""" |
296 """Flush the entities data from in-memory structures to a temporary table.""" |
297 metagen = self.metagen |
297 metagen = self.metagen |
298 for etype, data in self._data_entities.items(): |
298 for etype, data in self._data_entities.items(): |
299 if not data: |
299 if not data: |
300 # There is no data for these etype for this flush round. |
300 # There is no data for these etype for this flush round. |
301 continue |
301 continue |