crates/paimon/src/catalog/filesystem.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Filesystem catalog implementation for Apache Paimon.
//!
//! Reference: [org.apache.paimon.catalog.FileSystemCatalog](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java)

use std::collections::HashMap;

use crate::catalog::{Catalog, Database, Identifier, DB_LOCATION_PROP, DB_SUFFIX};
use crate::common::{CatalogOptions, Options};
use crate::error::{ConfigInvalidSnafu, Error, Result};
use crate::io::FileIO;
use crate::spec::{Schema, TableSchema};
use crate::table::{SchemaManager, Table};
use async_trait::async_trait;
use bytes::Bytes;
use opendal::raw::get_basename;
use snafu::OptionExt;

/// Name of the schema directory under each table path.
const SCHEMA_DIR: &str = "schema";
/// Prefix for schema files (under the schema directory).
const SCHEMA_PREFIX: &str = "schema-";

fn make_path(parent: &str, child: &str) -> String {
    format!("{parent}/{child}")
}

/// Filesystem catalog implementation.
///
/// This catalog stores metadata on a filesystem with the following structure:
/// ```text
/// warehouse/
///   ├── database1.db/
///   │   ├── table1/
///   │   │   └── schema/
///   │   │       ├── schema-0
///   │   │       ├── schema-1
///   │   │       └── ...
///   │   └── table2/
///   │       └── schema/
///   │           ├── schema-0
///   │           └── ...
///   └── database2.db/
///       └── ...
/// ```
///
/// Reference: [org.apache.paimon.catalog.FileSystemCatalog](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java)
#[derive(Clone, Debug)]
pub struct FileSystemCatalog {
    file_io: FileIO,
    warehouse: String,
}

impl FileSystemCatalog {
    /// Create a new filesystem catalog from configuration options.
    ///
    /// # Arguments
    /// * `options` - Configuration options containing warehouse path and storage configs (S3, OSS, etc.)
    ///
    /// # Required Options
    /// * `warehouse` - The root warehouse path (e.g., `/path/to/warehouse`, `s3://bucket/warehouse`)
    ///
    /// # Example
    /// ```ignore
    /// use paimon::{FileSystemCatalog, Options, CatalogOptions};
    ///
    /// // Local filesystem
    /// let mut options = Options::new();
    /// options.set(CatalogOptions::WAREHOUSE, "/tmp/warehouse");
    /// let catalog = FileSystemCatalog::new(options)?;
    ///
    /// // S3 with credentials
    /// let mut options = Options::new();
    /// options.set(CatalogOptions::WAREHOUSE, "s3://bucket/warehouse");
    /// options.set("s3.access-key-id", "...");
    /// options.set("s3.secret-access-key", "...");
    /// let catalog = FileSystemCatalog::new(options)?;
    /// ```
    pub fn new(options: Options) -> crate::Result<Self> {
        let warehouse =
            options
                .get(CatalogOptions::WAREHOUSE)
                .cloned()
                .context(ConfigInvalidSnafu {
                    message: format!("Missing required option: {}", CatalogOptions::WAREHOUSE),
                })?;

        let file_io = FileIO::from_path(&warehouse)?
            .with_props(options.to_map().iter())
            .build()?;

        Ok(Self { file_io, warehouse })
    }

    /// Get the warehouse path.
    pub fn warehouse(&self) -> &str {
        &self.warehouse
    }

    /// Get the FileIO instance.
    pub fn file_io(&self) -> &FileIO {
        &self.file_io
    }

    /// Get the path for a database (warehouse / `name` + [DB_SUFFIX]).
    fn database_path(&self, database_name: &str) -> String {
        make_path(
            &self.warehouse,
            format!("{database_name}{DB_SUFFIX}").as_str(),
        )
    }

    /// Get the path for a table (warehouse / `database.db` / table).
    fn table_path(&self, identifier: &Identifier) -> String {
        make_path(
            &self.database_path(identifier.database()),
            identifier.object(),
        )
    }

    /// Path to the schema directory under a table path.
    fn schema_dir_path(&self, table_path: &str) -> String {
        make_path(table_path, SCHEMA_DIR)
    }

    /// Get the schema file path for a given version (table_path/schema/schema-{version}).
    fn schema_file_path(&self, table_path: &str, schema_id: i64) -> String {
        make_path(
            make_path(table_path, SCHEMA_DIR).as_str(),
            format!("{SCHEMA_PREFIX}{schema_id}").as_str(),
        )
    }

    /// List directories in the given path.
    async fn list_directories(&self, path: &str) -> Result<Vec<String>> {
        let statuses = self.file_io.list_status(path).await?;
        let mut dirs = Vec::new();
        for status in statuses {
            if status.is_dir {
                if let Some(p) = get_basename(status.path.as_str())
                    // opendal get_basename will contain "/" for directory,
                    // we need to strip suffix to get the real base name
                    .strip_suffix("/")
                {
                    dirs.push(p.to_string());
                }
            }
        }
        Ok(dirs)
    }

    /// Load the latest schema for a table (highest schema-{version} file under table_path/schema).
    async fn load_latest_table_schema(&self, table_path: &str) -> Result<Option<TableSchema>> {
        let manager = SchemaManager::new(self.file_io.clone(), table_path.to_string());
        Ok(manager.latest().await?.map(|arc| (*arc).clone()))
    }

    /// Save a table schema to a file.
    async fn save_table_schema(&self, table_path: &str, schema: &TableSchema) -> Result<()> {
        let schema_dir = self.schema_dir_path(table_path);
        self.file_io.mkdirs(&schema_dir).await?;
        let schema_path = self.schema_file_path(table_path, schema.id());
        let output_file = self.file_io.new_output(&schema_path)?;
        let content =
            Bytes::from(
                serde_json::to_string(schema).map_err(|e| Error::DataInvalid {
                    message: format!("Failed to serialize schema: {e}"),
                    source: Some(Box::new(e)),
                })?,
            );
        output_file.write(content).await?;
        Ok(())
    }

    /// Check if a database exists.
    async fn database_exists(&self, name: &str) -> Result<bool> {
        self.file_io.exists(&self.database_path(name)).await
    }

    /// Check if a table exists.
    async fn table_exists(&self, identifier: &Identifier) -> Result<bool> {
        self.file_io.exists(&self.table_path(identifier)).await
    }
}

#[async_trait]
impl Catalog for FileSystemCatalog {
    async fn list_databases(&self) -> Result<Vec<String>> {
        let dirs = self.list_directories(&self.warehouse).await?;
        Ok(dirs
            .into_iter()
            .filter_map(|name| name.strip_suffix(DB_SUFFIX).map(|s| s.to_string()))
            .collect())
    }

    async fn create_database(
        &self,
        name: &str,
        ignore_if_exists: bool,
        properties: HashMap<String, String>,
    ) -> Result<()> {
        Identifier::validate_database_name(name)?;

        if properties.contains_key(DB_LOCATION_PROP) {
            return Err(Error::ConfigInvalid {
                message: "Cannot specify location for a database when using fileSystem catalog."
                    .to_string(),
            });
        }

        let path = self.database_path(name);
        let database_exists = self.database_exists(name).await?;

        if !ignore_if_exists && database_exists {
            return Err(Error::DatabaseAlreadyExist {
                database: name.to_string(),
            });
        }

        if !database_exists {
            self.file_io.mkdirs(&path).await?;
        }

        Ok(())
    }

    async fn get_database(&self, name: &str) -> Result<Database> {
        Identifier::validate_database_name(name)?;

        if !self.database_exists(name).await? {
            return Err(Error::DatabaseNotExist {
                database: name.to_string(),
            });
        }

        Ok(Database::new(name.to_string(), HashMap::new(), None))
    }

    async fn drop_database(
        &self,
        name: &str,
        ignore_if_not_exists: bool,
        cascade: bool,
    ) -> Result<()> {
        Identifier::validate_database_name(name)?;

        let path = self.database_path(name);

        let database_exists = self.database_exists(name).await?;
        if !ignore_if_not_exists && !database_exists {
            return Err(Error::DatabaseNotExist {
                database: name.to_string(),
            });
        }

        if !database_exists {
            return Ok(());
        }

        let tables = self.list_directories(&path).await?;
        if !tables.is_empty() && !cascade {
            return Err(Error::DatabaseNotEmpty {
                database: name.to_string(),
            });
        }

        self.file_io.delete_dir(&path).await?;
        Ok(())
    }

    async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
        identifier.validate()?;

        let table_path = self.table_path(identifier);

        if !self.table_exists(identifier).await? {
            return Err(Error::TableNotExist {
                full_name: identifier.full_name(),
            });
        }

        let schema = self
            .load_latest_table_schema(&table_path)
            .await?
            .ok_or_else(|| Error::TableNotExist {
                full_name: identifier.full_name(),
            })?;

        Ok(Table::new(
            self.file_io.clone(),
            identifier.clone(),
            table_path,
            schema,
            None,
        ))
    }

    async fn list_tables(&self, database_name: &str) -> Result<Vec<String>> {
        Identifier::validate_database_name(database_name)?;

        let path = self.database_path(database_name);

        if !self.database_exists(database_name).await? {
            return Err(Error::DatabaseNotExist {
                database: database_name.to_string(),
            });
        }

        self.list_directories(&path).await
    }

    async fn create_table(
        &self,
        identifier: &Identifier,
        creation: Schema,
        ignore_if_exists: bool,
    ) -> Result<()> {
        identifier.validate()?;

        let table_path = self.table_path(identifier);

        let table_exists = self.table_exists(identifier).await?;

        if !ignore_if_exists && table_exists {
            return Err(Error::TableAlreadyExist {
                full_name: identifier.full_name(),
            });
        }

        if !self.database_exists(identifier.database()).await? {
            return Err(Error::DatabaseNotExist {
                database: identifier.database().to_string(),
            });
        }

        // todo: consider with lock
        if !table_exists {
            self.file_io.mkdirs(&table_path).await?;
            let table_schema = TableSchema::new(0, &creation);
            self.save_table_schema(&table_path, &table_schema).await?;
        }

        Ok(())
    }

    async fn drop_table(&self, identifier: &Identifier, ignore_if_not_exists: bool) -> Result<()> {
        identifier.validate()?;

        let table_path = self.table_path(identifier);

        let table_exists = self.table_exists(identifier).await?;

        if !ignore_if_not_exists && !table_exists {
            return Err(Error::TableNotExist {
                full_name: identifier.full_name(),
            });
        }

        if !table_exists {
            return Ok(());
        }

        self.file_io.delete_dir(&table_path).await?;
        Ok(())
    }

    async fn rename_table(
        &self,
        from: &Identifier,
        to: &Identifier,
        ignore_if_not_exists: bool,
    ) -> Result<()> {
        from.validate()?;
        to.validate()?;

        let from_path = self.table_path(from);
        let to_path = self.table_path(to);

        let table_exists = self.table_exists(from).await?;

        if !ignore_if_not_exists && !table_exists {
            return Err(Error::TableNotExist {
                full_name: from.full_name(),
            });
        }

        if !table_exists {
            return Ok(());
        }

        if self.table_exists(to).await? {
            return Err(Error::TableAlreadyExist {
                full_name: to.full_name(),
            });
        }

        self.file_io.rename(&from_path, &to_path).await?;
        Ok(())
    }

    async fn alter_table(
        &self,
        identifier: &Identifier,
        changes: Vec<crate::spec::SchemaChange>,
        ignore_if_not_exists: bool,
    ) -> Result<()> {
        identifier.validate()?;

        let table_path = self.table_path(identifier);
        if !self.table_exists(identifier).await? {
            if ignore_if_not_exists {
                return Ok(());
            }
            return Err(Error::TableNotExist {
                full_name: identifier.full_name(),
            });
        }

        let current = self
            .load_latest_table_schema(&table_path)
            .await?
            .ok_or_else(|| Error::TableNotExist {
                full_name: identifier.full_name(),
            })?;

        let new_schema = current
            .apply_changes(changes)
            .map_err(|e| fill_table_name(e, identifier))?;
        self.save_table_schema(&table_path, &new_schema).await
    }
}

/// `TableSchema::apply_changes` returns column errors without a table name;
/// fill in the identifier's full name so the message identifies the table.
fn fill_table_name(err: Error, identifier: &Identifier) -> Error {
    match err {
        Error::ColumnNotExist { column, .. } => Error::ColumnNotExist {
            full_name: identifier.full_name(),
            column,
        },
        Error::ColumnAlreadyExist { column, .. } => Error::ColumnAlreadyExist {
            full_name: identifier.full_name(),
            column,
        },
        other => other,
    }
}

#[cfg(test)]
// Skip on Windows: these tests list directories, and opendal's `fs` lister
// panics (`StripPrefixError`) when listing under a drive-rooted path with the
// `root="/"` operator setup. Single-file ops work after the drive-letter fix;
// directory listing needs the opendal root rework tracked in #397.
#[cfg(not(windows))]
mod tests {
    use super::*;
    use tempfile::TempDir;

    /// Returns a temp dir guard (keep alive for test duration) and a catalog using it as warehouse.
    fn create_test_catalog() -> (TempDir, FileSystemCatalog) {
        let temp_dir = TempDir::new().unwrap();
        let warehouse = temp_dir.path().to_str().unwrap().to_string();
        let mut options = Options::new();
        options.set(CatalogOptions::WAREHOUSE, warehouse);
        let catalog = FileSystemCatalog::new(options).unwrap();
        (temp_dir, catalog)
    }

    fn testing_schema() -> Schema {
        Schema::builder()
            .column(
                "id",
                crate::spec::DataType::Int(crate::spec::IntType::new()),
            )
            .build()
            .unwrap()
    }

    #[tokio::test]
    async fn test_database_operations() {
        let (_temp_dir, catalog) = create_test_catalog();

        // create and list
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        catalog
            .create_database("db2", false, HashMap::new())
            .await
            .unwrap();
        let databases = catalog.list_databases().await.unwrap();
        assert_eq!(databases.len(), 2);
        assert!(databases.contains(&"db1".to_string()));
        assert!(databases.contains(&"db2".to_string()));

        // create same db without ignore_if_exists -> error
        let result = catalog.create_database("db1", false, HashMap::new()).await;
        assert!(result.is_err());
        assert!(matches!(result, Err(Error::DatabaseAlreadyExist { .. })));

        // create same db with ignore_if_exists -> ok
        catalog
            .create_database("db1", true, HashMap::new())
            .await
            .unwrap();

        // create_database with location property -> error (filesystem catalog)
        let mut props = HashMap::new();
        props.insert(DB_LOCATION_PROP.to_string(), "/some/path".to_string());
        let result = catalog.create_database("dbx", false, props).await;
        assert!(result.is_err());
        assert!(matches!(result, Err(Error::ConfigInvalid { .. })));

        // drop empty database
        catalog.drop_database("db1", false, false).await.unwrap();
        let databases = catalog.list_databases().await.unwrap();
        assert_eq!(databases.len(), 1);
        assert!(databases.contains(&"db2".to_string()));

        // drop non-empty database without cascade -> error
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        catalog
            .create_table(&Identifier::new("db1", "table1"), testing_schema(), false)
            .await
            .unwrap();
        let result = catalog.drop_database("db1", false, false).await;
        assert!(result.is_err());
        assert!(matches!(result, Err(Error::DatabaseNotEmpty { .. })));

        // drop database with cascade
        catalog.drop_database("db1", false, true).await.unwrap();
        assert!(!catalog.database_exists("db1").await.unwrap());
    }

    #[tokio::test]
    async fn test_create_database_should_reject_path_traversal_name() {
        let (temp_dir, catalog) = create_test_catalog();
        let escaped_path = temp_dir.path().join("escaped.db");

        let result = catalog
            .create_database("../escaped", false, HashMap::new())
            .await;

        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
        assert!(!escaped_path.exists());
    }

    #[tokio::test]
    async fn test_drop_database_should_reject_path_traversal_name() {
        let (temp_dir, catalog) = create_test_catalog();
        let escaped_path = temp_dir.path().join("escaped.db");
        std::fs::create_dir(&escaped_path).unwrap();

        let result = catalog.drop_database("../escaped", false, true).await;

        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
        assert!(escaped_path.exists());
    }

    #[tokio::test]
    async fn test_table_operations() {
        let (_temp_dir, catalog) = create_test_catalog();
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();

        // create and list tables
        let schema = testing_schema();
        catalog
            .create_table(&Identifier::new("db1", "table1"), schema.clone(), false)
            .await
            .unwrap();
        catalog
            .create_table(&Identifier::new("db1", "table2"), schema, false)
            .await
            .unwrap();
        let tables = catalog.list_tables("db1").await.unwrap();
        assert_eq!(tables.len(), 2);
        assert!(tables.contains(&"table1".to_string()));
        assert!(tables.contains(&"table2".to_string()));

        // get_table and check schema
        let schema_with_name = Schema::builder()
            .column(
                "id",
                crate::spec::DataType::Int(crate::spec::IntType::new()),
            )
            .column(
                "name",
                crate::spec::DataType::VarChar(crate::spec::VarCharType::string_type()),
            )
            .build()
            .unwrap();
        catalog
            .create_table(&Identifier::new("db1", "table3"), schema_with_name, false)
            .await
            .unwrap();
        let table = catalog
            .get_table(&Identifier::new("db1", "table3"))
            .await
            .unwrap();
        let table_schema = table.schema();
        assert_eq!(table_schema.id(), 0);
        assert_eq!(table_schema.fields().len(), 2);

        // drop table
        catalog
            .drop_table(&Identifier::new("db1", "table1"), false)
            .await
            .unwrap();
        let tables = catalog.list_tables("db1").await.unwrap();
        assert_eq!(tables.len(), 2);
        assert!(!tables.contains(&"table1".to_string()));
    }

    #[tokio::test]
    async fn test_create_table_should_reject_path_traversal_object_name() {
        let (temp_dir, catalog) = create_test_catalog();
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        let escaped_path = temp_dir.path().join("table_escape");

        let result = catalog
            .create_table(
                &Identifier::new("db1", "../../table_escape"),
                testing_schema(),
                false,
            )
            .await;

        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
        assert!(!escaped_path.exists());
    }

    #[tokio::test]
    async fn test_rename_table_should_reject_path_traversal_target_name() {
        let (temp_dir, catalog) = create_test_catalog();
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        let source = Identifier::new("db1", "source");
        catalog
            .create_table(&source, testing_schema(), false)
            .await
            .unwrap();
        let escaped_path = temp_dir.path().join("renamed_escape");

        let result = catalog
            .rename_table(
                &source,
                &Identifier::new("db1", "../../renamed_escape"),
                false,
            )
            .await;

        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
        assert!(!escaped_path.exists());
        assert!(catalog.get_table(&source).await.is_ok());
    }

    #[tokio::test]
    async fn test_list_partitions_default_table_not_found_errors() {
        let (_temp_dir, catalog) = create_test_catalog();
        let id = Identifier::new("nope_db", "nope_table");
        let result = catalog.list_partitions(&id).await;
        assert!(
            matches!(
                result,
                Err(Error::DatabaseNotExist { .. } | Error::TableNotExist { .. })
            ),
            "expected TableNotExist/DatabaseNotExist, got {result:?}"
        );
    }

    #[tokio::test]
    async fn test_list_partitions_default_empty_table_returns_empty() {
        let (_temp_dir, catalog) = create_test_catalog();
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        let id = Identifier::new("db1", "t1");
        catalog
            .create_table(&id, testing_schema(), false)
            .await
            .unwrap();
        let parts = catalog.list_partitions(&id).await.unwrap();
        assert!(
            parts.is_empty(),
            "table without snapshots should yield no partitions"
        );
    }

    /// Mirrors Java `CatalogTestBase.testListPartitionsPaged`: the default impl
    /// returns the same full result regardless of `max_results` / `page_token`.
    #[tokio::test]
    async fn test_list_partitions_paged_default_ignores_max_and_token() {
        let (_temp_dir, catalog) = create_test_catalog();
        catalog
            .create_database("db1", false, HashMap::new())
            .await
            .unwrap();
        let id = Identifier::new("db1", "t1");
        catalog
            .create_table(&id, testing_schema(), false)
            .await
            .unwrap();
        for (max_results, page_token) in [
            (None, None),
            (Some(2), None),
            (Some(2), Some("dt=20250101")),
            (Some(8), None),
            (Some(8), Some("dt=20250101")),
        ] {
            let page = catalog
                .list_partitions_paged(&id, max_results, page_token)
                .await
                .unwrap();
            assert!(
                page.elements.is_empty(),
                "empty table → empty page for max_results={max_results:?}, page_token={page_token:?}"
            );
            assert!(
                page.next_page_token.is_none(),
                "default impl never paginates"
            );
        }
    }

    use crate::spec::{
        ColumnMove, DataField, DataType, IntType, RowType, SchemaChange, VarCharType,
    };

    /// Two-column table (id INT, name VARCHAR) used by the alter-table tests.
    fn two_column_schema() -> Schema {
        Schema::builder()
            .column("id", DataType::Int(IntType::new()))
            .column("name", DataType::VarChar(VarCharType::string_type()))
            .build()
            .unwrap()
    }

    async fn setup_table(catalog: &FileSystemCatalog, schema: Schema) -> Identifier {
        catalog
            .create_database("db", false, HashMap::new())
            .await
            .unwrap();
        let id = Identifier::new("db", "t");
        catalog.create_table(&id, schema, false).await.unwrap();
        id
    }

    #[tokio::test]
    async fn test_alter_table_column_changes() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, two_column_schema()).await;

        // Add a column at the end; it must take highest_field_id + 1.
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::add_column(
                    "age".to_string(),
                    DataType::Int(IntType::new()),
                )],
                false,
            )
            .await
            .unwrap();
        let ts = catalog.get_table(&id).await.unwrap();
        let ts = ts.schema();
        let names: Vec<&str> = ts.fields().iter().map(|f| f.name()).collect();
        assert_eq!(names, vec!["id", "name", "age"]);
        let age = ts.fields().iter().find(|f| f.name() == "age").unwrap();
        assert_eq!(age.id(), 2, "new column gets highest_field_id + 1");
        assert_eq!(ts.id(), 1, "schema version bumped");

        // Add a column moved to the front.
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::add_column_with_description_and_column_move(
                    "rowkey".to_string(),
                    DataType::Int(IntType::new()),
                    "primary".to_string(),
                    ColumnMove::move_first("rowkey".to_string()),
                )],
                false,
            )
            .await
            .unwrap();
        let ts = catalog.get_table(&id).await.unwrap();
        let ts = ts.schema();
        assert_eq!(ts.fields()[0].name(), "rowkey");
        assert_eq!(ts.fields()[0].description(), Some("primary"));

        // Converting nullable `id` to NOT NULL below is rejected by default;
        // allow it explicitly. The flag is read from the pre-alter options, so
        // it must be set in a separate alter.
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::set_option(
                    "alter-column-null-to-not-null.disabled".to_string(),
                    "false".to_string(),
                )],
                false,
            )
            .await
            .unwrap();

        // Rename, update comment, update type, update nullability, drop.
        catalog
            .alter_table(
                &id,
                vec![
                    SchemaChange::rename_column("name".to_string(), "full_name".to_string()),
                    SchemaChange::update_column_comment("id".to_string(), "the id".to_string()),
                    SchemaChange::update_column_type(
                        "age".to_string(),
                        DataType::BigInt(crate::spec::BigIntType::new()),
                    ),
                    SchemaChange::update_column_nullability("id".to_string(), false),
                    SchemaChange::drop_column("rowkey".to_string()),
                ],
                false,
            )
            .await
            .unwrap();
        let ts = catalog.get_table(&id).await.unwrap();
        let ts = ts.schema();
        let names: Vec<&str> = ts.fields().iter().map(|f| f.name()).collect();
        assert_eq!(names, vec!["id", "full_name", "age"]);
        let id_field = ts.fields().iter().find(|f| f.name() == "id").unwrap();
        assert_eq!(id_field.description(), Some("the id"));
        assert!(!id_field.data_type().is_nullable());
        let age_field = ts.fields().iter().find(|f| f.name() == "age").unwrap();
        assert!(matches!(age_field.data_type(), DataType::BigInt(_)));
    }

    #[tokio::test]
    async fn test_alter_table_reposition_column() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, two_column_schema()).await;

        // Move `name` before `id`.
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::update_column_position(
                    ColumnMove::move_first("name".to_string()),
                )],
                false,
            )
            .await
            .unwrap();
        let ts = catalog.get_table(&id).await.unwrap();
        let names: Vec<&str> = ts.schema().fields().iter().map(|f| f.name()).collect();
        assert_eq!(names, vec!["name", "id"]);
    }

    #[tokio::test]
    async fn test_alter_table_errors() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, two_column_schema()).await;

        // Add a duplicate column -> ColumnAlreadyExist.
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::add_column(
                    "name".to_string(),
                    DataType::Int(IntType::new()),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::ColumnAlreadyExist { .. }),
            "got {err:?}"
        );

        // Drop a missing column -> ColumnNotExist.
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::drop_column("ghost".to_string())],
                false,
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ColumnNotExist { .. }), "got {err:?}");

        // Altering a missing table: ignored vs error.
        let missing = Identifier::new("db", "nope");
        catalog
            .alter_table(
                &missing,
                vec![SchemaChange::update_column_comment(
                    "id".to_string(),
                    "x".to_string(),
                )],
                true,
            )
            .await
            .unwrap();
        let err = catalog
            .alter_table(
                &missing,
                vec![SchemaChange::update_column_comment(
                    "id".to_string(),
                    "x".to_string(),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::TableNotExist { .. }), "got {err:?}");
    }

    #[tokio::test]
    async fn test_alter_table_drop_primary_key_column_rejected() {
        let (_tmp, catalog) = create_test_catalog();
        let schema = Schema::builder()
            .column("id", DataType::Int(IntType::new()))
            .column("name", DataType::VarChar(VarCharType::string_type()))
            .primary_key(["id"])
            .build()
            .unwrap();
        let id = setup_table(&catalog, schema).await;

        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::drop_column("id".to_string())],
                false,
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");
    }

    #[tokio::test]
    async fn test_alter_table_add_not_null_column_rejected() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, two_column_schema()).await;

        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::add_column(
                    "age".to_string(),
                    DataType::Int(IntType::with_nullable(false)),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("cannot specify NOT NULL"),
            "got {err:?}"
        );
    }

    /// Collect the IDs of all fields nested inside row types.
    fn nested_row_field_ids(data_type: &DataType, ids: &mut Vec<i32>) {
        if let DataType::Row(row) = data_type {
            for f in row.fields() {
                ids.push(f.id());
                nested_row_field_ids(f.data_type(), ids);
            }
        }
    }

    #[tokio::test]
    async fn test_alter_table_add_nested_column_reassigns_field_ids() {
        let (_tmp, catalog) = create_test_catalog();
        // Existing columns take IDs 0 and 1.
        let id = setup_table(&catalog, two_column_schema()).await;

        // Nested field IDs as requested by the caller deliberately collide
        // with the existing columns; they must all be reassigned.
        let nested = DataType::Row(RowType::new(vec![
            DataField::new(0, "a".to_string(), DataType::Int(IntType::new())),
            DataField::new(
                1,
                "b".to_string(),
                DataType::Row(RowType::new(vec![DataField::new(
                    2,
                    "c".to_string(),
                    DataType::Int(IntType::new()),
                )])),
            ),
        ]));
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::add_column("s".to_string(), nested)],
                false,
            )
            .await
            .unwrap();

        let table = catalog.get_table(&id).await.unwrap();
        let ts = table.schema();
        let s = ts.fields().iter().find(|f| f.name() == "s").unwrap();
        assert_eq!(s.id(), 2, "top-level column takes highest_field_id + 1");
        let mut ids = Vec::new();
        nested_row_field_ids(s.data_type(), &mut ids);
        ids.sort_unstable();
        assert_eq!(ids, vec![3, 4, 5], "nested IDs are fresh and unique");
        assert_eq!(ts.highest_field_id(), 5);
    }

    #[tokio::test]
    async fn test_alter_table_drop_all_columns_rejected() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, two_column_schema()).await;

        let err = catalog
            .alter_table(
                &id,
                vec![
                    SchemaChange::drop_column("id".to_string()),
                    SchemaChange::drop_column("name".to_string()),
                ],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("Cannot drop all fields"),
            "got {err:?}"
        );
    }

    /// Partitioned primary-key table: dt VARCHAR (partition), id INT, v INT,
    /// primary key (dt, id).
    fn partitioned_pk_schema() -> Schema {
        Schema::builder()
            .column("dt", DataType::VarChar(VarCharType::string_type()))
            .column("id", DataType::Int(IntType::new()))
            .column("v", DataType::Int(IntType::new()))
            .partition_keys(["dt"])
            .primary_key(["dt", "id"])
            .build()
            .unwrap()
    }

    #[tokio::test]
    async fn test_alter_table_key_column_guards() {
        let (_tmp, catalog) = create_test_catalog();
        let id = setup_table(&catalog, partitioned_pk_schema()).await;

        // Renaming a partition column is rejected.
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::rename_column(
                    "dt".to_string(),
                    "day".to_string(),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("Cannot rename partition column"),
            "got {err:?}"
        );

        // Updating the type of a partition column is rejected.
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::update_column_type(
                    "dt".to_string(),
                    DataType::VarChar(VarCharType::string_type()),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("Cannot update partition column"),
            "got {err:?}"
        );

        // Updating the type of a primary key column is rejected.
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::update_column_type(
                    "id".to_string(),
                    DataType::BigInt(crate::spec::BigIntType::new()),
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("Cannot update primary key"),
            "got {err:?}"
        );

        // Making a primary key column nullable is rejected ...
        let err = catalog
            .alter_table(
                &id,
                vec![SchemaChange::update_column_nullability(
                    "id".to_string(),
                    true,
                )],
                false,
            )
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("Cannot change nullability of primary key"),
            "got {err:?}"
        );

        // ... while NOT NULL stays allowed (it is already non-nullable).
        catalog
            .alter_table(
                &id,
                vec![SchemaChange::update_column_nullability(
                    "id".to_string(),
                    false,
                )],
                false,
            )
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_alter_table_rename_primary_key_column_propagates() {
        let (_tmp, catalog) = create_test_catalog();
        let schema = Schema::builder()
            .column("id", DataType::Int(IntType::new()))
            .column("name", DataType::VarChar(VarCharType::string_type()))
            .column("v", DataType::Int(IntType::new()))
            .primary_key(["id"])
            .option("bucket-key", "id")
            .option("sequence.field", "v")
            .build()
            .unwrap();
        let id = setup_table(&catalog, schema).await;

        catalog
            .alter_table(
                &id,
                vec![
                    SchemaChange::rename_column("id".to_string(), "user_id".to_string()),
                    SchemaChange::rename_column("v".to_string(), "ver".to_string()),
                ],
                false,
            )
            .await
            .unwrap();

        let table = catalog.get_table(&id).await.unwrap();
        let ts = table.schema();
        assert_eq!(ts.primary_keys(), ["user_id".to_string()]);
        assert_eq!(
            ts.options().get("bucket-key").map(String::as_str),
            Some("user_id")
        );
        assert_eq!(
            ts.options().get("sequence.field").map(String::as_str),
            Some("ver")
        );
    }
}