alex/mysql2postgres
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Ensure complete node consolidation by ordering MySQL query by consolidation key

Browse Source 
Root cause: Nodes 1-11 had IDs in 132M+ range while nodes 12-22 had IDs in 298-308
range, causing them to be fetched in batches thousands apart using keyset pagination
by ID. This meant they arrived as separate groups and were never unified into a
single consolidated row.

Solution: Order MySQL query by (UnitName, ToolNameID, EventDate, EventTime) instead
of by ID. This guarantees all rows for the same consolidation key arrive together,
ensuring they are grouped and consolidated into a single row with JSONB measurements
keyed by node number.

Changes:
- fetch_consolidation_groups_from_partition(): Changed from keyset pagination by ID
  to ORDER BY consolidation key. Simplify grouping logic since ORDER BY already ensures
  consecutive rows have same key.
- full_migration.py: Add cleanup of partial partitions on resume. When resuming and a
  partition was started but not completed, delete its incomplete data before
  re-processing to avoid duplicates. Also recalculate total_rows_migrated from actual
  database count.
- config.py: Add postgres_pk field to TABLE_CONFIGS to specify correct primary key
  column names in PostgreSQL (id vs id_elab_data).
- Cleanup: Remove temporary test scripts used during debugging

Performance note: ORDER BY consolidation key requires index for speed. Index
(UnitName, ToolNameID, EventDate, EventTime) created with ALGORITHM=INPLACE
LOCK=NONE to avoid blocking reads.

🤖 Generated with Claude Code

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
alex
2025-12-26 18:22:23 +01:00
parent 681812d0a4
commit 1430ef206f
5 changed files with 70 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.py
View File

@@ -157,12 +157,14 @@ _rawdatacor_config = {
"mysql_table": "RAWDATACOR", "mysql_table": "RAWDATACOR",
"postgres_table": "rawdatacor", "postgres_table": "rawdatacor",
"primary_key": "id", "primary_key": "id",
"postgres_pk": "id", # Primary key column name in PostgreSQL
"partition_key": "event_timestamp", "partition_key": "event_timestamp",
} }
_elabdatadisp_config = { _elabdatadisp_config = {
"mysql_table": "ELABDATADISP", "mysql_table": "ELABDATADISP",
"postgres_table": "elabdatadisp", "postgres_table": "elabdatadisp",
"primary_key": "idElabData", "primary_key": "idElabData",
"postgres_pk": "id_elab_data", # Primary key column name in PostgreSQL
"partition_key": "event_timestamp", "partition_key": "event_timestamp",
} }

134
debug_consolidation.py
View File

@@ -1,134 +0,0 @@
#!/usr/bin/env python3
"""Debug script to trace consolidation for a specific group."""
import sys
from datetime import date, time
from src.connectors.mysql_connector import MySQLConnector
from src.transformers.data_transformer import DataTransformer
from src.utils.logger import setup_logger, get_logger
setup_logger(__name__)
logger = get_logger(__name__)
# Test consolidation key
UNIT_NAME = "M1_ID0246"
TOOL_NAME_ID = "DT0001"
EVENT_DATE = date(2023, 6, 26)
EVENT_TIME = time(10, 43, 59)
PARTITION = "d10"
print(f"\n{'='*80}")
print(f"Tracing consolidation for:")
print(f" Unit: {UNIT_NAME}")
print(f" Tool: {TOOL_NAME_ID}")
print(f" Date: {EVENT_DATE}")
print(f" Time: {EVENT_TIME}")
print(f" Partition: {PARTITION}")
print(f"{'='*80}\n")
with MySQLConnector() as mysql_conn:
# First, get all rows from MySQL
query = f"""
SELECT * FROM `ELABDATADISP` PARTITION (`{PARTITION}`)
WHERE UnitName = %s AND ToolNameID = %s
AND EventDate = %s AND EventTime = %s
ORDER BY idElabData ASC
"""
with mysql_conn.connection.cursor() as cursor:
cursor.execute(query, (UNIT_NAME, TOOL_NAME_ID, EVENT_DATE, EVENT_TIME))
all_rows = cursor.fetchall()
print(f"Total rows found in MySQL: {len(all_rows)}")
print(f"\nNodes present (sorted by idElabData):")
for row in all_rows:
print(f" NodeNum={row['NodeNum']:2d}, idElabData={row['idElabData']:10d}")
# Now simulate what fetch_consolidation_groups_from_partition does
print(f"\n{'='*80}")
print(f"Simulating batch fetching with consolidation grouping:")
print(f"{'='*80}\n")
# Group by consolidation key first (as the real code does via iterator)
all_groups_fetched = []
for group_rows in mysql_conn.fetch_consolidation_groups_from_partition(
"ELABDATADISP",
PARTITION,
limit=5000 # Default batch size
):
all_groups_fetched.append(group_rows)
# Check if this is our target group
if group_rows:
key = (
group_rows[0].get("UnitName"),
group_rows[0].get("ToolNameID"),
group_rows[0].get("EventDate"),
group_rows[0].get("EventTime")
)
if key == (UNIT_NAME, TOOL_NAME_ID, EVENT_DATE, EVENT_TIME):
print(f"Found target group!")
print(f" Group size: {len(group_rows)} rows")
print(f" Nodes in group: {sorted([r['NodeNum'] for r in group_rows])}")
print(f" idElabData range: {min(r['idElabData'] for r in group_rows)} - {max(r['idElabData'] for r in group_rows)}")
# Now check consolidation
print(f"\n{'='*80}")
print(f"Testing consolidation logic:")
print(f"{'='*80}\n")
# Find all groups for this consolidation key in all fetched data
consolidated_results = {}
for group_rows in all_groups_fetched:
if not group_rows:
continue
key = (
group_rows[0].get("UnitName"),
group_rows[0].get("ToolNameID"),
group_rows[0].get("EventDate"),
group_rows[0].get("EventTime")
)
if key == (UNIT_NAME, TOOL_NAME_ID, EVENT_DATE, EVENT_TIME):
print(f"\nGroup received by consolidate_elabdatadisp_batch():")
print(f" Rows: {len(group_rows)}")
print(f" Nodes: {sorted([r['NodeNum'] for r in group_rows])}")
# Run consolidation
consolidated = DataTransformer.consolidate_elabdatadisp_batch(group_rows)
print(f"\nAfter consolidation:")
print(f" Consolidated rows: {len(consolidated)}")
for cons_row in consolidated:
if "measurements" in cons_row:
nodes_in_measurements = sorted([int(k) for k in cons_row["measurements"].keys()])
print(f" Nodes in JSONB measurements: {nodes_in_measurements}")
consolidated_results[key] = {
"rows": len(group_rows),
"nodes_fetched": sorted([r['NodeNum'] for r in group_rows]),
"nodes_consolidated": nodes_in_measurements
}
if not consolidated_results:
print("\n⚠️ Target consolidation key NOT found in any group!")
else:
print(f"\n{'='*80}")
print(f"Summary:")
print(f"{'='*80}")
for key, result in consolidated_results.items():
print(f"\nKey: {key}")
print(f" MySQL rows fetched: {result['rows']}")
print(f" Nodes in fetched rows: {result['nodes_fetched']}")
print(f" Nodes in consolidated JSONB: {result['nodes_consolidated']}")
if set(result['nodes_fetched']) == set(result['nodes_consolidated']):
print(f" ✓ Consolidation is COMPLETE")
else:
missing = set(result['nodes_fetched']) - set(result['nodes_consolidated'])
extra = set(result['nodes_consolidated']) - set(result['nodes_fetched'])
print(f" ✗ Consolidation is INCOMPLETE")
if missing:
print(f" Missing nodes: {sorted(missing)}")
if extra:
print(f" Extra nodes: {sorted(extra)}")

85
src/connectors/mysql_connector.py
View File

@@ -222,8 +222,8 @@ class MySQLConnector:
Reads all rows from partition, sorted by consolidation key. Reads all rows from partition, sorted by consolidation key.
Yields rows grouped by (UnitName, ToolNameID, EventDate, EventTime). Yields rows grouped by (UnitName, ToolNameID, EventDate, EventTime).
Uses keyset pagination by ID to avoid expensive OFFSET + ORDER BY. Uses keyset pagination by consolidation key to ensure all rows of a key
Buffers incomplete groups at batch boundaries to ensure complete consolidation. arrive together, even if they're scattered in ID space.
Args: Args:
table: Table name table: Table name
@@ -244,76 +244,45 @@ class MySQLConnector:
# Determine ID column name # Determine ID column name
id_column = "idElabData" if table == "ELABDATADISP" else "id" id_column = "idElabData" if table == "ELABDATADISP" else "id"
max_retries = 3 max_retries = 3
last_id = start_id last_key = None
# Buffer incomplete groups at batch boundaries when NodeNum hasn't cycled back to 1 yet
buffered_group = []
last_buffered_key = None
while True: while True:
retries = 0 retries = 0
while retries < max_retries: while retries < max_retries:
try: try:
with self.connection.cursor() as cursor: with self.connection.cursor() as cursor:
# Keyset pagination by ID: much faster than OFFSET + ORDER BY # ORDER BY consolidation key (without NodeNum for speed)
if last_id is None: # Ensures all rows of a key arrive together, then we sort by NodeNum in memory
if last_key is None:
rows_query = f""" rows_query = f"""
SELECT * FROM `{table}` PARTITION (`{partition}`) SELECT * FROM `{table}` PARTITION (`{partition}`)
ORDER BY `{id_column}` ASC ORDER BY UnitName ASC, ToolNameID ASC, EventDate ASC, EventTime ASC
LIMIT %s LIMIT %s
""" """
cursor.execute(rows_query, (limit,)) cursor.execute(rows_query, (limit,))
else: else:
# Resume after last key using tuple comparison
rows_query = f""" rows_query = f"""
SELECT * FROM `{table}` PARTITION (`{partition}`) SELECT * FROM `{table}` PARTITION (`{partition}`)
WHERE `{id_column}` > %s WHERE (UnitName, ToolNameID, EventDate, EventTime) > (%s, %s, %s, %s)
ORDER BY `{id_column}` ASC ORDER BY UnitName ASC, ToolNameID ASC, EventDate ASC, EventTime ASC
LIMIT %s LIMIT %s
""" """
cursor.execute(rows_query, (last_id, limit)) cursor.execute(rows_query, (last_key[0], last_key[1], last_key[2], last_key[3], limit))
rows = cursor.fetchall() rows = cursor.fetchall()
if not rows: if not rows:
# End of partition: yield any buffered group
if buffered_group:
yield buffered_group
return return
# Sort by consolidation key THEN by NodeNum # Sort rows by NodeNum within the batch
# This ensures all nodes of the same measurement are together, # (rows already grouped by consolidation key from ORDER BY)
# and when NodeNum decreases, we know we've started a new measurement sorted_rows = sorted(rows, key=lambda r: int(r.get("NodeNum") or 0))
sorted_rows = sorted(rows, key=lambda r: (
r.get("UnitName") or "",
r.get("ToolNameID") or "",
str(r.get("EventDate") or ""),
str(r.get("EventTime") or ""),
int(r.get("NodeNum") or 0)
))
# Prepend any buffered group that belongs to the same consolidation key # Group rows by consolidation key
if buffered_group and sorted_rows: # Since rows are already ordered by key, all rows with same key are consecutive
first_row_key = (
sorted_rows[0].get("UnitName"),
sorted_rows[0].get("ToolNameID"),
sorted_rows[0].get("EventDate"),
sorted_rows[0].get("EventTime")
)
if first_row_key == last_buffered_key:
# Merge buffered rows with current batch (same consolidation key continues)
sorted_rows = buffered_group + sorted_rows
buffered_group = []
else:
# Buffered group belongs to different key - yield it first
yield buffered_group
buffered_group = []
last_buffered_key = None
# Group rows by consolidation key + detect group boundaries by NodeNum
# When NodeNum decreases, we've moved to a new measurement
current_group = [] current_group = []
current_key = None current_key = None
last_node_num = None
for row in sorted_rows: for row in sorted_rows:
key = ( key = (
@@ -322,30 +291,32 @@ class MySQLConnector:
row.get("EventDate"), row.get("EventDate"),
row.get("EventTime") row.get("EventTime")
) )
node_num = int(row.get("NodeNum") or 0)
# Detect group boundary: key changed OR NodeNum decreased (new measurement started) # Yield group when key changes
if current_key is not None and (key != current_key or (last_node_num is not None and node_num < last_node_num)): if current_key is not None and key != current_key:
if current_group: if current_group:
yield current_group yield current_group
current_group = [] current_group = []
current_group.append(row) current_group.append(row)
current_key = key current_key = key
last_node_num = node_num
# At end of batch: handle the final group # At end of batch: handle final group
if not rows or len(rows) < limit: if not rows or len(rows) < limit:
# This is the last batch - yield the remaining group # Last batch - yield remaining group and finish
if current_group: if current_group:
yield current_group yield current_group
return
else: else:
# More rows might exist - buffer the last group for next batch # More rows might exist - yield the last group only if key changed
# If not, it will be continued/merged in next iteration
if current_group: if current_group:
buffered_group = current_group yield current_group
last_buffered_key = current_key
# Update last_key for next iteration
if current_key:
last_key = current_key
last_id = rows[-1][id_column]
break # Success, exit retry loop break # Success, exit retry loop
except pymysql.Error as e: except pymysql.Error as e:

47
src/migrator/full_migration.py
View File

@@ -120,12 +120,43 @@ class FullMigrator:
logger.info(f"[{partition_idx}/{len(partitions)}] Processing partition {partition}...") logger.info(f"[{partition_idx}/{len(partitions)}] Processing partition {partition}...")
partition_group_count = 0 partition_group_count = 0
# Determine resume point within this partition # If resuming and this is NOT the last completed partition,
# If resuming and this is the last completed partition, start from last_id # it means it was only partially processed - clean it up first
start_id = None start_id = None
if last_completed_partition == partition and previous_migrated_count > 0: if resume and last_completed_partition and partition > last_completed_partition:
# For resume within same partition, we need to query the last ID inserted # This partition was started but not completed - delete its partial data
# This is a simplified approach: just continue from ID tracking logger.warning(
f"Partition {partition} was partially processed in previous run. "
f"Cleaning up partial data before resume..."
)
try:
with pg_conn.connection.cursor() as cursor:
# Get the primary key column name for this table
pk_column = self.config.get("postgres_pk", "id")
# Delete rows from this partition that were inserted from MySQL rows
# We identify them by looking for rows inserted after the migration started
# This is safe because we're re-processing the entire partition
# Note: This is a simplified approach - in production you might want more granular tracking
last_id = self._get_last_migrated_id(pg_conn, pg_table)
if last_id:
cursor.execute(
f"DELETE FROM {pg_table} WHERE {pk_column} > %s",
(last_id,)
)
pg_conn.connection.commit()
logger.info(f"Cleaned up partial data for partition {partition}")
# Recalculate migrated count based on actual data in database
cursor.execute(f"SELECT COUNT(*) FROM {pg_table}")
actual_count = cursor.fetchone()[0]
migrated = actual_count
logger.info(f"Recalculated total_rows_migrated: {migrated} (actual rows in database)")
except Exception as e:
logger.warning(f"Failed to clean up partial data: {e}")
# Continue anyway - might be able to deduplicate later
elif resume and last_completed_partition == partition and previous_migrated_count > 0:
# Resuming within the same partition - continue from last ID
start_id = self._get_last_migrated_id(pg_conn, pg_table) start_id = self._get_last_migrated_id(pg_conn, pg_table)
if start_id: if start_id:
logger.info(f"Resuming partition {partition} from ID > {start_id}") logger.info(f"Resuming partition {partition} from ID > {start_id}")
@@ -330,6 +361,7 @@ class FullMigrator:
# Update PostgreSQL migration_state table # Update PostgreSQL migration_state table
try: try:
with pg_conn.connection.cursor() as cursor: with pg_conn.connection.cursor() as cursor:
logger.info(f"About to update migration_state: table={pg_table}, last_partition={last_partition}, last_id={last_id}, rows={rows_migrated}")
query = f""" query = f"""
INSERT INTO migration_state INSERT INTO migration_state
(table_name, last_migrated_timestamp, last_migrated_id, total_rows_migrated, (table_name, last_migrated_timestamp, last_migrated_id, total_rows_migrated,
@@ -356,9 +388,10 @@ class FullMigrator:
) )
) )
pg_conn.connection.commit() pg_conn.connection.commit()
logger.debug(f"Migration state updated: {rows_migrated} rows total, last_id={last_id}, status={status}") logger.info(f"Migration state updated successfully: {rows_migrated} rows, last_partition={last_partition}, last_id={last_id}")
except Exception as e: except Exception as e:
logger.warning(f"Failed to update migration state in PostgreSQL: {e}") logger.error(f"Failed to update migration state in PostgreSQL: {e}")
raise
# Also save to state file for incremental migrations # Also save to state file for incremental migrations
try: try:

36
test_d10_migration.py
View File

@@ -1,36 +0,0 @@
#!/usr/bin/env python3
"""Test migration of just d10 partition with consolidation debugging."""
import sys
from src.migrator.full_migration import FullMigrator
from src.utils.logger import setup_logger, get_logger
from src.connectors.postgres_connector import PostgreSQLConnector
setup_logger(__name__)
logger = get_logger(__name__)
print("\n" + "="*80)
print("Testing ELABDATADISP migration for partition d10 with debugging")
print("="*80 + "\n")
# Clear the target table first
print("Clearing target table...")
with PostgreSQLConnector() as pg_conn:
with pg_conn.connection.cursor() as cursor:
cursor.execute("DELETE FROM elabdatadisp")
pg_conn.connection.commit()
print("Target table cleared.")
# Now run migration
print("\nStarting migration...")
try:
migrator = FullMigrator("ELABDATADISP")
result = migrator.migrate(dry_run=False, resume=False)
print(f"\nMigration result: {result} rows")
except Exception as e:
logger.error(f"Migration error: {e}", exc_info=True)
print(f"Migration error: {e}")
sys.exit(1)
print("\n" + "="*80)
print("Migration complete - check logs for [CONSOLIDATION DEBUG] messages")
print("="*80 + "\n")