micromegas_analytics/lakehouse/

migration.rs

use crate::arrow_utils::parse_parquet_metadata;
use crate::lakehouse::log_stats_view::{log_stats_ddl_text, log_stats_view_definition};
use anyhow::{Context, Result};
use micromegas_ingestion::remote_data_lake::acquire_lock;
use micromegas_tracing::prelude::*;
use sqlx::Executor;
use sqlx::Row;

pub const LATEST_LAKEHOUSE_SCHEMA_VERSION: i32 = 10;

async fn read_lakehouse_schema_version(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> i32 {
    match sqlx::query(
        "SELECT version
         FROM lakehouse_migration;",
    )
    .fetch_one(&mut **tr)
    .await
    {
        Ok(row) => row.get("version"),
        Err(e) => {
            info!(
                "Error reading data lake schema version, assuming version 0: {}",
                e
            );
            0
        }
    }
}

/// Migrates the lakehouse schema to the latest version.
pub async fn migrate_lakehouse(pool: sqlx::Pool<sqlx::Postgres>) -> Result<()> {
    let mut tr = pool.begin().await?;
    let mut current_version = read_lakehouse_schema_version(&mut tr).await;
    drop(tr);
    info!("current lakehouse schema: {}", current_version);
    if current_version != LATEST_LAKEHOUSE_SCHEMA_VERSION {
        let mut tr = pool.begin().await?;
        acquire_lock(&mut tr, 1).await?;
        current_version = read_lakehouse_schema_version(&mut pool.begin().await?).await;
        if LATEST_LAKEHOUSE_SCHEMA_VERSION == current_version {
            return Ok(());
        }
        if let Err(e) = execute_lakehouse_migration(pool.clone()).await {
            error!("Error migrating database: {}", e);
            return Err(e);
        }
        current_version = read_lakehouse_schema_version(&mut tr).await;
    }
    assert_eq!(current_version, LATEST_LAKEHOUSE_SCHEMA_VERSION);
    Ok(())
}

/// Executes the lakehouse migration steps.
async fn execute_lakehouse_migration(pool: sqlx::Pool<sqlx::Postgres>) -> Result<()> {
    let mut current_version = read_lakehouse_schema_version(&mut pool.begin().await?).await;
    if 0 == current_version {
        info!("creating v1 lakehouse_schema");
        let mut tr = pool.begin().await?;
        create_tables(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 1 == current_version {
        info!("upgrade lakehouse schema to v2");
        let mut tr = pool.begin().await?;
        upgrade_v1_to_v2(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 2 == current_version {
        info!("upgrade lakehouse schema to v3");
        let mut tr = pool.begin().await?;
        upgrade_v2_to_v3(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 3 == current_version {
        info!("upgrade lakehouse schema to v4");
        let mut tr = pool.begin().await?;
        upgrade_v3_to_v4(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 4 == current_version {
        info!("upgrade lakehouse schema to v5");
        let mut tr = pool.begin().await?;
        upgrade_v4_to_v5(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 5 == current_version {
        info!("upgrade lakehouse schema to v6");
        let mut tr = pool.begin().await?;
        upgrade_v5_to_v6(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 6 == current_version {
        info!("upgrade lakehouse schema to v7");
        let mut tr = pool.begin().await?;
        upgrade_v6_to_v7(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 7 == current_version {
        info!("upgrade lakehouse schema to v8");
        let mut tr = pool.begin().await?;
        upgrade_v7_to_v8(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 8 == current_version {
        info!("upgrade lakehouse schema to v9");
        let mut tr = pool.begin().await?;
        upgrade_v8_to_v9(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    if 9 == current_version {
        info!("upgrade lakehouse schema to v10");
        let mut tr = pool.begin().await?;
        upgrade_v9_to_v10(&mut tr).await?;
        current_version = read_lakehouse_schema_version(&mut tr).await;
        tr.commit().await?;
    }
    assert_eq!(current_version, LATEST_LAKEHOUSE_SCHEMA_VERSION);
    Ok(())
}

async fn create_partitions_table(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Every table in the set shares the same schema - that schema can change over time, in which case the partition has to be rebuilt.
    // The instance id is a unique key to the table within the table set.
    //    * Example 1: if there is a process_log table for each process, the instance id could be the process_id.
    //      It would not clash with an instance of process_metrics table for the same process.
    //    * Example 2:  if there is a table instance for each metric, the view_instance_id could be the name of the metric.

    // source_data_hash can be used to detect that the partition is out of date (sha1 of the block_ids, for example)
    tr.execute("
         CREATE TABLE lakehouse_partitions(
                  view_set_name VARCHAR(255),
                  view_instance_id VARCHAR(255),
                  begin_insert_time TIMESTAMPTZ,
                  end_insert_time TIMESTAMPTZ,
                  min_event_time TIMESTAMPTZ,
                  max_event_time TIMESTAMPTZ,
                  updated TIMESTAMPTZ,
                  file_path VARCHAR(2047),
                  file_size BIGINT,
                  file_schema_hash bytea,
                  source_data_hash bytea
                  );
         CREATE INDEX lh_part_begin_insert on lakehouse_partitions(view_set_name, view_instance_id, begin_insert_time);
         CREATE INDEX lh_part_end_insert on lakehouse_partitions(view_set_name, view_instance_id, end_insert_time);
         CREATE INDEX lh_part_min_time on lakehouse_partitions(view_set_name, view_instance_id, min_event_time);
         CREATE INDEX lh_part_max_time on lakehouse_partitions(view_set_name, view_instance_id, max_event_time);
")
        .await
        .with_context(|| "Creating table blocks and its indices")?;
    Ok(())
}

async fn create_temp_files_table(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // partitions that are out of date can still be referenced until they expire
    tr.execute(
        "
         CREATE TABLE temporary_files(
                  file_path VARCHAR(2047),
                  file_size BIGINT,
                  expiration TIMESTAMPTZ );
         CREATE INDEX temporary_files_expiration on temporary_files(expiration);
",
    )
    .await
    .with_context(|| "Creating temporary_files table")?;
    Ok(())
}

async fn create_migration_table(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    sqlx::query("CREATE table lakehouse_migration(version integer);")
        .execute(&mut **tr)
        .await
        .with_context(|| "Creating table lakehouse_migration")?;
    sqlx::query("INSERT INTO lakehouse_migration VALUES(1);")
        .execute(&mut **tr)
        .await
        .with_context(|| "Recording the initial lakehouse schema version")?;
    Ok(())
}

async fn create_tables(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    create_partitions_table(tr).await?;
    create_temp_files_table(tr).await?;
    create_migration_table(tr).await?;
    Ok(())
}

async fn upgrade_v1_to_v2(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // file_metadata is meant to be the serialized ParquetMetaData
    // which can be found in the footer of the file
    tr.execute("ALTER TABLE lakehouse_partitions ADD file_metadata bytea;")
        .await
        .with_context(|| "adding column file_metadata to lakehouse_partitions table")?;
    tr.execute("UPDATE lakehouse_migration SET version=2;")
        .await
        .with_context(|| "Updating lakehouse schema version to 2")?;
    Ok(())
}

async fn upgrade_v2_to_v3(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Add num_rows column to store row count separately from file_metadata
    tr.execute("ALTER TABLE lakehouse_partitions ADD num_rows BIGINT;")
        .await
        .with_context(|| "adding column num_rows to lakehouse_partitions table")?;

    // Add index on file_path for efficient on-demand metadata loading
    tr.execute("CREATE INDEX lakehouse_partitions_file_path ON lakehouse_partitions(file_path);")
        .await
        .with_context(|| "creating index on file_path")?;

    // Populate num_rows column for existing partitions
    populate_num_rows_column(tr)
        .await
        .with_context(|| "populating num_rows column")?;

    // Make num_rows NOT NULL after populating existing data
    tr.execute("ALTER TABLE lakehouse_partitions ALTER COLUMN num_rows SET NOT NULL;")
        .await
        .with_context(|| "setting num_rows column to NOT NULL")?;

    tr.execute("UPDATE lakehouse_migration SET version=3;")
        .await
        .with_context(|| "Updating lakehouse schema version to 3")?;
    Ok(())
}

async fn populate_num_rows_column(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    info!("populating num_rows column for existing partitions");

    let mut total_count = 0;
    let batch_size = 1000;

    loop {
        // Fetch partitions in batches to avoid loading all metadata into memory
        let rows = sqlx::query("SELECT file_path, file_metadata FROM lakehouse_partitions WHERE file_metadata IS NOT NULL AND num_rows IS NULL LIMIT $1")
            .bind(batch_size)
            .fetch_all(&mut **tr)
            .await?;

        if rows.is_empty() {
            break;
        }

        let mut batch_count = 0;
        for row in rows {
            let file_path: String = row.try_get("file_path")?;
            let file_metadata_buffer: Vec<u8> = row.try_get("file_metadata")?;

            // Parse metadata only for this partition
            match parse_parquet_metadata(&file_metadata_buffer.into()) {
                Ok(file_metadata) => {
                    let num_rows = file_metadata.file_metadata().num_rows();

                    // Update just this partition
                    if let Err(e) = sqlx::query(
                        "UPDATE lakehouse_partitions SET num_rows = $1 WHERE file_path = $2",
                    )
                    .bind(num_rows)
                    .bind(&file_path)
                    .execute(&mut **tr)
                    .await
                    {
                        error!(
                            "failed to update num_rows for partition {}: {}",
                            file_path, e
                        );
                        continue;
                    }

                    batch_count += 1;
                }
                Err(e) => {
                    error!(
                        "failed to parse metadata for partition {}: {}",
                        file_path, e
                    );
                    // For partitions with unparseable metadata, set num_rows to 0 as a fallback
                    if let Err(e2) = sqlx::query(
                        "UPDATE lakehouse_partitions SET num_rows = 0 WHERE file_path = $1",
                    )
                    .bind(&file_path)
                    .execute(&mut **tr)
                    .await
                    {
                        error!(
                            "failed to set fallback num_rows for partition {}: {}",
                            file_path, e2
                        );
                    }
                }
            }
        }

        total_count += batch_count;
        info!(
            "populated num_rows for {} partitions (total: {})",
            batch_count, total_count
        );
    }

    info!("populated num_rows for {} total partitions", total_count);
    Ok(())
}

async fn upgrade_v3_to_v4(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Create dedicated partition_metadata table
    tr.execute(
        "CREATE TABLE partition_metadata(
            file_path VARCHAR(2047) PRIMARY KEY,
            metadata bytea NOT NULL,
            insert_time TIMESTAMPTZ NOT NULL
        );",
    )
    .await
    .with_context(|| "creating partition_metadata table")?;

    // Migrate existing metadata from lakehouse_partitions to partition_metadata
    migrate_metadata_to_new_table(tr)
        .await
        .with_context(|| "migrating metadata to partition_metadata table")?;

    // Drop the file_metadata column from lakehouse_partitions after successful migration
    tr.execute("ALTER TABLE lakehouse_partitions DROP COLUMN file_metadata;")
        .await
        .with_context(|| "dropping file_metadata column from lakehouse_partitions")?;

    tr.execute("UPDATE lakehouse_migration SET version=4;")
        .await
        .with_context(|| "Updating lakehouse schema version to 4")?;
    Ok(())
}

async fn migrate_metadata_to_new_table(
    tr: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    info!("migrating metadata to partition_metadata table");

    // First, get all file paths that have metadata (small data)
    let file_paths: Vec<String> = sqlx::query_scalar(
        "SELECT file_path 
         FROM lakehouse_partitions 
         WHERE file_metadata IS NOT NULL
         ORDER BY file_path",
    )
    .fetch_all(&mut **tr)
    .await?;

    let total_to_migrate = file_paths.len();
    info!(
        "found {} partitions with metadata to migrate",
        total_to_migrate
    );

    let mut total_count = 0;
    let batch_size = 10; // Small batch size since metadata can be large

    // Process in batches to avoid loading too much metadata at once
    for chunk in file_paths.chunks(batch_size) {
        // Build a query to fetch just this batch
        let placeholders: Vec<String> = (1..=chunk.len()).map(|i| format!("${}", i)).collect();
        let query_str = format!(
            "SELECT file_path, file_metadata, updated 
             FROM lakehouse_partitions 
             WHERE file_path IN ({})",
            placeholders.join(", ")
        );

        let mut query = sqlx::query(&query_str);
        for path in chunk {
            query = query.bind(path);
        }

        let rows = query.fetch_all(&mut **tr).await?;

        for row in rows {
            let file_path: String = row.try_get("file_path")?;
            let file_metadata: Vec<u8> = row.try_get("file_metadata")?;
            let updated: chrono::DateTime<chrono::Utc> = row.try_get("updated")?;

            // Insert into new partition_metadata table (with ON CONFLICT for migration safety)
            if let Err(e) = sqlx::query(
                "INSERT INTO partition_metadata (file_path, metadata, insert_time) 
                 VALUES ($1, $2, $3)
                 ON CONFLICT (file_path) DO NOTHING",
            )
            .bind(&file_path)
            .bind(&file_metadata)
            .bind(updated)
            .execute(&mut **tr)
            .await
            {
                error!(
                    "failed to migrate metadata for partition {}: {}",
                    file_path, e
                );
                continue;
            }

            total_count += 1;
        }

        if total_count % 100 == 0 || total_count == total_to_migrate {
            info!(
                "migrated {}/{} partition metadata entries",
                total_count, total_to_migrate
            );
        }
    }

    info!("migrated metadata for {} total partitions", total_count);
    Ok(())
}

async fn upgrade_v4_to_v5(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Add partition_format_version column to lakehouse_partitions
    // Default to 1 for existing partitions (Arrow 56.0 format)
    tr.execute(
        "ALTER TABLE lakehouse_partitions
         ADD COLUMN partition_format_version INTEGER NOT NULL DEFAULT 1;",
    )
    .await
    .with_context(|| "adding partition_format_version to lakehouse_partitions")?;
    // Add partition_format_version column to partition_metadata
    // Default to 1 for existing metadata (Arrow 56.0 format)
    tr.execute(
        "ALTER TABLE partition_metadata
         ADD COLUMN partition_format_version INTEGER NOT NULL DEFAULT 1;",
    )
    .await
    .with_context(|| "adding partition_format_version to partition_metadata")?;
    tr.execute("UPDATE lakehouse_migration SET version=5;")
        .await
        .with_context(|| "Updating lakehouse schema version to 5")?;
    info!("added partition_format_version columns to both tables");
    Ok(())
}

async fn upgrade_v5_to_v6(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    tr.execute("DROP TABLE partition_metadata;")
        .await
        .with_context(|| "dropping partition_metadata table")?;
    tr.execute("UPDATE lakehouse_migration SET version=6;")
        .await
        .with_context(|| "Updating lakehouse schema version to 6")?;
    Ok(())
}

async fn upgrade_v6_to_v7(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Nullable, no backfill needed: NULL means "no ordering guarantee", which is automatically
    // correct for every partition written before this column existed.
    tr.execute("ALTER TABLE lakehouse_partitions ADD COLUMN sort_order TEXT[];")
        .await
        .with_context(|| "adding sort_order column to lakehouse_partitions")?;
    tr.execute("CREATE EXTENSION IF NOT EXISTS btree_gist;")
        .await
        .with_context(|| "creating btree_gist extension (required by the partition overlap exclusion constraint); on PostgreSQL <= 12, or if this role lacks CREATE on the database, a superuser must run CREATE EXTENSION btree_gist once")?;
    // Detect-then-fail: the exclusion constraint below cannot be added NOT VALID, so if legacy
    // overlapping partitions exist, surface exactly which rows conflict instead of failing with a
    // raw constraint error. Zero-width ranges are excluded to match tstzrange semantics (an empty
    // range never overlaps anything). Scoped to same-schema pairs, matching the constraint: a
    // cross-schema overlap is the legal schema-rollout coexistence state, not a conflict.
    let conflicts = sqlx::query(
        "SELECT a.view_set_name, a.view_instance_id,
                a.file_path AS file_path_a, b.file_path AS file_path_b,
                a.begin_insert_time AS begin_a, a.end_insert_time AS end_a,
                b.begin_insert_time AS begin_b, b.end_insert_time AS end_b
         FROM lakehouse_partitions a
         JOIN lakehouse_partitions b
           ON a.view_set_name = b.view_set_name
          AND a.view_instance_id = b.view_instance_id
          AND a.file_schema_hash = b.file_schema_hash
          AND a.ctid < b.ctid
          AND a.begin_insert_time < b.end_insert_time
          AND a.end_insert_time > b.begin_insert_time
          AND a.begin_insert_time < a.end_insert_time
          AND b.begin_insert_time < b.end_insert_time
         LIMIT 20;",
    )
    .fetch_all(&mut **tr)
    .await
    .with_context(|| "detecting overlapping partitions before adding exclusion constraint")?;
    if !conflicts.is_empty() {
        let mut msg = String::from(
            "lakehouse_partitions contains partitions with overlapping insert-time ranges; \
             retire them (e.g. retire_partition_by_metadata) and restart to complete the \
             migration. Conflicting pairs (first 20):",
        );
        for row in &conflicts {
            let view_set_name: String = row.try_get("view_set_name")?;
            let view_instance_id: String = row.try_get("view_instance_id")?;
            let file_path_a: Option<String> = row.try_get("file_path_a")?;
            let file_path_b: Option<String> = row.try_get("file_path_b")?;
            let begin_a: chrono::DateTime<chrono::Utc> = row.try_get("begin_a")?;
            let end_a: chrono::DateTime<chrono::Utc> = row.try_get("end_a")?;
            let begin_b: chrono::DateTime<chrono::Utc> = row.try_get("begin_b")?;
            let end_b: chrono::DateTime<chrono::Utc> = row.try_get("end_b")?;
            msg.push_str(&format!(
                "\n  {view_set_name}/{view_instance_id}: {file_path_a:?} [{begin_a}, {end_a}] overlaps {file_path_b:?} [{begin_b}, {end_b}]"
            ));
        }
        anyhow::bail!(msg);
    }
    // Enforced by the index even across concurrent transactions (the second conflicting writer
    // blocks, then errors) -- closes the race a SELECT-based check under READ COMMITTED cannot:
    // two concurrent writers whose snapshots don't see each other's uncommitted partitions.
    // tstzrange is '[)' so adjacent partitions sharing a boundary do not conflict, and the write
    // path's retire+insert runs in one transaction so replacing a partition never self-conflicts.
    // Scoped by file_schema_hash (btree_gist supports bytea equality): after a schema-hash bump,
    // old-schema partitions legally coexist with overlapping new-schema writes until
    // retire_incompatible_partitions cleans them up -- e.g. an old-schema daily merged partition
    // vs a new hourly write that overlaps it without containing it, which the insert path's
    // containment-based retire cannot remove. Queries filter partitions by schema hash, so
    // cross-schema overlap never yields duplicate rows; only same-schema overlap is a bug.
    tr.execute(
        "ALTER TABLE lakehouse_partitions ADD CONSTRAINT lakehouse_partitions_no_overlap
         EXCLUDE USING gist (
             view_set_name WITH =,
             view_instance_id WITH =,
             file_schema_hash WITH =,
             tstzrange(begin_insert_time, end_insert_time) WITH &&
         );",
    )
    .await
    .with_context(|| "adding partition overlap exclusion constraint")?;
    tr.execute("UPDATE lakehouse_migration SET version=7;")
        .await
        .with_context(|| "Updating lakehouse schema version to 7")?;
    Ok(())
}

async fn upgrade_v7_to_v8(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Nullable, no backfill: NULL means "not recorded" -- either the partition was written
    // before this column existed, or its view never declares a `Concatenated` event-time
    // ordering. This is the same nullable/no-backfill shape as upgrade_v6_to_v7's sort_order
    // column; unlike that migration, this one adds no constraint.
    tr.execute("ALTER TABLE lakehouse_partitions ADD COLUMN max_sort_key_time TIMESTAMPTZ;")
        .await
        .with_context(|| "adding max_sort_key_time column to lakehouse_partitions")?;
    tr.execute("UPDATE lakehouse_migration SET version=8;")
        .await
        .with_context(|| "Updating lakehouse schema version to 8")?;
    Ok(())
}

async fn upgrade_v8_to_v9(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // Admin-managed query deny list. One `match_expr` column,
    // not a column per matcher. No index: the table holds at most
    // `MICROMEGAS_QUERY_DENY_MAX_RULES` rows and every replica reads all of it on each refresh
    // tick. No `SCHEMA_VERSION` (partition file-schema) change -- no partition content is affected.
    tr.execute(
        "CREATE TABLE query_deny_list(
             rule_id      UUID PRIMARY KEY,
             created_at   TIMESTAMPTZ  NOT NULL,
             created_by   VARCHAR(255) NOT NULL,
             reason       TEXT         NOT NULL,
             match_expr   TEXT         NOT NULL,
             last_hit_at  TIMESTAMPTZ
         );",
    )
    .await
    .with_context(|| "creating query_deny_list table")?;
    tr.execute("UPDATE lakehouse_migration SET version=9;")
        .await
        .with_context(|| "Updating lakehouse schema version to 9")?;
    Ok(())
}

async fn upgrade_v9_to_v10(tr: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
    // DDL-defined eagerly materialized views: persistence for `CREATE [OR REPLACE]
    // MATERIALIZED VIEW` / `DROP MATERIALIZED VIEW`. `view_options` is TEXT (serde_json), not
    // JSONB -- the column is always read back as a whole and round-tripped, so JSONB's indexing
    // buys nothing. No SCHEMA_VERSION (partition file-schema) change -- no partition content is
    // affected by this migration.
    tr.execute(
        "CREATE TABLE lakehouse_view_set_definitions(
             view_set_name          VARCHAR(255) PRIMARY KEY,
             definition_sql         TEXT NOT NULL,
             extract_query          TEXT NOT NULL,
             count_src_query        TEXT NOT NULL,
             merge_partitions_query TEXT NOT NULL,
             update_group           INTEGER NOT NULL,
             view_options           TEXT NOT NULL DEFAULT '{}',
             created_at             TIMESTAMPTZ NOT NULL DEFAULT now(),
             updated_at             TIMESTAMPTZ NOT NULL DEFAULT now(),
             updated_by             TEXT
         );",
    )
    .await
    .with_context(|| "creating lakehouse_view_set_definitions table")?;

    // `log_stats` becomes data: its only distinction from a DDL-defined view was that its SQL
    // lived in Rust. Seeded here, from the same `ViewDefinition` `view_factory.rs` used to build
    // it in code (see `log_stats_view.rs`), so this is a pure data move -- its `file_schema_hash`
    // cannot change, and existing partitions survive the upgrade unaffected.
    // `ON CONFLICT DO NOTHING` keeps this idempotent and freezes the seeded text at this version:
    // a later release changing `log_stats`'s shipped SQL must ship its own migration step that
    // upserts the row, not just edit `log_stats_view.rs`'s consts.
    let def = log_stats_view_definition();
    let view_options = serde_json::to_string(&def.options)
        .with_context(|| "serializing log_stats's seeded view_options")?;
    sqlx::query(
        "INSERT INTO lakehouse_view_set_definitions
             (view_set_name, definition_sql, extract_query, count_src_query,
              merge_partitions_query, update_group, view_options, updated_by)
         VALUES ($1, $2, $3, $4, $5, $6, $7, 'migration')
         ON CONFLICT (view_set_name) DO NOTHING;",
    )
    .bind(&def.view_set_name)
    .bind(log_stats_ddl_text())
    .bind(&def.extract_query)
    .bind(&def.count_src_query)
    .bind(&def.merge_partitions_query)
    .bind(def.update_group)
    .bind(view_options)
    .execute(&mut **tr)
    .await
    .with_context(|| "seeding log_stats into lakehouse_view_set_definitions")?;

    tr.execute("UPDATE lakehouse_migration SET version=10;")
        .await
        .with_context(|| "Updating lakehouse schema version to 10")?;
    Ok(())
}