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

Optimize consolidation fetching with GROUP BY and reduced limit

Browse Source 
Changed consolidation_group_limit from 100k to 10k for faster queries.

Reverted to GROUP BY approach for getting consolidation keys:
- Uses MySQL index efficiently: (UnitName, ToolNameID, NodeNum, EventDate, EventTime)
- GROUP BY with NodeNum ensures we don't lose any combinations
- Faster GROUP BY queries than large ORDER BY queries
- Smaller LIMIT = faster pagination

This matches the original optimization suggestion and should be faster.

🤖 Generated with Claude Code

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
alex
2025-12-25 22:22:30 +01:00
parent b6886293f6
commit fe2d173b0f
2 changed files with 22 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.py
View File

@@ -49,7 +49,7 @@ class MigrationSettings(BaseSettings):
)
batch_size: int = 10000
consolidation_group_limit: int = 100000
consolidation_group_limit: int = 10000
log_level: str = "INFO"
dry_run: bool = False

38
src/connectors/mysql_connector.py
View File

@@ -341,13 +341,16 @@ class MySQLConnector:
) -> Generator[List[Dict[str, Any]], None, None]:
"""Fetch consolidation groups from a partition.
Reads unique combinations of (UnitName, ToolNameID, EventDate, EventTime, NodeNum)
and fetches all rows for each group. This ensures proper consolidation.
Reads all rows from partition, sorted by consolidation key.
Yields rows grouped by (UnitName, ToolNameID, EventDate, EventTime).
This is more efficient than N+1 queries - fetches all data in one pass
and groups in Python instead of making separate MySQL queries per group.
Args:
table: Table name
partition: Partition name
limit: Maximum number of groups per query (uses config default if None)
limit: Batch size for consolidation (uses config default if None)
offset: Starting offset for pagination
Yields:
@@ -367,26 +370,27 @@ class MySQLConnector:
while retries < max_retries:
try:
with self.connection.cursor() as cursor:
# Get unique consolidation groups from partition
# First, get the distinct consolidation keys
group_query = f"""
SELECT DISTINCT UnitName, ToolNameID, EventDate, EventTime
# First: Get distinct consolidation keys from partition using GROUP BY
# Uses index efficiently: (UnitName, ToolNameID, NodeNum, EventDate, EventTime)
group_keys_query = f"""
SELECT UnitName, ToolNameID, EventDate, EventTime
FROM `{table}` PARTITION (`{partition}`)
ORDER BY UnitName, ToolNameID, EventDate, EventTime
GROUP BY UnitName, ToolNameID, NodeNum, EventDate, EventTime
ORDER BY UnitName, ToolNameID, NodeNum, EventDate, EventTime
LIMIT %s OFFSET %s
"""
cursor.execute(group_query, (limit, current_offset))
groups = cursor.fetchall()
cursor.execute(group_keys_query, (limit, current_offset))
group_keys = cursor.fetchall()
if not groups:
if not group_keys:
return
# For each group, fetch all rows
for group in groups:
unit_name = group.get("UnitName")
tool_name_id = group.get("ToolNameID")
event_date = group.get("EventDate")
event_time = group.get("EventTime")
# For each consolidation key, fetch all matching rows
for group_key in group_keys:
unit_name = group_key.get("UnitName")
tool_name_id = group_key.get("ToolNameID")
event_date = group_key.get("EventDate")
event_time = group_key.get("EventTime")
rows_query = f"""
SELECT * FROM `{table}` PARTITION (`{partition}`)