Differential D1059

Make the migration less memory consuming
ClosedPublic
Actions

Authored by jskladan on Nov 24 2016, 4:52 PM.

Details

Reviewers
• ralph2
tflink
mkrizek
Commits
rRSDB9eec2bb614fd: Make the migration less memory consuming
Summary

The psycopg does not use server-side cursor, and tries to just load all the
results of a query to the hosts memory.
This lessens the burden, although it still is quite memory-consuming because
we want to have the session which can rollback, and this basically stores all
the changed results locally, before applying the changes to DB.
This ended up using about 10 gigs of memory on dev (about 50k duplicate jobs,
about 3.5M results to be moved), which is still bad, but manageable enough
so I don't want to spend more time on it.

Test Plan

Tested in production! (dev actually...)

Diff Detail

Repository
rRSDB resultsdb
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
jskladan retitled this revision from to Make the migration less memory consuming.Nov 24 2016, 4:52 PM
jskladan updated this object.
jskladan edited the test plan for this revision. (Show Details)
jskladan added reviewers: mkrizek, tflink, • ralph2.
jskladan updated this revision to Diff 2696.Nov 24 2016, 5:18 PM
  • add yield_per to step 2 of schema upgrade
jskladan updated this revision to Diff 2697.Nov 24 2016, 7:26 PM
  • add yield_per to step 2 of schema upgrade
  • .
  • more logs
jskladan updated this revision to Diff 2698.Nov 24 2016, 9:00 PM
  • add yield_per to step 2 of schema upgrade
  • .
  • more logs
  • even moar logs
mkrizek accepted this revision.Nov 28 2016, 1:24 PM
This revision is now accepted and ready to land.Nov 28 2016, 1:24 PM
Closed by commit rRSDB9eec2bb614fd: Make the migration less memory consuming (authored by Josef Skladanka <jskladan@redhat.com>). · Explain WhyDec 6 2016, 2:03 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 2704

View Options

alembic/versions/540dbe71fa91_change_schema_to_v2_0_step_1.py

Show All 11 Lines
12branch_labels = None 12​branch_labels = None
13depends_on = None 13​depends_on = None
14 14​
15from alembic import op 15​from alembic import op
16import sqlalchemy as sa 16​import sqlalchemy as sa
17from sqlalchemy.ext.declarative import declarative_base 17​from sqlalchemy.ext.declarative import declarative_base
18from sqlalchemy.orm import relation, sessionmaker 18​from sqlalchemy.orm import relation, sessionmaker
19from sqlalchemy.sql import text 19​from sqlalchemy.sql import text
20​import logging
20 21​
21Session = sessionmaker() 22​Session = sessionmaker()
22Base = declarative_base() 23​Base = declarative_base()
23 24​
24 25​
25class Job(Base): 26​class Job(Base):
26 __tablename__ = 'job' 27​ __tablename__ = 'job'
27 28​
28 id = sa.Column(sa.Integer, primary_key=True) 29​ id = sa.Column(sa.Integer, primary_key=True)
29 uuid = sa.Column(sa.String(36), unique=True) 30​ uuid = sa.Column(sa.String(36), unique=True)
30 results = relation('Result', backref='job') 31​ results = relation('Result', backref='job')
31 32​
32 33​
33class Result(Base): 34​class Result(Base):
34 __tablename__ = 'result' 35​ __tablename__ = 'result'
35 36​
36 id = sa.Column(sa.Integer, primary_key=True) 37​ id = sa.Column(sa.Integer, primary_key=True)
37 job_id = sa.Column(sa.Integer, sa.ForeignKey('job.id')) 38​ job_id = sa.Column(sa.Integer, sa.ForeignKey('job.id'))
38 39​
39 40​
40def upgrade(): 41​def upgrade():
41 # Merge duplicate Jobs 42​ # Merge duplicate Jobs
43​ logger = logging.getLogger('alembic')
42 connection = op.get_bind() 44​ connection = op.get_bind()
43 session = Session(bind=connection) 45​ session = Session(bind=connection)
44 merge_targets = {} 46​ merge_targets = {}
45 jobs_to_delete = [] 47​ jobs_to_delete = []
48​
49​ job_count_query = connection.execute("select count(*) from job where uuid in (select uuid from job group by uuid having count(uuid) > 1);")
50​ job_count = -1
51​ for row in job_count_query:
52​ job_count = row[0]
53​
54​ logger.info("Jobs marked for inspection: %s", job_count)
55​
46 job_query = session.query(Job).from_statement(text( 56​ job_query = session.query(Job).from_statement(text(
47 "select id, uuid from job where uuid in (select uuid from job group by uuid having count(uuid) > 1) order by id;")) 57​ "select id, uuid from job where uuid in (select uuid from job group by uuid having count(uuid) > 1) order by id;")
58​ ).yield_per(100)
59​
60​ j = r = 0
48 for job in job_query: 61​ for job in job_query:
62​ j += 1
49 primary = merge_targets.setdefault(job.uuid, job) 63​ primary = merge_targets.setdefault(job.uuid, job)
50 if primary.id != job.id: 64​ if primary.id != job.id:
51 for result in job.results: 65​ for result in job.results:
66​ r += 1
52 result.job_id = primary.id 67​ result.job_id = primary.id
53 session.add(result) 68​ session.add(result)
54 jobs_to_delete.append(job) 69​ jobs_to_delete.append(job)
70​ if not j % 1000:
71​ logger.info("Jobs seen: %s out of %s", j, job_count)
72​ logger.info("Results marked for move: %s", r)
73​ session.commit()
55 session.commit() 74​ session.commit()
75​ logger.info("Removing duplicate jobs")
56 for job in jobs_to_delete: 76​ for job in jobs_to_delete:
57 session.delete(job) 77​ session.delete(job)
58 session.commit() 78​ session.commit()
59 79​
80​ logger.info("Changing table structure")
81​
60 # JOB 82​ # JOB
61 op.rename_table('job', 'group') 83​ op.rename_table('job', 'group')
62 op.alter_column('group', 'name', new_column_name='description') 84​ op.alter_column('group', 'name', new_column_name='description')
63 op.drop_column(u'group', 'status') 85​ op.drop_column(u'group', 'status')
64 op.drop_column(u'group', 'start_time') 86​ op.drop_column(u'group', 'start_time')
65 op.drop_column(u'group', 'end_time') 87​ op.drop_column(u'group', 'end_time')
66 op.create_unique_constraint(None, 'group', ['uuid']) 88​ op.create_unique_constraint(None, 'group', ['uuid'])
67 op.create_index( 89​ op.create_index(
▲ Show 20 Lines • Show All 69 Lines • Show Last 20 Lines
View Options

alembic/versions/dbfab576c81_change_schema_to_v2_0_step_2.py

Show All 11 Lines
12branch_labels = None 12​branch_labels = None
13depends_on = None 13​depends_on = None
14 14​
15from alembic import op 15​from alembic import op
16import sqlalchemy as db 16​import sqlalchemy as db
17from sqlalchemy.ext.declarative import declarative_base 17​from sqlalchemy.ext.declarative import declarative_base
18from sqlalchemy.orm import relationship, relation, sessionmaker 18​from sqlalchemy.orm import relationship, relation, sessionmaker
19import uuid 19​import uuid
20​import logging
20 21​
21Session = sessionmaker() 22​Session = sessionmaker()
22Base = declarative_base() 23​Base = declarative_base()
23 24​
24 25​
25db.relationship = relationship 26​db.relationship = relationship
26db.relation = relation 27​db.relation = relation
27 28​
Show All 31 Lines56 class Result(Base):
59 job_id = db.Column(db.Integer, db.ForeignKey('group.id')) 60​ job_id = db.Column(db.Integer, db.ForeignKey('group.id'))
60 testcase_id = db.Column(db.Integer, db.ForeignKey('testcase.id')) 61​ testcase_id = db.Column(db.Integer, db.ForeignKey('testcase.id'))
61 testcase_name = db.Column(db.Text) 62​ testcase_name = db.Column(db.Text)
62 63​
63 groups = db.relationship("Group", secondary='groups_to_results', backref="results") 64​ groups = db.relationship("Group", secondary='groups_to_results', backref="results")
64 job = db.relation('Group') # , lazy = False) 65​ job = db.relation('Group') # , lazy = False)
65 testcase = db.relation('Testcase', backref='results') # , lazy = False) 66​ testcase = db.relation('Testcase', backref='results') # , lazy = False)
66 67​
68​ logger = logging.getLogger('alembic')
67 connection = op.get_bind() 69​ connection = op.get_bind()
68 session = Session(bind=connection) 70​ session = Session(bind=connection)
69 for group in session.query(Group): 71​ i = 0
72​ for group in session.query(Group).yield_per(100):
73​ i += 1
70 if not group.uuid: 74​ if not group.uuid:
71 group.uuid = str(uuid.uuid1()) 75​ group.uuid = str(uuid.uuid1())
72 session.add(group) 76​ session.add(group)
77​ if not i % 1000:
78​ logger.info("Traversed %s groups", i)
73 session.commit() 79​ session.commit()
74 for result in session.query(Result): 80​ logger.info("Final group commit")
81​ session.commit()
82​ i = 0
83​ logger.info("Starting results")
84​ for result in session.query(Result).yield_per(100):
85​ i += 1
75 result.groups = [result.job] 86​ result.groups = [result.job]
76 result.testcase_name = result.testcase.name 87​ result.testcase_name = result.testcase.name
77 session.add(result) 88​ session.add(result)
78 session.commit() 89​ if not i % 1000:
90​ logger.info("Traversed %s results", i)
91​ session.commit()
92​ logger.info("Final result commit")
93​ session.commit()
94​ logger.info("Removing the columns")
79 op.drop_column('result', 'testcase_id') 95​ op.drop_column('result', 'testcase_id')
80 op.drop_column('result', 'job_id') 96​ op.drop_column('result', 'job_id')
81 97​
82 98​
83def downgrade(): 99​def downgrade():
84 class Result(Base): 100​ class Result(Base):
85 __tablename__ = 'result' 101​ __tablename__ = 'result'
86 102​
Show All 20 Lines