Spark Small File Compaction Guard

id: spark-small-file-compaction-guard
namespace: company.team
description: Compact small-file Hive partitions with Spark through a staged swap
  that proves every row survived before the live path changes.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: spark_master
    type: STRING
    defaults: "{{ secret('SPARK_MASTER_URL') }}"
    description: Spark master URL, for example spark://master:7077 or yarn.
  - id: table_path
    type: STRING
    defaults: "{{ secret('LAKE_TABLE_PATH') }}"
    description: Root of one Hive-partitioned Parquet table, for example
      hdfs://nn:8020/lake/events.
  - id: partition_column
    type: STRING
    defaults: dt
    description: Name of the single partition column. Only directories named
      <column>=<value> are considered.
  - id: target_file_mb
    type: INT
    defaults: 128
    min: 1
    max: 2048
    description: Wanted size of one data file. It sets both the fragmentation
      threshold and the output file count.
  - id: max_avg_file_pct
    type: INT
    defaults: 25
    min: 1
    max: 99
    description: A partition counts as fragmented only when its average data file is
      smaller than this percentage of target_file_mb.
  - id: min_data_files
    type: INT
    defaults: 50
    min: 2
    max: 100000
    description: Minimum number of data files in a partition before compaction is
      worth a rewrite.
  - id: max_partition_mb
    type: INT
    defaults: 2048
    min: 1
    max: 1048576
    description: Bounded change. A partition larger than this is reported and never
      rewritten.
  - id: quiet_minutes
    type: INT
    defaults: 60
    min: 0
    max: 10080
    description: A partition whose newest file is younger than this is treated as
      still receiving writes and is spared.
  - id: max_partitions
    type: INT
    defaults: 5
    min: 1
    max: 100
    description: Largest number of partitions one approved run may rewrite. The rest
      are deferred to the next run.
  - id: protected_partitions
    type: ARRAY
    itemType: STRING
    defaults: []
    description: Partition directory names that must never be rewritten, for example
      dt=2026-10-15.
  - id: dry_run
    type: BOOL
    defaults: true
  - id: auto_remediate
    type: BOOL
    defaults: false
  - id: slack_webhook_url
    type: STRING
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"

variables:
  spark_conf:
    spark.executor.memory: 900m
    spark.executor.cores: "1"
    spark.cores.max: "1"
    spark.driver.memory: 900m
    spark.sql.shuffle.partitions: "4"
    spark.sql.sources.partitionOverwriteMode: static
    spark.hadoop.dfs.replication: "1"
  params: |
    {{ {
      'table': inputs.table_path,
      'partition_column': inputs.partition_column,
      'target_file_mb': inputs.target_file_mb,
      'max_avg_file_pct': inputs.max_avg_file_pct,
      'min_data_files': inputs.min_data_files,
      'max_partition_mb': inputs.max_partition_mb,
      'quiet_minutes': inputs.quiet_minutes,
      'max_partitions': inputs.max_partitions,
      'protected': inputs.protected_partitions,
      'dry_run': inputs.dry_run
    } | toJson }}
  lib: |
    import hashlib
    import json
    import math
    import time

    from pyspark.sql import SparkSession, functions as F

    MOD_A = 1000003
    MOD_B = 1000033
    STAGING = "_compaction/staging"
    BACKUP = "_compaction/backup"


    def session(app):
        spark = SparkSession.builder.appName(app).getOrCreate()
        spark.sparkContext.setLogLevel("WARN")
        return spark


    def emit(payload):
        print("::" + json.dumps({"outputs": payload}) + "::", flush=True)


    def stop(message):
        raise SystemExit("GUARD_REFUSED " + message)


    def hpath(spark, path):
        return spark.sparkContext._jvm.org.apache.hadoop.fs.Path(path)


    def filesystem(spark, path):
        hconf = spark.sparkContext._jsc.hadoopConfiguration()
        return hpath(spark, path).getFileSystem(hconf)


    def listing(spark, fs, path):
        p = hpath(spark, path)
        if not fs.exists(p):
            return None
        out = []
        for st in fs.listStatus(p):
            if st.isFile():
                out.append({
                    "name": st.getPath().getName(),
                    "len": int(st.getLen()),
                    "mtime": int(st.getModificationTime()),
                })
        return out


    def subdirs(spark, fs, path):
        p = hpath(spark, path)
        if not fs.exists(p):
            return []
        return sorted(st.getPath().getName() for st in fs.listStatus(p) if st.isDirectory())


    def files_digest(files):
        items = sorted("{0}:{1}:{2}".format(f["name"], f["len"], f["mtime"]) for f in files)
        return hashlib.sha256("|".join(items).encode("utf-8")).hexdigest()


    def metrics(df):
        """Row count plus two order independent checksums over every column as text."""
        cols = sorted(df.columns)
        key = F.xxhash64(*[F.coalesce(F.col(c).cast("string"), F.lit("@null@")) for c in cols])
        row = df.agg(
            F.count(F.lit(1)).alias("rows"),
            F.coalesce(F.sum(F.pmod(key, F.lit(MOD_A))), F.lit(0)).alias("ck_a"),
            F.coalesce(F.sum(F.pmod(key, F.lit(MOD_B))), F.lit(0)).alias("ck_b"),
        ).first()
        return {"rows": int(row["rows"]), "ck_a": int(row["ck_a"]), "ck_b": int(row["ck_b"])}


    def facts(spark, fs, path, with_metrics=True):
        """Layout and content facts of one partition directory. None when it is absent."""
        found = listing(spark, fs, path)
        if found is None:
            return None
        data = [f for f in found if f["name"].endswith(".parquet")]
        hidden = [f for f in found if f["name"].startswith("_") or f["name"].startswith(".")]
        unknown = [f["name"] for f in found if f not in data and f not in hidden]
        total = sum(f["len"] for f in data)
        newest = max([f["mtime"] for f in found], default=0)
        out = {
            "data_files": len(data),
            "other_files": len(found) - len(data),
            "total_bytes": total,
            "avg_bytes": total // max(1, len(data)),
            "newest_mtime": newest,
            "age_minutes": int((time.time() * 1000 - newest) // 60000) if newest else -1,
            "unknown_files": unknown,
            "files_digest": files_digest(found),
        }
        if with_metrics and data:
            out.update(metrics(spark.read.parquet(path)))
        else:
            out.update({"rows": 0, "ck_a": 0, "ck_b": 0})
        out["fingerprint"] = "{0}:{1}:{2}:{3}:{4}:{5}".format(
            out["data_files"], out["total_bytes"], out["newest_mtime"],
            out["rows"], out["ck_a"], out["ck_b"])
        return out


    def reconcile(spark, fs, table, apply_restore, apply_cleanup):
        """Finish or undo whatever an interrupted or refused earlier pass left behind."""
        actions = []
        backups = subdirs(spark, fs, table + "/" + BACKUP)
        staged = subdirs(spark, fs, table + "/" + STAGING)
        for name in backups:
            bpath = "{0}/{1}/{2}".format(table, BACKUP, name)
            live = "{0}/{1}".format(table, name)
            spath = "{0}/{1}/{2}".format(table, STAGING, name)
            back = facts(spark, fs, bpath)
            if back is None:
                continue
            current = facts(spark, fs, live)
            if current is None or current["data_files"] == 0:
                action = "restore_live_from_backup"
                if apply_restore:
                    if current is not None:
                        fs.delete(hpath(spark, live), True)
                    if not fs.rename(hpath(spark, bpath), hpath(spark, live)):
                        stop("could not restore " + name + " from its backup copy")
                    if fs.exists(hpath(spark, spath)):
                        fs.delete(hpath(spark, spath), True)
                    action = "restored_live_from_backup"
            elif (current["rows"], current["ck_a"], current["ck_b"]) == (
                    back["rows"], back["ck_a"], back["ck_b"]):
                action = "swap_committed_backup_removable"
                if apply_cleanup:
                    fs.delete(hpath(spark, bpath), True)
                    if fs.exists(hpath(spark, spath)):
                        fs.delete(hpath(spark, spath), True)
                    action = "swap_committed_backup_removed"
            else:
                action = "conflict"
            actions.append({
                "partition": name,
                "action": action,
                "live_rows": None if current is None else current["rows"],
                "backup_rows": back["rows"],
                "live_data_files": None if current is None else current["data_files"],
                "backup_data_files": back["data_files"],
            })
        for name in staged:
            if name in backups:
                continue
            live = "{0}/{1}".format(table, name)
            current = facts(spark, fs, live, with_metrics=False)
            if current is None or current["data_files"] == 0:
                actions.append({"partition": name, "action": "conflict",
                                "live_rows": None, "backup_rows": None,
                                "live_data_files": None, "backup_data_files": None})
                continue
            action = "orphan_staging_removable"
            if apply_cleanup:
                fs.delete(hpath(spark, "{0}/{1}/{2}".format(table, STAGING, name)), True)
                action = "orphan_staging_removed"
            actions.append({"partition": name, "action": action,
                            "live_rows": current["rows"], "backup_rows": None,
                            "live_data_files": current["data_files"], "backup_data_files": None})
        return actions


    def scan(spark, params, apply_restore, apply_cleanup):
        """One detector, used before and after the approval. It also classifies every spared partition."""
        table = params["table"].rstrip("/")
        fs = filesystem(spark, table)
        if not fs.exists(hpath(spark, table)):
            stop("table root " + table + " does not exist")
        target_bytes = int(params["target_file_mb"]) * 1024 * 1024
        max_avg = target_bytes * int(params["max_avg_file_pct"]) // 100
        quiet_ms = int(params["quiet_minutes"]) * 60000
        max_part_bytes = int(params["max_partition_mb"]) * 1024 * 1024
        protected = set(params["protected"] or [])
        prefix = params["partition_column"] + "="

        actions = reconcile(spark, fs, table, apply_restore, apply_cleanup)
        now_ms = time.time() * 1000

        candidates = []
        spared = []
        for name in subdirs(spark, fs, table):
            if not name.startswith(prefix):
                continue
            part = facts(spark, fs, "{0}/{1}".format(table, name))
            part["partition"] = name
            reasons = []
            if name in protected:
                reasons.append("protected")
            if part["unknown_files"]:
                reasons.append("unknown_files")
            if part["total_bytes"] == 0:
                reasons.append("empty")
            if part["data_files"] < int(params["min_data_files"]):
                reasons.append("few_data_files")
            if part["avg_bytes"] >= max_avg:
                reasons.append("average_file_size_ok")
            if part["total_bytes"] > max_part_bytes:
                reasons.append("partition_too_large")
            if quiet_ms > 0 and (now_ms - part["newest_mtime"]) < quiet_ms:
                reasons.append("active_writes")
            if reasons:
                part["reason"] = reasons[0]
                part["reasons"] = reasons
                spared.append(part)
            else:
                part["target_files"] = max(1, int(math.ceil(part["total_bytes"] / float(target_bytes))))
                candidates.append(part)

        candidates.sort(key=lambda r: (-r["data_files"], r["partition"]))
        limit = int(params["max_partitions"])
        plan = candidates[:limit]
        deferred = [r["partition"] for r in candidates[limit:]]
        quiet = [p for p in spared if p["reason"] != "active_writes"] + plan
        conflicts = [a for a in actions if a["action"] == "conflict"]
        return {
            "table": table,
            "plan": plan,
            "spared": spared,
            "deferred": deferred,
            "actions": actions,
            "vars": {
                "table": table,
                "partitions_seen": len(plan) + len(spared) + len(deferred),
                "plan_count": len(plan),
                "spared_count": len(spared),
                "deferred_count": len(deferred),
                "has_plan": "yes" if plan else "no",
                "plan_partitions_json": json.dumps([p["partition"] for p in plan]),
                "plan_json": json.dumps(plan),
                "spared_json": json.dumps([
                    {k: p[k] for k in ("partition", "reason", "reasons", "data_files", "total_bytes",
                                       "avg_bytes", "newest_mtime", "age_minutes", "rows", "ck_a",
                                       "ck_b", "files_digest")}
                    for p in spared]),
                "plan_fingerprint": "|".join(
                    "{0}={1}".format(p["partition"], p["fingerprint"]) for p in plan),
                "planned_data_files": sum(p["data_files"] for p in plan),
                "planned_bytes": sum(p["total_bytes"] for p in plan),
                "planned_rows": sum(p["rows"] for p in plan),
                "planned_target_files": sum(p["target_files"] for p in plan),
                "quiet_rows": sum(p["rows"] for p in quiet),
                "quiet_ck_a": sum(p["ck_a"] for p in quiet),
                "quiet_ck_b": sum(p["ck_b"] for p in quiet),
                "table_rows": sum(p["rows"] for p in plan + spared),
                "table_data_files": sum(p["data_files"] for p in plan + spared),
                "table_bytes": sum(p["total_bytes"] for p in plan + spared),
                "residue_actions": len(actions),
                "residue_json": json.dumps(actions),
                "residue_conflict": "yes" if conflicts else "no",
                "spark_version": spark.version,
                "report": " ".join(
                    ["PLAN " + (",".join("{0}[{1} files avg {2}B -> {3} files]".format(
                        p["partition"], p["data_files"], p["avg_bytes"], p["target_files"])
                        for p in plan) or "none")]
                    + ["SPARED " + (",".join("{0}[{1}]".format(p["partition"], p["reason"])
                                             for p in spared) or "none")]
                    + (["DEFERRED " + ",".join(deferred)] if deferred else [])
                    + (["RESIDUE " + ",".join("{0}[{1}]".format(a["partition"], a["action"])
                                              for a in actions)] if actions else [])),
            },
        }

tasks:
  - id: detect
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: scan_table
        type: io.kestra.plugin.spark.PythonSubmit
        master: "{{ inputs.spark_master }}"
        configurations: "{{ vars.spark_conf }}"
        timeout: PT15M
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: kw7
          user: root
        mainScript: |
          {{ render(vars.lib) }}

          PARAMS = json.loads(r'''{{ render(vars.params) }}''')

          spark = session("kestra-compaction-guard-detect")
          # A restore is pure recovery of a partition that is missing or empty, so it runs in every
          # mode. Removing a superseded backup is housekeeping and waits for a non dry run.
          found = scan(spark, PARAMS, apply_restore=True, apply_cleanup=not PARAMS["dry_run"])
          spark.stop()
          emit(found["vars"])

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: >-
      Table {{ outputs.scan_table.vars.table }} on Spark {{
      outputs.scan_table.vars.spark_version }}: {{
      outputs.scan_table.vars.partitions_seen }} partitions, {{
      outputs.scan_table.vars.table_data_files }} data files, {{
      outputs.scan_table.vars.table_rows }} rows. {{
      outputs.scan_table.vars.report }}

  - id: residue_conflict
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_table.vars.residue_conflict == 'yes' }}"
    then:
      - id: conflicting_residue
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          A leftover backup copy disagrees with the live partition, so an
          earlier pass cannot be reconciled automatically. Compare the live
          partition with its copy under _compaction/backup and resolve it by
          hand before running this flow again.

  - id: incident
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_table.vars.has_plan == 'yes' }}"
    then:
      - id: finding
        type: io.kestra.plugin.core.http.Request
        uri: "{{ inputs.slack_webhook_url }}"
        method: POST
        contentType: application/json
        body: >-
          {{ {'text': 'Small-file partitions found in ' ~
          outputs.scan_table.vars.table ~ '. '
              ~ outputs.scan_table.vars.report ~ ' dry_run=' ~ inputs.dry_run ~ ' auto_remediate='
              ~ inputs.auto_remediate ~ '. Execution ' ~ execution.id} | toJson }}
      - id: change_enabled
        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: PT30M
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Approve rewriting the planned partitions of this table only.
          - id: approved_change
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval.onResume.approved ?? false }}"
            then:
              - id: recheck
                type: io.kestra.plugin.core.flow.WorkingDirectory
                tasks:
                  - id: rescan_table
                    type: io.kestra.plugin.spark.PythonSubmit
                    master: "{{ inputs.spark_master }}"
                    configurations: "{{ vars.spark_conf }}"
                    timeout: PT15M
                    taskRunner:
                      type: io.kestra.plugin.scripts.runner.docker.Docker
                      networkMode: kw7
                      user: root
                    mainScript: |
                      {{ render(vars.lib) }}

                      PARAMS = json.loads(r'''{{ render(vars.params) }}''')
                      APPROVED = r'''{{ outputs.scan_table.vars.plan_fingerprint }}'''

                      spark = session("kestra-compaction-guard-recheck")
                      found = scan(spark, PARAMS, apply_restore=False, apply_cleanup=False)
                      spark.stop()
                      out = found["vars"]
                      out["stale"] = "no" if out["plan_fingerprint"] == APPROVED else "yes"
                      out["approved_fingerprint"] = APPROVED
                      emit(out)
              - id: unchanged_plan
                type: io.kestra.plugin.core.flow.If
                condition: >-
                  {{ outputs.rescan_table.vars.stale == 'no'
                     and outputs.rescan_table.vars.residue_conflict == 'no'
                     and outputs.rescan_table.vars.plan_count == outputs.scan_table.vars.plan_count
                     and outputs.rescan_table.vars.planned_rows == outputs.scan_table.vars.planned_rows
                     and outputs.rescan_table.vars.table == outputs.scan_table.vars.table }}
                then:
                  - id: compact_partitions
                    type: io.kestra.plugin.core.flow.Loop
                    values: "{{ fromJson(outputs.rescan_table.vars.plan_partitions_json) }}"
                    concurrencyLimit: 1
                    tasks:
                      - id: rewrite
                        type: io.kestra.plugin.core.flow.WorkingDirectory
                        tasks:
                          - id: stage_and_swap
                            type: io.kestra.plugin.spark.PythonSubmit
                            master: "{{ inputs.spark_master }}"
                            configurations: "{{ vars.spark_conf }}"
                            timeout: PT20M
                            taskRunner:
                              type: io.kestra.plugin.scripts.runner.docker.Docker
                              networkMode: kw7
                              user: root
                            mainScript: |
                              {{ render(vars.lib) }}

                              PARAMS = json.loads(r'''{{ render(vars.params) }}''')
                              PLAN = json.loads(r'''{{ outputs.rescan_table.vars.plan_json }}''')
                              PARTITION = r'''{{ item.value }}'''

                              table = PARAMS["table"].rstrip("/")
                              entry = None
                              for candidate in PLAN:
                                  if candidate["partition"] == PARTITION:
                                      entry = candidate
                              if entry is None:
                                  stop(PARTITION + " is not part of the approved plan")

                              live = "{0}/{1}".format(table, PARTITION)
                              stage = "{0}/{1}/{2}".format(table, STAGING, PARTITION)
                              backup = "{0}/{1}/{2}".format(table, BACKUP, PARTITION)

                              spark = session("kestra-compaction-guard-rewrite")
                              fs = filesystem(spark, table)

                              # Eligibility is read again here, immediately before the write, because
                              # the approval and the rescan are both older than this moment.
                              before = facts(spark, fs, live)
                              if before is None:
                                  stop(PARTITION + " disappeared after the plan was approved")
                              if before["fingerprint"] != entry["fingerprint"]:
                                  stop("{0} changed after the plan was approved: {1} is not {2}".format(
                                      PARTITION, before["fingerprint"], entry["fingerprint"]))
                              for path in (stage, backup):
                                  if fs.exists(hpath(spark, path)):
                                      fs.delete(hpath(spark, path), True)
                              fs.mkdirs(hpath(spark, "{0}/{1}".format(table, BACKUP)))

                              # Step 1: build the compacted copy beside the table. Nothing live moves yet,
                              # and _compaction is hidden from Spark's own partition discovery.
                              target = int(entry["target_files"])
                              try:
                                  spark.read.parquet(live).repartition(target) \
                                      .write.mode("overwrite").parquet(stage)
                              except Exception as exc:
                                  if fs.exists(hpath(spark, stage)):
                                      fs.delete(hpath(spark, stage), True)
                                  stop("the staged write for {0} was refused: {1}".format(
                                      PARTITION, str(exc)[:400]))

                              staged = facts(spark, fs, stage)
                              if staged is None or (staged["rows"], staged["ck_a"], staged["ck_b"]) != (
                                      before["rows"], before["ck_a"], before["ck_b"]):
                                  fs.delete(hpath(spark, stage), True)
                                  stop("the staged copy of {0} does not conserve the data".format(PARTITION))
                              if staged["data_files"] > target or staged["data_files"] >= before["data_files"]:
                                  fs.delete(hpath(spark, stage), True)
                                  stop("the staged copy of {0} is not less fragmented".format(PARTITION))

                              # Step 2: last read before the swap. A writer that touched the partition
                              # while the copy was built invalidates the copy.
                              again = facts(spark, fs, live, with_metrics=False)
                              if again["files_digest"] != before["files_digest"]:
                                  fs.delete(hpath(spark, stage), True)
                                  stop(PARTITION + " was written to while the compacted copy was built")

                              # Step 3: two directory renames. HDFS renames are atomic, so the live path
                              # always points at a complete partition.
                              if not fs.rename(hpath(spark, live), hpath(spark, backup)):
                                  fs.delete(hpath(spark, stage), True)
                                  stop("could not move the live copy of " + PARTITION + " aside")
                              if not fs.rename(hpath(spark, stage), hpath(spark, live)):
                                  fs.rename(hpath(spark, backup), hpath(spark, live))
                                  stop("could not publish " + PARTITION + ", the original is back in place")

                              after = facts(spark, fs, live)
                              conserved = (after["rows"], after["ck_a"], after["ck_b"]) == (
                                  before["rows"], before["ck_a"], before["ck_b"])
                              if not conserved or after["data_files"] > target:
                                  fs.delete(hpath(spark, live), True)
                                  fs.rename(hpath(spark, backup), hpath(spark, live))
                                  stop("the published copy of {0} failed verification, the original is back in place".format(PARTITION))
                              fs.delete(hpath(spark, backup), True)
                              spark.stop()
                              emit({
                                  "partition": PARTITION,
                                  "files_before": before["data_files"],
                                  "files_after": after["data_files"],
                                  "avg_before": before["avg_bytes"],
                                  "avg_after": after["avg_bytes"],
                                  "rows": after["rows"],
                                  "ck_a": after["ck_a"],
                                  "bytes_before": before["total_bytes"],
                                  "bytes_after": after["total_bytes"],
                                  "target_files": target,
                              })
                  - id: verify
                    type: io.kestra.plugin.core.flow.WorkingDirectory
                    tasks:
                      - id: audit_table
                        type: io.kestra.plugin.spark.PythonSubmit
                        master: "{{ inputs.spark_master }}"
                        configurations: "{{ vars.spark_conf }}"
                        timeout: PT15M
                        taskRunner:
                          type: io.kestra.plugin.scripts.runner.docker.Docker
                          networkMode: kw7
                          user: root
                        mainScript: |
                          {{ render(vars.lib) }}

                          PARAMS = json.loads(r'''{{ render(vars.params) }}''')
                          PLAN = json.loads(r'''{{ outputs.rescan_table.vars.plan_json }}''')
                          SPARED = json.loads(r'''{{ outputs.scan_table.vars.spared_json }}''')

                          table = PARAMS["table"].rstrip("/")
                          spark = session("kestra-compaction-guard-verify")
                          fs = filesystem(spark, table)
                          problems = []
                          lines = []

                          for entry in PLAN:
                              name = entry["partition"]
                              now = facts(spark, fs, "{0}/{1}".format(table, name))
                              if now is None:
                                  problems.append(name + " is gone")
                                  continue
                              if (now["rows"], now["ck_a"], now["ck_b"]) != (
                                      entry["rows"], entry["ck_a"], entry["ck_b"]):
                                  problems.append("{0} lost or changed data: {1}/{2} rows".format(
                                      name, now["rows"], entry["rows"]))
                              if now["data_files"] > int(entry["target_files"]):
                                  problems.append("{0} still holds {1} data files".format(
                                      name, now["data_files"]))
                              if now["data_files"] >= entry["data_files"]:
                                  problems.append(name + " was not compacted")
                              lines.append("{0}: {1} files -> {2} files, avg {3}B -> {4}B, {5} rows kept".format(
                                  name, entry["data_files"], now["data_files"],
                                  entry["avg_bytes"], now["avg_bytes"], now["rows"]))

                          for entry in SPARED:
                              name = entry["partition"]
                              now = facts(spark, fs, "{0}/{1}".format(table, name))
                              if now is None:
                                  problems.append("spared partition " + name + " is gone")
                                  continue
                              if entry["reason"] == "active_writes":
                                  # This partition was still receiving writes, so new files are allowed
                                  # while losing rows or files is not.
                                  if now["rows"] < entry["rows"] or now["data_files"] < entry["data_files"]:
                                      problems.append("spared partition " + name + " lost data")
                                  lines.append("{0}: spared ({1}), {2} rows now, {3} rows before".format(
                                      name, entry["reason"], now["rows"], entry["rows"]))
                              else:
                                  if now["files_digest"] != entry["files_digest"]:
                                      problems.append("spared partition " + name + " was modified")
                                  lines.append("{0}: spared ({1}), byte identical".format(name, entry["reason"]))

                          left = subdirs(spark, fs, table + "/" + BACKUP) + subdirs(spark, fs, table + "/" + STAGING)
                          if left:
                              problems.append("working directories left behind: " + ",".join(left))

                          current = scan(spark, PARAMS, apply_restore=False, apply_cleanup=False)
                          spark.stop()
                          out = current["vars"]
                          out["all_ok"] = "no" if problems else "yes"
                          out["problems_json"] = json.dumps(problems)
                          out["residue_dirs"] = len(left)
                          out["summary"] = " | ".join(lines)
                          emit(out)
                  - id: outcome
                    type: io.kestra.plugin.core.flow.If
                    condition: >-
                      {{ outputs.audit_table.vars.all_ok == 'yes'
                         and outputs.audit_table.vars.residue_dirs == 0
                         and outputs.audit_table.vars.quiet_rows == outputs.scan_table.vars.quiet_rows
                         and outputs.audit_table.vars.quiet_ck_a == outputs.scan_table.vars.quiet_ck_a
                         and outputs.audit_table.vars.quiet_ck_b == outputs.scan_table.vars.quiet_ck_b
                         and outputs.audit_table.vars.table_rows >= outputs.scan_table.vars.table_rows
                         and outputs.audit_table.vars.table_data_files < outputs.scan_table.vars.table_data_files }}
                    then:
                      - id: recovered
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Compaction verified. Table {{
                          outputs.audit_table.vars.table }} went from {{
                          outputs.scan_table.vars.table_data_files }} to {{
                          outputs.audit_table.vars.table_data_files }} data
                          files with {{ outputs.audit_table.vars.table_rows }}
                          rows and checksum {{
                          outputs.audit_table.vars.quiet_ck_a }} unchanged on
                          every quiet partition. {{
                          outputs.audit_table.vars.summary }}
                    else:
                      - id: unresolved
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: >-
                          Verification failed after the swap: {{
                          outputs.audit_table.vars.problems_json }}. Rows or
                          checksums moved, a partition is still fragmented, a
                          spared partition was touched or a working directory
                          was left behind. Inspect the table before retrying.
                else:
                  - id: stale_approval
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      The approval is stale. The partitions changed between the
                      plan and the approval, so nothing was written. Approved
                      plan {{ outputs.rescan_table.vars.approved_fingerprint }},
                      current plan {{ outputs.rescan_table.vars.plan_fingerprint
                      }}. Run the flow again to read the table and approve a
                      fresh plan.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: Approval was not granted. Every partition was left exactly as it was
                  found.
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              Read-only mode. {{ outputs.scan_table.vars.plan_count }}
              partitions holding {{ outputs.scan_table.vars.planned_data_files
              }} data files would be rewritten into {{
              outputs.scan_table.vars.planned_target_files }} files after
              approval. Nothing was written.

triggers:
  - id: nightly_layout_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "20 3 * * *"
    inputs:
      dry_run: true
      auto_remediate: false

errors:
  - id: alert_failure
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.slack_webhook_url }}"
    method: POST
    contentType: application/json
    body: >-
      {{ {'text': 'Spark small-file compaction guard failed on ' ~
      inputs.table_path ~ '. A refused
          write, a stale approval, a conflicting backup copy or a failed verification needs
          investigation. No partition is left half written. Execution ' ~ execution.id} | toJson }}