Remove set_options from stats-gathering query to improve its performance (#7)

* Remove set_options from stats-gathering query to improve its performance

* Upgrade Confluent lib so ARM64 builds will work

Author: woodlee

plan_monitor/collect.py

@@ -54,7 +54,6 @@ def poll_db(db_identifier: str, odbc_conn_string: str, stop_event: mp.Event,
                             "db_identifier": db_identifier,
                             "plan_handle": f'0x{row.plan_handle.hex()}',
                             "sql_handle": f'0x{row.sql_handle.hex()}',
-                            "set_options": row.set_options,
                             "creation_time": int(row.creation_time.replace(tzinfo=db_tz).timestamp() * 1000),
                             "last_execution_time": int(
                                 row.last_execution_time.replace(tzinfo=db_tz).timestamp() * 1000),

plan_monitor/detect.py

@@ -198,7 +198,6 @@ def detect() -> None:
                 db_identifier: str = msg_val['db_identifier']
                 sql_handle: str = msg_val['sql_handle']
                 stats_query_time: int = msg_val['stats_query_time']
-                set_options: int = msg_val['set_options']
                 msg_coordinates: str = common.msg_coordinates(msg)
 
                 caught_up: bool = time.time() - (stats_query_time / 1000) < config.MAX_ALLOWED_EVALUATION_LAG_SECONDS
@@ -220,7 +219,6 @@ def detect() -> None:
                         out_msg_val['prior_plans'] = [p.to_dict() for p in prior_plans]
                         out_msg_val['db_identifier'] = db_identifier
                         out_msg_val['sql_handle'] = sql_handle
-                        out_msg_val['set_options'] = set_options
                         out_msg_key = message_schemas.key_from_value(msg_val)
                         try:
                             kafka_producer.produce(topic=config.BAD_PLANS_TOPIC, key=out_msg_key, value=out_msg_val)

plan_monitor/message_schemas.py

@@ -17,10 +17,6 @@ def get_key_schema(record_name: str) -> str:
                 "name": "db_identifier",
                 "type": "string"
             },
-            {
-                "name": "set_options",
-                "type": "int"
-            },
             {
                 "name": "sql_handle",
                 "type": "string"
@@ -31,7 +27,6 @@ def get_key_schema(record_name: str) -> str:
 
 def key_from_value(message_value: Dict[str, Any]) -> Dict[str, Any]:
     return {"db_identifier": message_value["db_identifier"],
-            "set_options": message_value["set_options"],
             "sql_handle": message_value["sql_handle"]}
 
 
@@ -52,10 +47,6 @@ def key_from_value(message_value: Dict[str, Any]) -> Dict[str, Any]:
         "name": "sql_handle",
         "type": "string"
     },
-    {
-        "name": "set_options",
-        "type": "int"
-    },
     {
         "name": "creation_time",
         "type": "long",
@@ -117,7 +108,7 @@ def key_from_value(message_value: Dict[str, Any]) -> Dict[str, Any]:
 })
 
 PRIOR_PLANS_SCHEMA = [x for x in SINGLE_PLAN_STATS_FIELDS
-                      if x['name'] not in ('db_identifier', 'sql_handle', 'set_options')]
+                      if x['name'] not in ('db_identifier', 'sql_handle')]
 
 BAD_PLANS_MESSAGE_VALUE_AVRO_SCHEMA = json.dumps({
     "name": "bad_plans_value",

plan_monitor/queries.py

@@ -22,7 +22,6 @@
     SELECT
         qs.plan_handle AS plan_handle
         , qs.sql_handle AS sql_handle
-        , CAST(epa.value AS INT) AS set_options
         , MIN(qs.creation_time) AS creation_time
         , MAX(qs.last_execution_time) AS last_execution_time
         , MAX(qs.execution_count) AS execution_count
@@ -37,9 +36,7 @@
         , GETDATE() AS stats_query_time
     FROM recent_plans
     JOIN sys.dm_exec_query_stats AS qs ON (recent_plans.plan_handle = qs.plan_handle)
-    CROSS APPLY sys.dm_exec_plan_attributes(qs.plan_handle) AS epa
-    WHERE epa.attribute = 'set_options'
-    GROUP BY qs.plan_handle, qs.sql_handle, epa.value
+    GROUP BY qs.plan_handle, qs.sql_handle
 )
 SELECT * FROM agged
 WHERE (total_logical_writes = 0 OR statement_count > 1)
@@ -51,7 +48,6 @@
 class StatsDmvsQueryResult(NamedTuple):
     plan_handle: bytes
     sql_handle: bytes
-    set_options: int
     creation_time: datetime
     last_execution_time: datetime
     execution_count: int

plan_monitor/stats_store.py

@@ -56,7 +56,7 @@ def __len__(self) -> int:
 
     def register_stats_from_message(self, msg: Dict[str, Any],
                                     msg_coordinates: str) -> List[PlanStats]:
-        query_key = (msg["db_identifier"], msg["set_options"], msg["sql_handle"])
+        query_key = (msg["db_identifier"], msg["sql_handle"])
         current_stats = PlanStats(
             msg["plan_handle"], msg['creation_time'], msg['last_execution_time'], msg['stats_query_time'],
             msg['statement_count'], msg['execution_count'], msg['total_elapsed_time'], msg['total_logical_reads'],

requirements.txt

@@ -1,4 +1,4 @@
-confluent-kafka[avro]==1.8.2
+confluent-kafka[avro]==2.6.0
 markupsafe==2.0.1
 pyodbc==4.0.32
 python-dateutil==2.8.2