import gc import pandas as pd from concurrent.futures import ThreadPoolExecutor from typing import List, Optional import sys # Fix import conflict: project has a local 'deltalake' directory that shadows the package # Remove it from sys.modules if it's the local empty module if 'deltalake' in sys.modules: del sys.modules['deltalake'] try: from deltalake import DeltaTable, write_deltalake except ImportError: # If deltalake package is not installed, create dummy classes DeltaTable = None write_deltalake = None from o11ydiagnosis.subtasks.slowlogconverter.slowlog import SlowLog from o11ydiagnosis.subtasks.slowlogconverter.log import log as logger from o11ydiagnosis.subtasks.slowlogconverter.parse_query import parse_binary_plan from datetime import datetime, timezone, date, timedelta import pyarrow as pa import time import os slowlogs_schema = pa.schema([ pa.field("id", pa.int64(), nullable=False), pa.field("time", pa.float64(), nullable=False), pa.field("digest", pa.string(), nullable=False), pa.field("date", pa.string(), nullable=False), # partition column pa.field("instance", pa.string(), nullable=True), pa.field("txn_start_ts", pa.string(), nullable=True), pa.field("user", pa.string(), nullable=True), pa.field("host", pa.string(), nullable=True), pa.field("conn_id", pa.string(), nullable=True), pa.field("session_alias", pa.string(), nullable=True), pa.field("exec_retry_count", pa.string(), nullable=True), pa.field("exec_retry_time", pa.float64(), nullable=True), pa.field("query_time", pa.float64(), nullable=True), pa.field("parse_time", pa.float64(), nullable=True), pa.field("compile_time", pa.float64(), nullable=True), pa.field("rewrite_time", pa.float64(), nullable=True), pa.field("preproc_subqueries", pa.string(), nullable=True), pa.field("preproc_subqueries_time", pa.float64(), nullable=True), pa.field("optimize_time", pa.float64(), nullable=True), pa.field("wait_ts", pa.float64(), nullable=True), pa.field("prewrite_time", pa.float64(), nullable=True), pa.field("wait_prewrite_binlog_time", pa.float64(), nullable=True), pa.field("commit_time", pa.float64(), nullable=True), pa.field("get_commit_ts_time", pa.float64(), nullable=True), pa.field("commit_backoff_time", pa.float64(), nullable=True), pa.field("backoff_types", pa.string(), nullable=True), pa.field("resolve_lock_time", pa.float64(), nullable=True), pa.field("local_latch_wait_time", pa.float64(), nullable=True), pa.field("write_keys", pa.int64(), nullable=True), pa.field("write_size", pa.int64(), nullable=True), pa.field("prewrite_region", pa.int64(), nullable=True), pa.field("txn_retry", pa.int64(), nullable=True), pa.field("cop_time", pa.float64(), nullable=True), pa.field("process_time", pa.float64(), nullable=True), pa.field("wait_time", pa.float64(), nullable=True), pa.field("backoff_time", pa.float64(), nullable=True), pa.field("lockkeys_time", pa.float64(), nullable=True), pa.field("request_count", pa.string(), nullable=True), pa.field("total_keys", pa.string(), nullable=True), pa.field("process_keys", pa.string(), nullable=True), pa.field("rocksdb_delete_skipped_count", pa.string(), nullable=True), pa.field("rocksdb_key_skipped_count", pa.string(), nullable=True), pa.field("rocksdb_block_cache_hit_count", pa.string(), nullable=True), pa.field("rocksdb_block_read_count", pa.string(), nullable=True), pa.field("rocksdb_block_read_byte", pa.string(), nullable=True), pa.field("db", pa.string(), nullable=True), pa.field("index_names", pa.string(), nullable=True), pa.field("is_internal", pa.bool_(), nullable=True), pa.field("stats", pa.string(), nullable=True), pa.field("cop_proc_avg", pa.float64(), nullable=True), pa.field("cop_proc_p90", pa.float64(), nullable=True), pa.field("cop_proc_max", pa.float64(), nullable=True), pa.field("cop_proc_addr", pa.string(), nullable=True), pa.field("cop_wait_avg", pa.float64(), nullable=True), pa.field("cop_wait_p90", pa.float64(), nullable=True), pa.field("cop_wait_max", pa.float64(), nullable=True), pa.field("cop_wait_addr", pa.string(), nullable=True), pa.field("mem_max", pa.int64(), nullable=True), pa.field("disk_max", pa.int64(), nullable=True), pa.field("kv_total", pa.float64(), nullable=True), pa.field("pd_total", pa.float64(), nullable=True), pa.field("backoff_total", pa.float64(), nullable=True), pa.field("write_sql_response_total", pa.float64(), nullable=True), pa.field("result_rows", pa.int64(), nullable=True), pa.field("warnings", pa.string(), nullable=True), pa.field("backoff_detail", pa.string(), nullable=True), pa.field("prepared", pa.bool_(), nullable=True), pa.field("succ", pa.bool_(), nullable=True), pa.field("is_explicit_txn", pa.bool_(), nullable=True), pa.field("is_write_cache_table", pa.bool_(), nullable=True), pa.field("plan_from_cache", pa.bool_(), nullable=True), pa.field("plan_from_binding", pa.bool_(), nullable=True), pa.field("has_more_results", pa.bool_(), nullable=True), pa.field("resource_group", pa.string(), nullable=True), pa.field("request_unit_read", pa.float64(), nullable=True), pa.field("request_unit_v2", pa.float64(), nullable=True), pa.field("read_pool_task_details", pa.string(), nullable=True), pa.field("request_unit_write", pa.float64(), nullable=True), pa.field("time_queued_by_rc", pa.float64(), nullable=True), pa.field("plan_digest", pa.string(), nullable=True), pa.field("prev_stmt", pa.string(), nullable=True), pa.field("serverless_tenant_id", pa.string(), nullable=True), pa.field("serverless_project_id", pa.string(), nullable=True), pa.field("serverless_cluster_id", pa.string(), nullable=True), pa.field("worker_role", pa.string(), nullable=True), pa.field("is_tiflash", pa.bool_(), nullable=True), pa.field("query", pa.string(), nullable=True), pa.field("binary_plan", pa.string(), nullable=True), pa.field("plan", pa.string(), nullable=True), pa.field("decoded_plan", pa.string(), nullable=True), ]) plans_schema = pa.schema([ pa.field("plan_digest", pa.string(), nullable=False), pa.field("digest", pa.string(), nullable=True), pa.field("binary_plan", pa.string(), nullable=True), pa.field("plan", pa.string(), nullable=True), pa.field("decoded_plan", pa.string(), nullable=True), ]) queries_schema = pa.schema([ pa.field("digest", pa.string(), nullable=False), pa.field("query", pa.string(), nullable=True), ]) def write_slowlogs_to_deltalake(added_logs: List[SlowLog], region: str, cluster: str, base_path: str = "./deltalake", optimize: bool = True, partition_date: str = None, merge_write: bool = True): slowlogs_path = f"{base_path}/slowlogs" slowlogs_data = [] # List of uint64 fields that need to be converted to string uint64_fields_to_convert = [ "txn_start_ts", "conn_id", "exec_retry_count", "preproc_subqueries", "request_count", "total_keys", "process_keys", "rocksdb_delete_skipped_count", "rocksdb_key_skipped_count", "rocksdb_block_cache_hit_count", "rocksdb_block_read_count", "rocksdb_block_read_byte" ] # If no partition_date is provided, use the date from the first log entry if partition_date is None and added_logs: dt_object_utc = datetime.fromtimestamp(added_logs[0].time, tz=timezone.utc) partition_date = dt_object_utc.strftime("%Y%m%d") for log in added_logs: slowlogs_dict = log.__dict__.copy() slowlogs_dict["date"] = partition_date # Convert specific uint64 fields to string for field in uint64_fields_to_convert: if field in slowlogs_dict and slowlogs_dict[field] is not None: slowlogs_dict[field] = str(slowlogs_dict[field]) slowlogs_data.append(slowlogs_dict) slowlogs_data_start = time.time() if slowlogs_data: slowlogs_df = pd.DataFrame(slowlogs_data) if 'binary_plan' in slowlogs_df.columns: slowlogs_df['decoded_plan'] = slowlogs_df['binary_plan'].apply( lambda x: parse_binary_plan(x) if x is not None and x != "" else None ) else: slowlogs_df['decoded_plan'] = None # Ensure all schema columns exist in DataFrame, fill missing columns with None schema_column_names = [field.name for field in slowlogs_schema] for col in schema_column_names: if col not in slowlogs_df.columns: slowlogs_df[col] = None # Ensure id field is not null (required by schema) if 'id' in slowlogs_df.columns: # Fill null id values with 0 or remove rows with null id null_id_mask = slowlogs_df['id'].isna() if null_id_mask.any(): # Use a default id value for null ids (e.g., 0 or generate from timestamp) slowlogs_df.loc[null_id_mask, 'id'] = int(datetime.now().timestamp() * 1000) # Reorder columns to match schema order slowlogs_df = slowlogs_df[schema_column_names] slowlogs_table = pa.Table.from_pandas(slowlogs_df, schema=slowlogs_schema, preserve_index=False) write_to_table(region=region, s3_path=slowlogs_path, merge_predicate=f"target.id = source.id and target.instance = source.instance and target.date = '{partition_date}'", source_pyarrow_table=slowlogs_table, partition_cols=['date'], merge_write=merge_write) del slowlogs_table del slowlogs_df gc.collect() write_slowlogs_data_end = time.time() logger.info(f"[{cluster}] {slowlogs_path} write_slowlogs_data_end, cost {write_slowlogs_data_end - slowlogs_data_start:.2f} seconds") if optimize: try: try_to_optimize_table(region, slowlogs_path, partition_date) optimize_table_slowlogs_end = time.time() logger.info(f"[{cluster}] {slowlogs_path} optimize_table_slowlogs_end, cost {optimize_table_slowlogs_end - write_slowlogs_data_end:.2f} seconds") except Exception as e: logger.error(f"[{cluster}] {slowlogs_path} Failed to optimize slowlogs table: {str(e)}") logger.info(f"[{cluster}] Data written successfully to Delta Lake tables") def write_slowlogs_to_deltalake_multitables(added_logs: List[SlowLog], region: str, cluster: str, base_path: str = "./deltalake", optimize: bool = True, partition_date: str = None, merge_write: bool = True): queries_path = f"{base_path}/queries" plans_path = f"{base_path}/plans" slowlogs_path = f"{base_path}/slowlogs" # Prepare data for each table queries_data = [] plans_data = [] slowlogs_data = [] # List of uint64 fields that need to be converted to string uint64_fields_to_convert = [ "txn_start_ts", "conn_id", "exec_retry_count", "preproc_subqueries", "request_count", "total_keys", "process_keys", "rocksdb_delete_skipped_count", "rocksdb_key_skipped_count", "rocksdb_block_cache_hit_count", "rocksdb_block_read_count", "rocksdb_block_read_byte" ] # If no partition_date is provided, use the date from the first log entry if partition_date is None and added_logs: dt_object_utc = datetime.fromtimestamp(added_logs[0].time, tz=timezone.utc) partition_date = dt_object_utc.strftime("%Y%m%d") for log in added_logs: # Queries table data if log.digest: queries_data.append({"digest": log.digest, "query": log.query}) # Plans table data if log.plan_digest: plans_data.append({ "plan_digest": log.plan_digest, "digest": log.digest, "binary_plan": log.binary_plan, "plan": log.plan, }) slowlogs_dict = log.__dict__.copy() slowlogs_dict.pop("query", None) slowlogs_dict.pop("binary_plan", None) slowlogs_dict.pop("plan", None) slowlogs_dict["date"] = partition_date # Convert specific uint64 fields to string for field in uint64_fields_to_convert: if field in slowlogs_dict and slowlogs_dict[field] is not None: slowlogs_dict[field] = str(slowlogs_dict[field]) slowlogs_data.append(slowlogs_dict) queries_data_start = time.time() if queries_data: queries_df = pd.DataFrame(queries_data).drop_duplicates(subset=["digest"]) # Ensure all schema columns exist in DataFrame, fill missing columns with None schema_column_names = [field.name for field in queries_schema] for col in schema_column_names: if col not in queries_df.columns: queries_df[col] = None # Reorder columns to match schema order queries_df = queries_df[schema_column_names] queries_table = pa.Table.from_pandas(queries_df, schema=queries_schema, preserve_index=False) write_to_table(region=region, s3_path=queries_path, merge_predicate="target.digest = source.digest", source_pyarrow_table=queries_table, partition_cols=None) write_queries_data_end = time.time() logger.info(f"[{cluster}] write_queries_data_end, cost {write_queries_data_end - queries_data_start:.2f} seconds") if optimize: try_to_optimize_table(region, queries_path, None) # No date-based optimization for queries table optimize_table_queries_end = time.time() logger.info(f"[{cluster}] optimize_table_queries_end, cost {optimize_table_queries_end - write_queries_data_end:.2f} seconds") plan_data_start = time.time() if plans_data: plans_df = pd.DataFrame(plans_data).drop_duplicates(subset=["plan_digest"]) if 'binary_plan' in plans_df.columns: plans_df['decoded_plan'] = plans_df['binary_plan'].apply( lambda x: parse_binary_plan(x) if x is not None and x != "" else None ) else: plans_df['decoded_plan'] = None # Ensure all schema columns exist in DataFrame, fill missing columns with None schema_column_names = [field.name for field in plans_schema] for col in schema_column_names: if col not in plans_df.columns: plans_df[col] = None # Reorder columns to match schema order plans_df = plans_df[schema_column_names] plans_table = pa.Table.from_pandas(plans_df, schema=plans_schema, preserve_index=False) write_to_table(region=region, s3_path=plans_path, merge_predicate="target.plan_digest = source.plan_digest", source_pyarrow_table=plans_table, partition_cols=None) write_plans_data_end = time.time() logger.info(f"[{cluster}] write_plans_data_end, cost {write_plans_data_end - plan_data_start:.2f} seconds") if optimize: try_to_optimize_table(region, plans_path, None) # No date-based optimization for plans table optimize_table_plans_end = time.time() logger.info(f"[{cluster}] optimize_table_plans_end, cost {optimize_table_plans_end - write_plans_data_end:.2f} seconds") slowlogs_data_start = time.time() if slowlogs_data: slowlogs_df = pd.DataFrame(slowlogs_data) if 'binary_plan' in slowlogs_df.columns: slowlogs_df['decoded_plan'] = slowlogs_df['binary_plan'].apply( lambda x: parse_binary_plan(x) if x is not None and x != "" else None ) else: slowlogs_df['decoded_plan'] = None # Ensure all schema columns exist in DataFrame, fill missing columns with None schema_column_names = [field.name for field in slowlogs_schema] for col in schema_column_names: if col not in slowlogs_df.columns: slowlogs_df[col] = None # Ensure id field is not null (required by schema) if 'id' in slowlogs_df.columns: null_id_mask = slowlogs_df['id'].isna() if null_id_mask.any(): # Fill null id values with 0 slowlogs_df.loc[null_id_mask, 'id'] = int(datetime.now().timestamp() * 1000) # Reorder columns to match schema order slowlogs_df = slowlogs_df[schema_column_names] slowlogs_table = pa.Table.from_pandas(slowlogs_df, schema=slowlogs_schema, preserve_index=False) write_to_table(region=region, s3_path=slowlogs_path, merge_predicate=f"target.id = source.id and target.instance = source.instance and target.date = '{partition_date}'", source_pyarrow_table=slowlogs_table, partition_cols=['date'], merge_write=merge_write) write_slowlogs_data_end = time.time() logger.info(f"[{cluster}] write_slowlogs_data_end, cost {write_slowlogs_data_end - slowlogs_data_start:.2f} seconds") if optimize: try: try_to_optimize_table(region, slowlogs_path, partition_date) optimize_table_slowlogs_end = time.time() logger.info(f"[{cluster}] optimize_table_slowlogs_end, cost {optimize_table_slowlogs_end - write_slowlogs_data_end:.2f} seconds") except Exception as e: logger.error(f"[{cluster}] Failed to optimize slowlogs table: {str(e)}") logger.info("Data written successfully to Delta Lake tables") # global thread pool, limit the number of concurrent writes DELTA_WRITE_POOL = ThreadPoolExecutor(max_workers=10) def write_to_table(region: str, s3_path: str, merge_predicate: str, source_pyarrow_table: pa.Table, partition_cols: Optional[List[str]] = None, merge_write: bool = True): if s3_path.startswith('abfss://'): storage_options = { "AZURE_STORAGE_ACCOUNT_NAME": os.getenv("AZURE_STORAGE_ACCOUNT"), "AZURE_STORAGE_USE_EMULATOR": "false" } else: # Existing AWS S3 configuration endpoint_url = os.getenv("AWS_ENDPOINT_S3") if endpoint_url: storage_options = { 'AWS_REGION': region, 'AWS_ENDPOINT_URL': endpoint_url, 'AWS_COPY_IF_NOT_EXISTS': 'header-with-status:x-oss-forbid-overwrite:true:409' } else: storage_options = { 'AWS_REGION': region, } def _write_task(): try: # Backward compatibility: keep `merge_write` parameter, but always use append. # We rely on `schema_mode="merge"` for schema evolution. write_deltalake( table_or_uri=s3_path, data=source_pyarrow_table, mode="append", schema_mode="merge", storage_options=storage_options, partition_by=partition_cols if partition_cols else None, ) except Exception as e: if "No such file or directory" in str(e) or "Table not found" in str(e) or "no log files" in str(e) or "DeltaTableError" in str(e): logger.info(f"Delta table not found at {s3_path}.") write_deltalake( table_or_uri=s3_path, data=source_pyarrow_table, mode="append", schema_mode="merge", storage_options=storage_options, partition_by=partition_cols if partition_cols else None, ) logger.info(f"New table created at {s3_path} .") else: logger.error(str(e)) future = DELTA_WRITE_POOL.submit(_write_task) future.result() #optimize_executor = ThreadPoolExecutor(max_workers=1) def try_to_optimize_table(region: str, s3_path: str, partition_date: date = None): """ try to optimize table :param region: region :param s3_path: s3 path :param partition_date: partition date """ future = DELTA_WRITE_POOL.submit(optimize_table, region, s3_path, partition_date) future.result() # if need async, can cancel below comment # optimize_executor.submit(optimize_table, region, s3_path, partition_date) def optimize_table(region: str, s3_path: str, partition_date: date = None): if s3_path.startswith('abfss://'): storage_options = { "AZURE_STORAGE_ACCOUNT_NAME": os.getenv("AZURE_STORAGE_ACCOUNT"), "AZURE_STORAGE_USE_EMULATOR": "false" } else: # Existing AWS S3 configuration endpoint_url = os.getenv("AWS_ENDPOINT_S3") if endpoint_url: storage_options = { 'AWS_REGION': region, 'AWS_ENDPOINT_URL': endpoint_url, 'AWS_COPY_IF_NOT_EXISTS': 'header-with-status:x-oss-forbid-overwrite:true:409' } else: storage_options = { 'AWS_REGION': region, } try: logger.info(f"Attempting to optimize (compact) table at {s3_path}") if partition_date: logger.info(f"Optimizing partition: date={partition_date}") delta_table = DeltaTable(s3_path, storage_options=storage_options) logger.info(f"Table version: {delta_table.version()}") optimize_metrics = None if partition_date: # Handle both date formats: YYYYMMDD and YYYY-MM-DD. today = datetime.strptime(partition_date, "%Y%m%d").date() yesterday = today - timedelta(days=1) compact_range = (yesterday.strftime("%Y%m%d"), today.strftime("%Y%m%d")) dashed_range = (yesterday.strftime("%Y-%m-%d"), today.strftime("%Y-%m-%d")) # Probe partition value format from data file paths to avoid wasted compact attempts. def detect_date_partition_format() -> Optional[str]: try: files = delta_table.files() except Exception as e: logger.warning(f"Failed to list delta files for partition format detection: {e}") return None has_compact = False has_dashed = False for rel_path in files: marker = "date=" idx = rel_path.find(marker) if idx == -1: continue value = rel_path[idx + len(marker):].split("/", 1)[0] if len(value) == 8 and value.isdigit(): has_compact = True elif len(value) == 10 and value[4] == "-" and value[7] == "-": has_dashed = True if has_compact and has_dashed: return "mixed" if has_compact: return "compact" if has_dashed: return "dashed" return None date_format = detect_date_partition_format() if date_format == "compact": candidate_ranges = [compact_range] elif date_format == "dashed": candidate_ranges = [dashed_range] else: # Unknown or mixed legacy data: fallback to both ranges. candidate_ranges = [compact_range, dashed_range] logger.info(f"Detected date partition format: {date_format or 'unknown'}") for start_date, end_date in candidate_ranges: logger.info(f"Optimizing partitions {s3_path} from {start_date} to {end_date}") metrics = delta_table.optimize.compact( partition_filters=[ ("date", ">=", start_date), ("date", "<=", end_date), ], target_size=512 * 1024 * 1024, ) logger.info(f"Slow Query table compacted. Metrics({start_date}~{end_date}): {metrics}") optimize_metrics = metrics else: optimize_metrics = delta_table.optimize.compact(target_size=512 * 1024 * 1024) logger.info(f"Slow Query table compacted. Metrics: {optimize_metrics}") logger.info(f"Attempting to vacuum slowlogs table at {s3_path}") delta_table = DeltaTable(s3_path, storage_options=storage_options) logger.info(f"Table version: {delta_table.version()}") logger.info(f"Starting vacuum {s3_path}") delta_table.alter.set_table_properties({"delta.logRetentionDuration": "interval 2 hours"}) delta_table.create_checkpoint() vacuum_metrics = delta_table.vacuum(retention_hours=2, enforce_retention_duration=False, dry_run=False) delta_table.cleanup_metadata() logger.info(f"Slow Query table vacuum. Metrics: {vacuum_metrics}") except Exception as e: logger.error(f"Error optimizing slow query table at {s3_path}: {e}") def hash_minute(path): h = hash(path) result_0_to_59 = h % 60 return result_0_to_59 def build_test_data(): dummy_logs: List[SlowLog] = [] # SlowLog 1 log1 = SlowLog() # Create an empty instance log1.id = 6 log1.instance = "i1" log1.time = datetime(2025, 6, 6, 10, 0, 0, tzinfo=timezone.utc).timestamp() log1.txn_start_ts = 100 log1.user = "u1" log1.host = "h1" log1.conn_id = "c1" log1.session_alias = "s1" log1.exec_retry_count = 0 log1.exec_retry_time = 0.0 log1.query_time = 0.1 log1.parse_time = 0.01 log1.compile_time = 0.01 log1.rewrite_time = 0.01 log1.preproc_subqueries = 0 log1.preproc_subqueries_time = 0.0 log1.optimize_time = 0.01 log1.wait_ts = 0.0 log1.prewrite_time = 0.0 log1.wait_prewrite_binlog_time = 0.0 log1.commit_time = 0.0 log1.get_commit_ts_time = 0.0 log1.commit_backoff_time = 0.0 log1.backoff_types = "" log1.resolve_lock_time = 0.0 log1.local_latch_wait_time = 0.0 log1.write_keys = 0 log1.write_size = 0 log1.prewrite_region = 0 log1.txn_retry = 0 log1.cop_time = 0.0 log1.process_time = 0.0 log1.wait_time = 0.0 log1.backoff_time = 0.0 log1.lockkeys_time = 0.0 log1.request_count = 0 log1.total_keys = 0 log1.process_keys = 0 log1.rocksdb_delete_skipped_count = 0 log1.rocksdb_key_skipped_count = 0 log1.rocksdb_block_cache_hit_count = 0 log1.rocksdb_block_read_count = 0 log1.rocksdb_block_read_byte = 0 log1.db = "db1" log1.index_names = "" log1.is_internal = False log1.digest = "d3" log1.stats = "" log1.cop_proc_avg = 0.0 log1.cop_proc_p90 = 0.0 log1.cop_proc_max = 0.0 log1.cop_proc_addr = "" log1.cop_wait_avg = 0.0 log1.cop_wait_p90 = 0.0 log1.cop_wait_max = 0.0 log1.cop_wait_addr = "" log1.mem_max = 0 log1.disk_max = 0 log1.kv_total = 0.0 log1.pd_total = 0.0 log1.backoff_total = 0.0 log1.write_sql_response_total = 0.0 log1.result_rows = 0 log1.warnings = "" log1.backoff_detail = "" log1.prepared = False log1.succ = True log1.is_explicit_txn = True log1.is_write_cache_table = False log1.plan_from_cache = False log1.plan_from_binding = False log1.has_more_results = False log1.resource_group = "" log1.request_unit_read = 0.0 log1.request_unit_v2 = 0.0 log1.request_unit_write = 0.0 log1.time_queued_by_rc = 0.0 log1.plan = "plan_details_1" log1.plan_digest = "pd44444" log1.binary_plan = "tidb_decode_binary_plan('vRbgCrgWCgpIYXNoQWdnXzMzEuQRCg5JbmRleExvb2tVcF8zNBLdBwoMU2VsZWN0aW9uXzI2EqkCChFJASTwaVJhbmdlU2Nhbl8yNCG8vwBkL3y1QCkM3KKOEw80QDC6AjgCQAJKbApqCgtkaF9hcHBfNTA1MBISZGhfdXNlcl9kYXlfcmVwb3J0GkcKGmlkeF91c2VyX3dpdGhkcmF3X3ltZF9zaXRlEggFLxhpZHgSCHdpCR1kEgpjcmVhdGVfeW1kEglzaXRlX2NvZGVSK3IBlSg6WzM5OTAzNjMxLBEJwF0sIGtlZXAgb3JkZXI6ZmFsc2VqSnRpa3ZfdGFzazp7cHJvYyBtYXg6MHMsIG1pbjoBCBBhdmc6IAEJDHA4MDoFCAQ5NQUIIGl0ZXJzOjUsIAE9FHM6Mn1w/xEBBAF4EQow/wEaAQEh1OTnsx9luUIZARBSNGVxKD0YBC5kQhcBAC4V2JAsICI1MDUwIilaFHRpbWU6MS44M21zLCBsb29wczozYqECY29wCcokIHtudW06IDIsIAHOICA5NjUuMsK1cwnVFCA3ODUuMwUPBdwEODcRDwhwOTUyLQAIYXhfIRMoX2tleXM6IDIyNCwBI0IUAAh0b3QFFBg6IDUwNC44BVoBFCR3YWl0OiA5Mi45BROAY29wcl9jYWNoZV9oaXRfcmF0aW86IDAuMDAsIGJ1aWxkBbwIX2R1BRoYbjogMzEuMgHhAYiYZGlzdHNxbF9jb25jdXJyZW5jeTogMX0sIHJwY19pbmZvOntDb3A6AfgYX3JwYzoyLAGDCGFsXy0rIDcybXN9fWr+Af7vATnvCCwgc0H4KGRldGFpbDoge3RvAXAhMQhlc3MtNAgzMTQRiS4ZACxfc2l6ZTogMzg5MzYRICllQDMyOSwgZ2V0X3NuYXBzaG90CcEIIDQxJUSocm9ja3NkYjoge2tleV9za2lwcGVkX2NvdW50OiAzMjcsIGJsb2NrOiB7YzVpDR4MMjV9fVqjAhQSxgcKCUiNAjw4EtEBChFUYWJsZVJvd0lEad0gNSEd1+xuvFvBQsQCDEojCiGC3QMEUhBCeQMAVk6KAQQxMCXxAGl9ewgxMDRNQVaBAwQyOCFDBGFzYYMEODZa4gBYGgECIWHdn+mPgMVAKTUErcHDJvA/MJ6FnihSa2dyb3VwIGJ5OnqLA5GDKCwgZnVuY3M6c3Vtfr4DPGZ1bmQpLT5Db2x1bW4jNjZxvAg4LjEB6mm8CDJitka8AwHXZb4MMTcuNgEsIRYIIDUzcU9lvAwxLjA0ARxluggxLjIFDWEwQW5RU0a2A0V0BSQUOiAzNy45AUN5sgQ1NAkSAGO+sQMIOTc1CdcBglayA4WSGF9leHRyYV82zAMEMDZqzgMhDWlgZdAIMjkxIRQMfX1q9lLRA+5HAklH7t0Dad0UMTgwMjk5JWoEYWwxi4EPAGdK3gMANQHPACx93wBiXscDCDUwMGbJAxwhmV9lun9O4zbkAhABQAFaEyU4DDEuMDPVMhAyYusBaeHEpXiVwQUpDCAxLjhFdSxmZXRjaF9oYW5kbGUFFgA1QSylryA6IDk5Mm5zLCCp6AAyxVQAfeEPCGJsZU5UAAg5OTMJ58WdCDEsIDIBAig1fSwgbmV4dDogewFRAF8FmxA6IDMzLmUqBRQAdAVbGGxvb2t1cF8RhQg3NS5F90YiABRyZXNwOiBpbAx9cLDiLpcHHCF+1ZtEbxHkPSsAASEqDFKzAWfqDwR16Xn1BDU0kTAcZmlyc3Ryb3d+NQSRaAgpLT6akgRW4AFQ0AJwYXJ0aWFsX3dvcmtlcjp7d2Fs9RMkMDI2ODM2NzQxczaKARrSCABfIaYAMUXA6cokNS4xMzM3OTAwNwEygS8kZXhlYzo1NS44NcmeARRFPwUrFDg1NjYxNQ51CARheEF6IDI2NzkzMzI5c4WQMhIAFH0sIGZpbmqoABA1MzQyMqqoABw0MDI4OTc4cwm/BagUMTUuNDI25UYyqAAUNDA0ODAwMqgAFDgxOTQwNC6oAA0SRH1w/fYCeP///////////wEYAQ==')" log1.prev_stmt = "" log1.query = "SELECT * FROM table1;" log1.serverless_tenant_id = "" log1.serverless_project_id = "" log1.serverless_cluster_id = "" log1.worker_role = "" log1.is_tiflash = False dummy_logs.append(log1) # SlowLog 2 log2 = SlowLog() # Create an empty instance log2.id = 2 log2.instance = "i1" log2.time = datetime(2025, 6, 6, 11, 30, 0, tzinfo=timezone.utc).timestamp() log2.txn_start_ts = 105 log2.user = "u2" log2.host = "h1" log2.conn_id = "c2" log2.session_alias = "s2" log2.exec_retry_count = 0 log2.exec_retry_time = 0.0 log2.query_time = 0.2 log2.parse_time = 0.02 log2.compile_time = 0.02 log2.rewrite_time = 0.02 log2.preproc_subqueries = 0 log2.preproc_subqueries_time = 0.0 log2.optimize_time = 0.02 log2.wait_ts = 0.0 log2.prewrite_time = 0.0 log2.wait_prewrite_binlog_time = 0.0 log2.commit_time = 0.0 log2.get_commit_ts_time = 0.0 log2.commit_backoff_time = 0.0 log2.backoff_types = "" log2.resolve_lock_time = 0.0 log2.local_latch_wait_time = 0.0 log2.write_keys = 0 log2.write_size = 0 log2.prewrite_region = 0 log2.txn_retry = 0 log2.cop_time = 0.0 log2.process_time = 0.0 log2.wait_time = 0.0 log2.backoff_time = 0.0 log2.lockkeys_time = 0.0 log2.request_count = 0 log2.total_keys = 0 log2.process_keys = 0 log2.rocksdb_delete_skipped_count = 0 log2.rocksdb_key_skipped_count = 0 log2.rocksdb_block_cache_hit_count = 0 log2.rocksdb_block_read_count = 0 log2.rocksdb_block_read_byte = 0 log2.db = "db2" log2.index_names = "" log2.is_internal = False log2.digest = "d2" log2.stats = "" log2.cop_proc_avg = 0.0 log2.cop_proc_p90 = 0.0 log2.cop_proc_max = 0.0 log2.cop_proc_addr = "" log2.cop_wait_avg = 0.0 log2.cop_wait_p90 = 0.0 log2.cop_wait_max = 0.0 log2.cop_wait_addr = "" log2.mem_max = 0 log2.disk_max = 0 log2.kv_total = 0.0 log2.pd_total = 0.0 log2.backoff_total = 0.0 log2.write_sql_response_total = 0.0 log2.result_rows = 0 log2.warnings = "" log2.backoff_detail = "" log2.prepared = False log2.succ = True log2.is_explicit_txn = True log2.is_write_cache_table = False log2.plan_from_cache = False log2.plan_from_binding = False log2.has_more_results = False log2.resource_group = "" log2.request_unit_read = 0.0 log2.request_unit_v2 = 0.0 log2.request_unit_write = 0.0 log2.time_queued_by_rc = 0.0 log2.plan = "plan_details_2" log2.plan_digest = "pd2" log2.binary_plan = "tidb_decode_binary_plan('vRbgCrgWCgpIYXNoQWdnXzMzEuQRCg5JbmRleExvb2tVcF8zNBLdBwoMU2VsZWN0aW9uXzI2EqkCChFJASTwaVJhbmdlU2Nhbl8yNCG8vwBkL3y1QCkM3KKOEw80QDC6AjgCQAJKbApqCgtkaF9hcHBfNTA1MBISZGhfdXNlcl9kYXlfcmVwb3J0GkcKGmlkeF91c2VyX3dpdGhkcmF3X3ltZF9zaXRlEggFLxhpZHgSCHdpCR1kEgpjcmVhdGVfeW1kEglzaXRlX2NvZGVSK3IBlSg6WzM5OTAzNjMxLBEJwF0sIGtlZXAgb3JkZXI6ZmFsc2VqSnRpa3ZfdGFzazp7cHJvYyBtYXg6MHMsIG1pbjoBCBBhdmc6IAEJDHA4MDoFCAQ5NQUIIGl0ZXJzOjUsIAE9FHM6Mn1w/xEBBAF4EQow/wEaAQEh1OTnsx9luUIZARBSNGVxKD0YBC5kQhcBAC4V2JAsICI1MDUwIilaFHRpbWU6MS44M21zLCBsb29wczozYqECY29wCcokIHtudW06IDIsIAHOICA5NjUuMsK1cwnVFCA3ODUuMwUPBdwEODcRDwhwOTUyLQAIYXhfIRMoX2tleXM6IDIyNCwBI0IUAAh0b3QFFBg6IDUwNC44BVoBFCR3YWl0OiA5Mi45BROAY29wcl9jYWNoZV9oaXRfcmF0aW86IDAuMDAsIGJ1aWxkBbwIX2R1BRoYbjogMzEuMgHhAYiYZGlzdHNxbF9jb25jdXJyZW5jeTogMX0sIHJwY19pbmZvOntDb3A6AfgYX3JwYzoyLAGDCGFsXy0rIDcybXN9fWr+Af7vATnvCCwgc0H4KGRldGFpbDoge3RvAXAhMQhlc3MtNAgzMTQRiS4ZACxfc2l6ZTogMzg5MzYRICllQDMyOSwgZ2V0X3NuYXBzaG90CcEIIDQxJUSocm9ja3NkYjoge2tleV9za2lwcGVkX2NvdW50OiAzMjcsIGJsb2NrOiB7YzVpDR4MMjV9fVqjAhQSxgcKCUiNAjw4EtEBChFUYWJsZVJvd0lEad0gNSEd1+xuvFvBQsQCDEojCiGC3QMEUhBCeQMAVk6KAQQxMCXxAGl9ewgxMDRNQVaBAwQyOCFDBGFzYYMEODZa4gBYGgECIWHdn+mPgMVAKTUErcHDJvA/MJ6FnihSa2dyb3VwIGJ5OnqLA5GDKCwgZnVuY3M6c3Vtfr4DPGZ1bmQpLT5Db2x1bW4jNjZxvAg4LjEB6mm8CDJitka8AwHXZb4MMTcuNgEsIRYIIDUzcU9lvAwxLjA0ARxluggxLjIFDWEwQW5RU0a2A0V0BSQUOiAzNy45AUN5sgQ1NAkSAGO+sQMIOTc1CdcBglayA4WSGF9leHRyYV82zAMEMDZqzgMhDWlgZdAIMjkxIRQMfX1q9lLRA+5HAklH7t0Dad0UMTgwMjk5JWoEYWwxi4EPAGdK3gMANQHPACx93wBiXscDCDUwMGbJAxwhmV9lun9O4zbkAhABQAFaEyU4DDEuMDPVMhAyYusBaeHEpXiVwQUpDCAxLjhFdSxmZXRjaF9oYW5kbGUFFgA1QSylryA6IDk5Mm5zLCCp6AAyxVQAfeEPCGJsZU5UAAg5OTMJ58WdCDEsIDIBAig1fSwgbmV4dDogewFRAF8FmxA6IDMzLmUqBRQAdAVbGGxvb2t1cF8RhQg3NS5F90YiABRyZXNwOiBpbAx9cLDiLpcHHCF+1ZtEbxHkPSsAASEqDFKzAWfqDwR16Xn1BDU0kTAcZmlyc3Ryb3d+NQSRaAgpLT6akgRW4AFQ0AJwYXJ0aWFsX3dvcmtlcjp7d2Fs9RMkMDI2ODM2NzQxczaKARrSCABfIaYAMUXA6cokNS4xMzM3OTAwNwEygS8kZXhlYzo1NS44NcmeARRFPwUrFDg1NjYxNQ51CARheEF6IDI2NzkzMzI5c4WQMhIAFH0sIGZpbmqoABA1MzQyMqqoABw0MDI4OTc4cwm/BagUMTUuNDI25UYyqAAUNDA0ODAwMqgAFDgxOTQwNC6oAA0SRH1w/fYCeP///////////wEYAQ==')" log2.prev_stmt = "" log2.query = "UPDATE table2 SET col=1;" log2.serverless_tenant_id = "" log2.serverless_project_id = "" log2.serverless_cluster_id = "" log2.worker_role = "" log2.is_tiflash = False dummy_logs.append(log2) # SlowLog 3 (Duplicated digest and plan_digest, but new date for slowlogs table) log3 = SlowLog() log3.id = 3 log3.instance = "i2" log3.time = datetime(2025, 6, 9, 9, 0, 0, tzinfo=timezone.utc).timestamp() log3.txn_start_ts = 110 log3.user = "u3" log3.host = "h2" log3.conn_id = "c3" log3.session_alias = "s3" log3.exec_retry_count = 0 log3.exec_retry_time = 0.0 log3.query_time = 0.3 log3.parse_time = 0.03 log3.compile_time = 0.03 log3.rewrite_time = 0.03 log3.preproc_subqueries = 0 log3.preproc_subqueries_time = 0.0 log3.optimize_time = 0.03 log3.wait_ts = 0.0 log3.prewrite_time = 0.0 log3.wait_prewrite_binlog_time = 0.0 log3.commit_time = 0.0 log3.get_commit_ts_time = 0.0 log3.commit_backoff_time = 0.0 log3.backoff_types = "" log3.resolve_lock_time = 0.0 log3.local_latch_wait_time = 0.0 log3.write_keys = 0 log3.write_size = 0 log3.prewrite_region = 0 log3.txn_retry = 0 log3.cop_time = 0.0 log3.process_time = 0.0 log3.wait_time = 0.0 log3.backoff_time = 0.0 log3.lockkeys_time = 0.0 log3.request_count = 0 log3.total_keys = 0 log3.process_keys = 0 log3.rocksdb_delete_skipped_count = 0 log3.rocksdb_key_skipped_count = 0 log3.rocksdb_block_cache_hit_count = 0 log3.rocksdb_block_read_count = 0 log3.rocksdb_block_read_byte = 0 log3.db = "db1" log3.index_names = "" log3.is_internal = False log3.digest = "d1" log3.stats = "" log3.cop_proc_avg = 0.0 log3.cop_proc_p90 = 0.0 log3.cop_proc_max = 0.0 log3.cop_proc_addr = "" log3.cop_wait_avg = 0.0 log3.cop_wait_p90 = 0.0 log3.cop_wait_max = 0.0 log3.cop_wait_addr = "" log3.mem_max = 0 log3.disk_max = 0 log3.kv_total = 0.0 log3.pd_total = 0.0 log3.backoff_total = 0.0 log3.write_sql_response_total = 0.0 log3.result_rows = 0 log3.warnings = "" log3.backoff_detail = "" log3.prepared = False log3.succ = True log3.is_explicit_txn = True log3.is_write_cache_table = False log3.plan_from_cache = False log3.plan_from_binding = False log3.has_more_results = False log3.resource_group = "" log3.request_unit_read = 0.0 log3.request_unit_v2 = 0.0 log3.request_unit_write = 0.0 log3.time_queued_by_rc = 0.0 log3.plan = "plan_details_1_updated" log3.plan_digest = "pd1" log3.binary_plan = "tidb_decode_binary_plan('vRbgCrgWCgpIYXNoQWdnXzMzEuQRCg5JbmRleExvb2tVcF8zNBLdBwoMU2VsZWN0aW9uXzI2EqkCChFJASTwaVJhbmdlU2Nhbl8yNCG8vwBkL3y1QCkM3KKOEw80QDC6AjgCQAJKbApqCgtkaF9hcHBfNTA1MBISZGhfdXNlcl9kYXlfcmVwb3J0GkcKGmlkeF91c2VyX3dpdGhkcmF3X3ltZF9zaXRlEggFLxhpZHgSCHdpCR1kEgpjcmVhdGVfeW1kEglzaXRlX2NvZGVSK3IBlSg6WzM5OTAzNjMxLBEJwF0sIGtlZXAgb3JkZXI6ZmFsc2VqSnRpa3ZfdGFzazp7cHJvYyBtYXg6MHMsIG1pbjoBCBBhdmc6IAEJDHA4MDoFCAQ5NQUIIGl0ZXJzOjUsIAE9FHM6Mn1w/xEBBAF4EQow/wEaAQEh1OTnsx9luUIZARBSNGVxKD0YBC5kQhcBAC4V2JAsICI1MDUwIilaFHRpbWU6MS44M21zLCBsb29wczozYqECY29wCcokIHtudW06IDIsIAHOICA5NjUuMsK1cwnVFCA3ODUuMwUPBdwEODcRDwhwOTUyLQAIYXhfIRMoX2tleXM6IDIyNCwBI0IUAAh0b3QFFBg6IDUwNC44BVoBFCR3YWl0OiA5Mi45BROAY29wcl9jYWNoZV9oaXRfcmF0aW86IDAuMDAsIGJ1aWxkBbwIX2R1BRoYbjogMzEuMgHhAYiYZGlzdHNxbF9jb25jdXJyZW5jeTogMX0sIHJwY19pbmZvOntDb3A6AfgYX3JwYzoyLAGDCGFsXy0rIDcybXN9fWr+Af7vATnvCCwgc0H4KGRldGFpbDoge3RvAXAhMQhlc3MtNAgzMTQRiS4ZACxfc2l6ZTogMzg5MzYRICllQDMyOSwgZ2V0X3NuYXBzaG90CcEIIDQxJUSocm9ja3NkYjoge2tleV9za2lwcGVkX2NvdW50OiAzMjcsIGJsb2NrOiB7YzVpDR4MMjV9fVqjAhQSxgcKCUiNAjw4EtEBChFUYWJsZVJvd0lEad0gNSEd1+xuvFvBQsQCDEojCiGC3QMEUhBCeQMAVk6KAQQxMCXxAGl9ewgxMDRNQVaBAwQyOCFDBGFzYYMEODZa4gBYGgECIWHdn+mPgMVAKTUErcHDJvA/MJ6FnihSa2dyb3VwIGJ5OnqLA5GDKCwgZnVuY3M6c3Vtfr4DPGZ1bmQpLT5Db2x1bW4jNjZxvAg4LjEB6mm8CDJitka8AwHXZb4MMTcuNgEsIRYIIDUzcU9lvAwxLjA0ARxluggxLjIFDWEwQW5RU0a2A0V0BSQUOiAzNy45AUN5sgQ1NAkSAGO+sQMIOTc1CdcBglayA4WSGF9leHRyYV82zAMEMDZqzgMhDWlgZdAIMjkxIRQMfX1q9lLRA+5HAklH7t0Dad0UMTgwMjk5JWoEYWwxi4EPAGdK3gMANQHPACx93wBiXscDCDUwMGbJAxwhmV9lun9O4zbkAhABQAFaEyU4DDEuMDPVMhAyYusBaeHEpXiVwQUpDCAxLjhFdSxmZXRjaF9oYW5kbGUFFgA1QSylryA6IDk5Mm5zLCCp6AAyxVQAfeEPCGJsZU5UAAg5OTMJ58WdCDEsIDIBAig1fSwgbmV4dDogewFRAF8FmxA6IDMzLmUqBRQAdAVbGGxvb2t1cF8RhQg3NS5F90YiABRyZXNwOiBpbAx9cLDiLpcHHCF+1ZtEbxHkPSsAASEqDFKzAWfqDwR16Xn1BDU0kTAcZmlyc3Ryb3d+NQSRaAgpLT6akgRW4AFQ0AJwYXJ0aWFsX3dvcmtlcjp7d2Fs9RMkMDI2ODM2NzQxczaKARrSCABfIaYAMUXA6cokNS4xMzM3OTAwNwEygS8kZXhlYzo1NS44NcmeARRFPwUrFDg1NjYxNQ51CARheEF6IDI2NzkzMzI5c4WQMhIAFH0sIGZpbmqoABA1MzQyMqqoABw0MDI4OTc4cwm/BagUMTUuNDI25UYyqAAUNDA0ODAwMqgAFDgxOTQwNC6oAA0SRH1w/fYCeP///////////wEYAQ==')" log3.prev_stmt = "" log3.query = "SELECT * FROM table1 UPDATED;" log3.serverless_tenant_id = "" log3.serverless_project_id = "" log3.serverless_cluster_id = "" log3.worker_role = "" log3.is_tiflash = False dummy_logs.append(log3) # SlowLog 4 (New id, but same digest and plan_digest as a previous one, and same date as ID 1) log4 = SlowLog() log4.id = 4 log4.instance = "i3" log4.time = datetime(2025, 6, 6, 12, 0, 0, tzinfo=timezone.utc).timestamp() log4.txn_start_ts = 115 log4.user = "u4" log4.host = "h2" log4.conn_id = "c4" log4.session_alias = "s4" log4.exec_retry_count = 0 log4.exec_retry_time = 0.0 log4.query_time = 0.4 log4.parse_time = 0.04 log4.compile_time = 0.04 log4.rewrite_time = 0.04 log4.preproc_subqueries = 0 log4.preproc_subqueries_time = 0.0 log4.optimize_time = 0.04 log4.wait_ts = 0.0 log4.prewrite_time = 0.0 log4.wait_prewrite_binlog_time = 0.0 log4.commit_time = 0.0 log4.get_commit_ts_time = 0.0 log4.commit_backoff_time = 0.0 log4.backoff_types = "" log4.resolve_lock_time = 0.0 log4.local_latch_wait_time = 0.0 log4.write_keys = 0 log4.write_size = 0 log4.prewrite_region = 0 log4.txn_retry = 0 log4.cop_time = 0.0 log4.process_time = 0.0 log4.wait_time = 0.0 log4.backoff_time = 0.0 log4.lockkeys_time = 0.0 log4.request_count = 0 log4.total_keys = 0 log4.process_keys = 0 log4.rocksdb_delete_skipped_count = 0 log4.rocksdb_key_skipped_count = 0 log4.rocksdb_block_cache_hit_count = 0 log4.rocksdb_block_read_count = 0 log4.rocksdb_block_read_byte = 0 log4.db = "db3" log4.index_names = "" log4.is_internal = False log4.digest = "d1" log4.stats = "" log4.cop_proc_avg = 0.0 log4.cop_proc_p90 = 0.0 log4.cop_proc_max = 0.0 log4.cop_proc_addr = "" log4.cop_wait_avg = 0.0 log4.cop_wait_p90 = 0.0 log4.cop_wait_max = 0.0 log4.cop_wait_addr = "" log4.mem_max = 0 log4.disk_max = 0 log4.kv_total = 0.0 log4.pd_total = 0.0 log4.backoff_total = 0.0 log4.write_sql_response_total = 0.0 log4.result_rows = 0 log4.warnings = "" log4.backoff_detail = "" log4.prepared = False log4.succ = True log4.is_explicit_txn = True log4.is_write_cache_table = False log4.plan_from_cache = False log4.plan_from_binding = False log4.has_more_results = False log4.resource_group = "" log4.request_unit_read = 0.0 log4.request_unit_v2 = 0.0 log4.request_unit_write = 0.0 log4.time_queued_by_rc = 0.0 log4.plan = "plan_details_1" log4.plan_digest = "pd1" log4.binary_plan = "tidb_decode_binary_plan('vRbgCrgWCgpIYXNoQWdnXzMzEuQRCg5JbmRleExvb2tVcF8zNBLdBwoMU2VsZWN0aW9uXzI2EqkCChFJASTwaVJhbmdlU2Nhbl8yNCG8vwBkL3y1QCkM3KKOEw80QDC6AjgCQAJKbApqCgtkaF9hcHBfNTA1MBISZGhfdXNlcl9kYXlfcmVwb3J0GkcKGmlkeF91c2VyX3dpdGhkcmF3X3ltZF9zaXRlEggFLxhpZHgSCHdpCR1kEgpjcmVhdGVfeW1kEglzaXRlX2NvZGVSK3IBlSg6WzM5OTAzNjMxLBEJwF0sIGtlZXAgb3JkZXI6ZmFsc2VqSnRpa3ZfdGFzazp7cHJvYyBtYXg6MHMsIG1pbjoBCBBhdmc6IAEJDHA4MDoFCAQ5NQUIIGl0ZXJzOjUsIAE9FHM6Mn1w/xEBBAF4EQow/wEaAQEh1OTnsx9luUIZARBSNGVxKD0YBC5kQhcBAC4V2JAsICI1MDUwIilaFHRpbWU6MS44M21zLCBsb29wczozYqECY29wCcokIHtudW06IDIsIAHOICA5NjUuMsK1cwnVFCA3ODUuMwUPBdwEODcRDwhwOTUyLQAIYXhfIRMoX2tleXM6IDIyNCwBI0IUAAh0b3QFFBg6IDUwNC44BVoBFCR3YWl0OiA5Mi45BROAY29wcl9jYWNoZV9oaXRfcmF0aW86IDAuMDAsIGJ1aWxkBbwIX2R1BRoYbjogMzEuMgHhAYiYZGlzdHNxbF9jb25jdXJyZW5jeTogMX0sIHJwY19pbmZvOntDb3A6AfgYX3JwYzoyLAGDCGFsXy0rIDcybXN9fWr+Af7vATnvCCwgc0H4KGRldGFpbDoge3RvAXAhMQhlc3MtNAgzMTQRiS4ZACxfc2l6ZTogMzg5MzYRICllQDMyOSwgZ2V0X3NuYXBzaG90CcEIIDQxJUSocm9ja3NkYjoge2tleV9za2lwcGVkX2NvdW50OiAzMjcsIGJsb2NrOiB7YzVpDR4MMjV9fVqjAhQSxgcKCUiNAjw4EtEBChFUYWJsZVJvd0lEad0gNSEd1+xuvFvBQsQCDEojCiGC3QMEUhBCeQMAVk6KAQQxMCXxAGl9ewgxMDRNQVaBAwQyOCFDBGFzYYMEODZa4gBYGgECIWHdn+mPgMVAKTUErcHDJvA/MJ6FnihSa2dyb3VwIGJ5OnqLA5GDKCwgZnVuY3M6c3Vtfr4DPGZ1bmQpLT5Db2x1bW4jNjZxvAg4LjEB6mm8CDJitka8AwHXZb4MMTcuNgEsIRYIIDUzcU9lvAwxLjA0ARxluggxLjIFDWEwQW5RU0a2A0V0BSQUOiAzNy45AUN5sgQ1NAkSAGO+sQMIOTc1CdcBglayA4WSGF9leHRyYV82zAMEMDZqzgMhDWlgZdAIMjkxIRQMfX1q9lLRA+5HAklH7t0Dad0UMTgwMjk5JWoEYWwxi4EPAGdK3gMANQHPACx93wBiXscDCDUwMGbJAxwhmV9lun9O4zbkAhABQAFaEyU4DDEuMDPVMhAyYusBaeHEpXiVwQUpDCAxLjhFdSxmZXRjaF9oYW5kbGUFFgA1QSylryA6IDk5Mm5zLCCp6AAyxVQAfeEPCGJsZU5UAAg5OTMJ58WdCDEsIDIBAig1fSwgbmV4dDogewFRAF8FmxA6IDMzLmUqBRQAdAVbGGxvb2t1cF8RhQg3NS5F90YiABRyZXNwOiBpbAx9cLDiLpcHHCF+1ZtEbxHkPSsAASEqDFKzAWfqDwR16Xn1BDU0kTAcZmlyc3Ryb3d+NQSRaAgpLT6akgRW4AFQ0AJwYXJ0aWFsX3dvcmtlcjp7d2Fs9RMkMDI2ODM2NzQxczaKARrSCABfIaYAMUXA6cokNS4xMzM3OTAwNwEygS8kZXhlYzo1NS44NcmeARRFPwUrFDg1NjYxNQ51CARheEF6IDI2NzkzMzI5c4WQMhIAFH0sIGZpbmqoABA1MzQyMqqoABw0MDI4OTc4cwm/BagUMTUuNDI25UYyqAAUNDA0ODAwMqgAFDgxOTQwNC6oAA0SRH1w/fYCeP///////////wEYAQ==')" log4.prev_stmt = "" log4.query = "SELECT * FROM table1;" log4.serverless_tenant_id = "" log4.serverless_project_id = "" log4.serverless_cluster_id = "" log4.worker_role = "" log4.is_tiflash = False dummy_logs.append(log4) return dummy_logs if __name__ == "__main__": print("__main__") dummy_logs = build_test_data() write_slowlogs_to_deltalake(dummy_logs, "us-west-2", "s3://o11y-dev-shared-us-west-2/deltalake") print("\nVerifying data in Delta Lake tables:") queries_dt = DeltaTable("s3://o11y-dev-shared-us-west-2/deltalake/queries", storage_options={"AWS_REGION": "us-west-2"}) print("\nQueries Table:") print(queries_dt.to_pandas()) plans_dt = DeltaTable("s3://o11y-dev-shared-us-west-2/deltalake/plans", storage_options={"AWS_REGION": "us-west-2"}) print("\nPlans Table:") print(plans_dt.to_pandas()) slowlogs_dt = DeltaTable("s3://o11y-dev-shared-us-west-2/deltalake/slowlogs", storage_options={"AWS_REGION": "us-west-2"}) print("\nSlowlogs Table:") print(slowlogs_dt.to_pandas()) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209238000/25d59432-9a9e-47cc-880b-d8ddd39458b7/slowlogs") # pd.set_option('display.max_rows', 20) # pd.set_option('display.max_columns', 20) # pd.set_option('display.max_colwidth', 50) # pd.set_option('display.width', 1000) # # queries_dt = DeltaTable("s3://o11y-prod-shared-us-east-1/datalake/1372813089209061633/413c0c17-49ee-4f66-bbfd-b5b339835184/queries", storage_options={"AWS_REGION": "us-east-1"}) # print("\nQueries Table:") # print(queries_dt.to_pandas()) # # # # plans_dt = DeltaTable("s3://o11y-prod-shared-ap-east-1/datalake/1372813089209238000/25d59432-9a9e-47cc-880b-d8ddd39458b7/plans", storage_options={"AWS_REGION": "ap-east-1"}) # # # print("\nPlans Table:") # # # print(plans_dt.to_pandas()) # # # slowlog_dt = DeltaTable("s3://o11y-prod-shared-us-east-1/datalake/1372813089209061633/413c0c17-49ee-4f66-bbfd-b5b339835184/slowlogs", storage_options={"AWS_REGION": "us-east-1"}) # print("\nSlowlogs Table:") # print(slowlog_dt.to_pandas()) # # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209238000/25d59432-9a9e-47cc-880b-d8ddd39458b7/slowlogs", True) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209238000/25d59432-9a9e-47cc-880b-d8ddd39458b7/plans", False) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209238000/25d59432-9a9e-47cc-880b-d8ddd39458b7/queries", False) # # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209259175/970e2198-2921-4893-bcb9-583933e0defb/slowlogs", True) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209259175/970e2198-2921-4893-bcb9-583933e0defb/plans", False) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209259175/970e2198-2921-4893-bcb9-583933e0defb/queries", False) # # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209265146/21688d2c-57d3-49f5-9313-d921146e4255/slowlogs", True) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209265146/21688d2c-57d3-49f5-9313-d921146e4255/plans", False) # optimize_table("ap-east-1", "s3://o11y-prod-shared-ap-east-1/datalake/1372813089209265146/21688d2c-57d3-49f5-9313-d921146e4255/queries", False)