Refresh stale InnoDB row estimates

# SPDX-License-Identifier: Apache-2.0
id: mysql-stale-innodb-statistics-refresh-guard
namespace: company.team
description: Refresh stale persistent InnoDB row estimates after a known bulk
  load using a locked single-table plan.
concurrency:
  limit: 1
  behavior: QUEUE
inputs:
  - id: mysql_url
    type: STRING
    defaults: jdbc:mysql://localhost:3306/app
  - id: mysql_username
    type: STRING
    defaults: "{{ secret('MYSQL_USERNAME') }}"
  - id: mysql_password
    type: SECRET
    defaults: "{{ secret('MYSQL_PASSWORD') }}"
  - id: schema_name
    type: STRING
    defaults: app
  - id: table_name
    type: STRING
    defaults: events
  - id: max_table_mb
    type: INT
    defaults: 16
    min: 1
    max: 128
  - id: min_rows
    type: INT
    defaults: 1000
    min: 1
  - id: count_timeout_ms
    type: INT
    defaults: 10000
    min: 1
    max: 10000
  - id: min_stats_age_seconds
    type: INT
    defaults: 3600
    min: 0
  - id: drift_ratio
    type: FLOAT
    defaults: 0.5
    min: 0.1
    max: 1
  - id: verify_tolerance
    type: FLOAT
    defaults: 0.25
    min: 0.01
    max: 0.5
  - id: dry_run
    type: BOOL
    defaults: true
  - id: auto_remediate
    type: BOOL
    defaults: false
  - id: slack_webhook_url
    type: SECRET
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
variables:
  jdbc_url: "{{ inputs.mysql_url }}{{ inputs.mysql_url contains '?' ? '&' : '?'
    }}connectTimeout=5000&socketTimeout=30000&sessionVariables=max_execution_ti\
    me={{ inputs.count_timeout_ms
    }},lock_wait_timeout=2,innodb_lock_wait_timeout=2"
tasks:
  - id: validate_target
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ inputs.schema_name | length > 0 and inputs.schema_name | length <=
        64 and inputs.schema_name | replace({'[A-Za-z0-9_]': ''}, regexp=true) |
        length == 0 }}"
      - "{{ inputs.table_name | length > 0 and inputs.table_name | length <= 64
        and inputs.table_name | replace({'[A-Za-z0-9_]': ''}, regexp=true) |
        length == 0 }}"
      - "{{ inputs.schema_name != 'mysql' and inputs.schema_name != 'sys' and
        inputs.schema_name != 'performance_schema' and inputs.schema_name !=
        'information_schema' }}"
    errorMessage: Choose an exact ASCII user schema/table. System schemas and SQL
      fragments are refused.
  - id: metadata
    type: io.kestra.plugin.jdbc.mysql.Query
    connectionPooling: false
    url: "{{ render(vars.jdbc_url) }}"
    username: "{{ inputs.mysql_username }}"
    password: "{{ inputs.mysql_password }}"
    fetchType: FETCH
    sql: |
      SELECT CAST(t.TABLE_ID AS SIGNED) AS table_id, CAST(s.FILE_SIZE AS SIGNED) AS file_bytes, CAST(st.n_rows AS SIGNED) AS estimated_rows,
        TIMESTAMPDIFF(SECOND,st.last_update,NOW()) AS stats_age_seconds
      FROM information_schema.INNODB_TABLES t
      JOIN information_schema.INNODB_TABLESPACES s ON s.SPACE=t.SPACE
      JOIN mysql.innodb_table_stats st ON st.database_name='{{ inputs.schema_name }}' AND st.table_name='{{ inputs.table_name }}'
      WHERE t.NAME='{{ inputs.schema_name }}/{{ inputs.table_name }}' AND t.SPACE_TYPE='Single'
  - id: supported
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ outputs.metadata.size == 1 }}"
      - "{{ outputs.metadata.rows[0].file_bytes <= inputs.max_table_mb * 1048576
        }}"
    errorMessage: Target is missing, partitioned, lacks persistent stats or exceeds
      the physical file-size cap.
  - id: count_rows
    type: io.kestra.plugin.jdbc.mysql.Query
    connectionPooling: false
    url: "{{ render(vars.jdbc_url) }}"
    username: "{{ inputs.mysql_username }}"
    password: "{{ inputs.mysql_password }}"
    fetchType: FETCH_ONE
    sql: "SELECT /*+ MAX_EXECUTION_TIME({{ inputs.count_timeout_ms }}) */ COUNT(*)
      AS actual_rows FROM `{{ inputs.schema_name }}`.`{{ inputs.table_name }}`"
  - id: decision
    type: io.kestra.plugin.core.output.OutputValues
    values:
      eligible: "{{ outputs.count_rows.row.actual_rows >= inputs.min_rows and
        outputs.metadata.rows[0].stats_age_seconds >=
        inputs.min_stats_age_seconds and
        ((outputs.metadata.rows[0].estimated_rows -
        outputs.count_rows.row.actual_rows) | abs) * 1.0 >=
        outputs.count_rows.row.actual_rows * inputs.drift_ratio }}"
  - id: report
    type: io.kestra.plugin.core.log.Log
    message: "InnoDB estimate {{ outputs.metadata.rows[0].estimated_rows }}, exact
      count {{ outputs.count_rows.row.actual_rows }}, stats age {{
      outputs.metadata.rows[0].stats_age_seconds }} seconds. Eligible={{
      outputs.decision.values.eligible }}. One table only; ANALYZE LOCAL does
      not replicate the maintenance."
  - id: findings
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.decision.values.eligible }}"
    then:
      - id: notify_findings
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ inputs.slack_webhook_url }}"
        payload: "{{ {'text': 'Stale InnoDB row estimate after bulk load. Review
          execution ' ~ execution.id ~ ' before approving ANALYZE LOCAL TABLE.'}
          | toJson }}"
      - id: live_mode
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approval
            type: io.kestra.plugin.core.flow.Pause
            pauseDuration: PT1H
            behavior: FAIL
            description: Approve one local statistics refresh. READ locks briefly block
              writers. Row estimates remain sampled and this does not promise a
              faster query plan.
            onResume:
              - id: approved
                type: BOOL
                required: false
          - id: approved_only
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval.onResume.approved ?? false }}"
            then:
              - id: remediate
                type: io.kestra.plugin.jdbc.mysql.Queries
                transaction: false
                connectionPooling: false
                url: "{{ render(vars.jdbc_url) }}"
                username: "{{ inputs.mysql_username }}"
                password: "{{ inputs.mysql_password }}"
                fetchType: FETCH
                sql: |
                  LOCK TABLES `{{ inputs.schema_name }}`.`{{ inputs.table_name }}` READ;
                  SELECT IF((SELECT TABLE_ID FROM information_schema.INNODB_TABLES WHERE NAME='{{ inputs.schema_name }}/{{ inputs.table_name }}')={{ outputs.metadata.rows[0].table_id }}
                    AND (SELECT FILE_SIZE FROM information_schema.INNODB_TABLESPACES WHERE SPACE=(SELECT SPACE FROM information_schema.INNODB_TABLES WHERE NAME='{{ inputs.schema_name }}/{{ inputs.table_name }}')) <= {{ inputs.max_table_mb }}*1048576
                    AND (SELECT COUNT(*) FROM `{{ inputs.schema_name }}`.`{{ inputs.table_name }}`)={{ outputs.count_rows.row.actual_rows }},1,JSON_EXTRACT('stale approved table or row count','$.x')) AS exact_plan;
                  ANALYZE LOCAL TABLE `{{ inputs.schema_name }}`.`{{ inputs.table_name }}`;
                  UNLOCK TABLES;
              - id: require_analyze_status
                type: io.kestra.plugin.core.execution.Assert
                conditions:
                  - "{{ outputs.remediate.outputs | toJson | jq('[.[] | .rows[]?
                    | select(.Op == \"analyze\")] | length == 1 and all(.[];
                    .Msg_type == \"status\" and .Msg_text == \"OK\")') | first
                    }}"
                errorMessage: ANALYZE did not return one status OK result. Inspect the
                  maintenance output before retrying.
              - id: verify
                type: io.kestra.plugin.jdbc.mysql.Query
                connectionPooling: false
                url: "{{ render(vars.jdbc_url) }}"
                username: "{{ inputs.mysql_username }}"
                password: "{{ inputs.mysql_password }}"
                fetchType: FETCH_ONE
                sql: |
                  SELECT CAST(st.n_rows AS SIGNED) AS estimate,(SELECT COUNT(*) FROM `{{ inputs.schema_name }}`.`{{ inputs.table_name }}`) AS actual_rows,
                    CAST(t.TABLE_ID AS SIGNED) AS table_id FROM mysql.innodb_table_stats st JOIN information_schema.INNODB_TABLES t ON t.NAME=CONCAT(st.database_name,'/',st.table_name)
                  WHERE st.database_name='{{ inputs.schema_name }}' AND st.table_name='{{ inputs.table_name }}'
              - id: require_refresh
                type: io.kestra.plugin.core.execution.Assert
                conditions:
                  - "{{ outputs.verify.row.table_id ==
                    outputs.metadata.rows[0].table_id }}"
                  - "{{ outputs.verify.row.actual_rows ==
                    outputs.count_rows.row.actual_rows }}"
                  - "{{ ((outputs.verify.row.estimate -
                    outputs.verify.row.actual_rows) | abs) * 1.0 <=
                    outputs.verify.row.actual_rows * inputs.verify_tolerance }}"
                errorMessage: ANALYZE did not produce an acceptable estimate or the approved
                  table/data changed. Inspect maintenance output and replan.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: Approval declined. Statistics were not refreshed.
errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ inputs.slack_webhook_url }}"
    payload: "{{ {'text': 'InnoDB statistics refresh failed in execution ' ~
      execution.id ~ '. Recount the target before retrying.'} | toJson }}"
triggers:
  - id: daily_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

Safety

READ locks block writers during maintenance. Choose a quiet window and one known bulk-load target.

How it works

Native MySQL Query reads one file-per-table InnoDB target. Count rows within a time budget then compare persistent n_rows with the exact count. Eligibility requires the row floor, stats age and abs(estimate-count)/count >= drift_ratio. Missing statistics or unsupported targets fail before maintenance. Live mode pauses for explicit approval. MySQL Queries takes a READ lock, checks TABLE_ID, count and physical size again then runs ANALYZE LOCAL TABLE on the same connection. An error status or a fresh estimate outside verify_tolerance fails the run. The final count and TABLE_ID must still match approval. Connection pooling is off; maintenance uses no transaction wrapper because MySQL implicitly commits these statements.

What you get

A bounded statistics refresh after a loader missed its final ANALYZE. InnoDB samples pages, so the estimate is checked with relative tolerance. This flow changes statistics, not application rows. LOCAL suppresses binary logging of maintenance. It makes no query-speed or plan-improvement claim.

Who it's for

MySQL operators managing tables with persistent statistics after bulk loads or purges, especially when STATS_AUTO_RECALC=0 requires explicit maintenance.

Why orchestrate this with Kestra

Connect the loader's target to a bounded count, approval and verified maintenance result. Slack reports stale estimates and failed runs. The disabled daily trigger forces read-only inputs. All SQL uses native MySQL Query or Queries; no client script or stored procedure is installed.

Tested end to end

MySQL 8.0.46, Kestra 2.0.6 (MySQL plugin 2.3.0) and 2.0.5 (plugin 2.1.0). A second MySQL instance repeated recovery. Independent mysql reads preserved count, key sum, checksum and ordered data SHA-256. After loading 1,000 rows, n_rows changed from 2 to 1,002. EXPLAIN already estimated 1,002 before refresh.

Case	Execution	Result
Healthy	6q34wrazmW3P6l7Z2clW2j	SUCCESS
Dry run	iEdplWhRZpm6D8MSrYPNo	No write
Declined	5d4U39kdb3Gyp7Fk4K4zXR	No write
Refresh	2Pb2vF1ZQ6tCLEneNCfNLe	2 to 1,002
Repeat	1Fg1uqc6FhoQeA4InSsdfJ	No write
Row changed	7WdeXyRPsoxmzYoH5qHpfD	FAILED
Refused write	5juaTVEWM42maGcTqYzgzB	FAILED

Prerequisites

MySQL 8.0.24 or later, unpartitioned file-per-table InnoDB with persistent statistics. Tested grants: SELECT and INSERT on the target, LOCK TABLES on its database, PROCESS and SELECT on mysql.innodb_table_stats. INSERT is required by ANALYZE. Use a JDBC URL without credentials and configure TLS for your server.

Secrets

MYSQL_USERNAME, MYSQL_PASSWORD and SLACK_WEBHOOK_URL. Password and webhook overrides are SECRET inputs.

Inputs

schema_name/table_name allow 1 to 64 ASCII letters, digits or underscores; system schemas are refused. max_table_mb defaults to 16 MiB (1 to 128). min_rows defaults to 1,000; min_stats_age_seconds to 3,600. drift_ratio defaults to 0.5 (0.1 to 1); verify_tolerance to 0.25 (0.01 to 0.5). count_timeout_ms defaults to 10,000 (1 to 10,000). dry_run=true and auto_remediate=false are safe defaults.

Quick start

Configure secrets, mysql_url and the target; run safe mode.
Review the count, estimate, size and stats age.
Set dry_run=false and auto_remediate=true; resume with approved=true.

How to extend

Trigger from the loader that owns this table. Limits: one target, no partitions or histograms. Metadata and row lock waits are 2 seconds; connect timeout is 5 seconds and socket timeout 30 seconds. Socket timeout does not guarantee server-side ANALYZE cancellation. Approval expires after one hour. Queue scope is this flow only. Stats age cannot be reread under LOCK TABLES, so already-fresh stats may be refreshed. Equal-count data edits are allowed; table replacement is refused. Recount after any failure before retrying.

Refresh stale InnoDB row estimates after a bulk load

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Refresh stale InnoDB row estimates after a bulk load

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Build an ETL Pipeline from a REST API Using Python, Polars, and DuckDB

Load a scheduled Apify web scrape into Postgres

Extract data, transform it, and load it in parallel to S3 and Postgres

Arrow Flight SQL Endpoint TLS Compliance Gate

Automated API SLA and Data Drift Quarantine Pipeline

Apify Actor Run Failure Sentinel

New to Kestra?