id: postgres-collation-drift-reindex
namespace: company.team
description: >-
After an OS or ICU upgrade changes PostgreSQL's text sort rules, rebuild every
index that uses a changed collation, after approval, then record the new
collation version and verify nothing is left.
inputs:
- id: jdbc_url
type: STRING
displayName: JDBC URL
description: The database to check. Collation versions are tracked per database,
so run the flow once per database.
defaults: jdbc:postgresql://localhost:5432/app
variables:
# Collations in use whose recorded version differs from the one the OS or ICU provides now.
# 100 is the OID of the database default collation, as stored in pg_index.indcollation.
drift_sql: |
SELECT 100::oid AS coll_oid, 'database' AS kind, d.datname::text AS name,
d.datcollversion AS recorded_version,
pg_database_collation_actual_version(d.oid) AS os_version,
format('ALTER DATABASE %I REFRESH COLLATION VERSION', d.datname) AS refresh_sql
FROM pg_database d
WHERE d.datname = current_database()
AND d.datcollversion IS DISTINCT FROM pg_database_collation_actual_version(d.oid)
UNION ALL
SELECT c.oid, 'collation', format('%I.%I', n.nspname, c.collname),
c.collversion, pg_collation_actual_version(c.oid),
format('ALTER COLLATION %I.%I REFRESH VERSION', n.nspname, c.collname)
FROM pg_collation c
JOIN pg_namespace n ON n.oid = c.collnamespace
WHERE c.collprovider <> 'd'
AND c.oid IN (SELECT attcollation FROM pg_attribute UNION SELECT unnest(indcollation::oid[]) FROM pg_index)
AND c.collversion IS DISTINCT FROM pg_collation_actual_version(c.oid)
health_sql: |
SELECT (SELECT count(*) FROM ({{ vars.drift_sql }}) d) AS drift_left,
count(*) AS invalid_indexes,
coalesce(string_agg(format('%I.%I', n.nspname, c.relname), ', '), '') AS invalid_names
FROM pg_index i
JOIN pg_class c ON c.oid = i.indexrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE NOT i.indisvalid
AND n.nspname NOT IN ('pg_catalog', 'information_schema') AND n.nspname !~ '^pg_toast'
tasks:
- id: drift
type: io.kestra.plugin.jdbc.postgresql.Query
description: Read-only. Compares recorded and current collation versions for the
database and every collation in use.
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: FETCH
sql: |
SELECT kind, name, recorded_version, os_version, refresh_sql
FROM ({{ vars.drift_sql }}) d
ORDER BY kind, name
- id: report
type: io.kestra.plugin.core.log.Log
message: >-
{% if outputs.drift.size == 0 %}No collation version mismatch in {{
inputs.jdbc_url }}.{% else %}Collation version mismatch in {{
inputs.jdbc_url }}: {% for r in outputs.drift.rows %}{{ r.kind }} {{
r.name }} was built with {{ r.recorded_version }}, the system now provides
{{ r.os_version }}. {% endfor %}{% endif %}
- id: stop_if_clean
type: io.kestra.plugin.core.execution.Exit
runIf: "{{ outputs.drift.size == 0 }}"
state: SUCCESS
- id: precheck
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "{{ render(vars.health_sql) }}"
- id: no_invalid_indexes
type: io.kestra.plugin.core.execution.Assert
description: An invalid index left by an earlier failed rebuild must be fixed
first, so it is never taken for a healthy one.
conditions:
- "{{ outputs.precheck.row.invalid_indexes == 0 }}"
errorMessage: "Invalid indexes found: {{ outputs.precheck.row.invalid_names }}.
A failed REINDEX CONCURRENTLY leaves a _ccnew copy: fix its cause, such as
duplicate keys, drop the invalid index, then run again."
- id: indexes
type: io.kestra.plugin.jdbc.postgresql.Query
description: Every index with a key that uses a changed collation. Partitioned
parent indexes are skipped, their partitions are listed.
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: FETCH
sql: |
SELECT format('%I.%I', n.nspname, ic.relname) AS index_name,
format('%I.%I', tn.nspname, tc.relname) AS table_name,
pg_size_pretty(pg_relation_size(i.indexrelid)) AS size,
format('REINDEX INDEX CONCURRENTLY %I.%I', n.nspname, ic.relname) AS reindex_sql
FROM pg_index i
JOIN pg_class ic ON ic.oid = i.indexrelid
JOIN pg_namespace n ON n.oid = ic.relnamespace
JOIN pg_class tc ON tc.oid = i.indrelid
JOIN pg_namespace tn ON tn.oid = tc.relnamespace
WHERE ic.relkind = 'i'
AND n.nspname NOT IN ('pg_catalog', 'information_schema') AND n.nspname !~ '^pg_toast'
AND i.indcollation::oid[] && ARRAY(SELECT coll_oid FROM ({{ vars.drift_sql }}) d)
ORDER BY pg_relation_size(i.indexrelid), 1
- id: plan
type: io.kestra.plugin.core.log.Log
message: >-
{{ outputs.indexes.size }} index(es) to rebuild: {% for r in
outputs.indexes.rows %}{{ r.index_name }} ({{ r.size }}){% if not
loop.last %}, {% endif %}{% endfor %}. Then: {% for r in
outputs.drift.rows %}{{ r.refresh_sql }}{% if not loop.last %}; {% endif
%}{% endfor %}.
- id: approve
type: io.kestra.plugin.core.flow.Pause
description: Resume to rebuild the indexes in the plan log. Without approval
within 24 hours, the run is cancelled and nothing changes.
pauseDuration: PT24H
behavior: CANCEL
- id: reindex
type: io.kestra.plugin.core.flow.Loop
description: One index at a time, without blocking writes. The first failure
stops the run before the version is refreshed.
runIf: "{{ outputs.indexes.size > 0 }}"
values: "{{ outputs.indexes.rows | toJson }}"
concurrencyLimit: 1
tasks:
- id: rebuild
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: NONE
sql: "{{ fromJson(item.value).reindex_sql }}"
- id: refresh
type: io.kestra.plugin.jdbc.postgresql.Queries
description: Records the new versions in one transaction, which also clears
PostgreSQL's warning. Runs only after every rebuild succeeded.
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: NONE
sql: |
{% for r in outputs.drift.rows %}
{{ r.refresh_sql }};
{% endfor %}
- id: verify
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ inputs.jdbc_url }}"
username: "{{ secret('POSTGRES_USERNAME') }}"
password: "{{ secret('POSTGRES_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "{{ render(vars.health_sql) }}"
- id: verified
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.verify.row.drift_left == 0 }}"
- "{{ outputs.verify.row.invalid_indexes == 0 }}"
errorMessage: "After the rebuild: {{ outputs.verify.row.drift_left }} collation
mismatch(es) and {{ outputs.verify.row.invalid_indexes }} invalid
index(es) left ({{ outputs.verify.row.invalid_names }})."
- id: summary
type: io.kestra.plugin.core.log.Log
message: "Rebuilt {{ outputs.indexes.size }} index(es) in {{ inputs.jdbc_url }}
and refreshed {{ outputs.drift.size }} collation version(s). No mismatch
or invalid index left."
triggers:
- id: after_upgrade
type: io.kestra.plugin.core.trigger.Webhook
description: Call this from the pipeline that upgrades the OS or the PostgreSQL image.
key: "{{ secret('POSTGRES_COLLATION_WEBHOOK_KEY') }}"
- id: weekly
type: io.kestra.plugin.core.trigger.Schedule
description: Catches upgrades made outside the pipeline. Shipped disabled.
cron: "0 6 * * 1"
disabled: true
