Distribute batch_status partitions evenly
@@ -84,14 +84,17 @@ def create_batch_tables(cluster, session):
time.sleep(60)
raise
+ keyspace_metadata = cluster.metadata.keyspaces[KEYSPACE]
+ if ('batch_status' in keyspace_metadata.tables and
+ 'txid_hash' not in keyspace_metadata.tables['batch_status'].columns):
+ session.execute('DROP TABLE batch_status')
+
logging.info('Trying to create batch_status')
create_table = """
CREATE TABLE IF NOT EXISTS batch_status (
- app text,
- transaction int,
+ txid_hash blob PRIMARY KEY,
applied boolean,
- op_id uuid,
- PRIMARY KEY ((app), transaction)
+ op_id uuid
)
"""
statement = SimpleStatement(create_table, retry_policy=NO_RETRIES)
@@ -104,19 +107,6 @@ def create_batch_tables(cluster, session):
time.sleep(60)
raise
- keyspace_metadata = cluster.metadata.keyspaces[KEYSPACE]
- if 'op_id' not in keyspace_metadata.tables['batch_status'].columns:
- alter_table = 'ALTER TABLE batch_status ADD op_id uuid'
- statement = SimpleStatement(alter_table, retry_policy=NO_RETRIES)
- try:
- session.execute(statement)
- except cassandra.OperationTimedOut:
- logging.warning(
- 'Encountered an operation timeout when altering batch_status. '
- 'Waiting 1 minute for schema to settle.')
- time.sleep(60)
- raise
-
def create_groups_table(session):
create_table = """
CREATE TABLE IF NOT EXISTS group_updates (
