CALL catalog_name.system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1);
CALL catalog_name.system.procedure_name(arg_1, arg_2, ... arg_n);

SparkSession spark = SparkSession  
    .builder()  
    .master("local")  
    .appName("Iceberg spark example")  
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")  
    .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")  
    .config("spark.sql.catalog.local.type", "hadoop") // 指定catalog类型  
    .config("spark.sql.catalog.local.warehouse", "iceberg_warehouse")  
    .getOrCreate();  

spark.sql("CALL local.system.rollback_to_snapshot('iceberg_db.table2', 3285133177610707025)");

spark.sql("CALL catalog_name.system.rewrite_data_files(table => 'db.sample', options => map('min-input-files','2'))");

-- spark sql  
-- 更改字段类型  
ALTER TABLE prod.db.sample ALTER COLUMN measurement TYPE double;  

-- 更新字段和 comment  
ALTER TABLE prod.db.sample ALTER COLUMN measurement TYPE double COMMENT 'unit is bytes per second'  

-- 更改字段顺序， FIRST/AFTER  
ALTER TABLE prod.db.sample ALTER COLUMN col FIRST  
ALTER TABLE prod.db.sample ALTER COLUMN nested.col AFTER other_col  

-- null 更改，如果该字段是主键则不支持  
ALTER TABLE prod.db.sample ALTER COLUMN id DROP NOT NULL  
ALTER TABLE prod.db.sample ALTER COLUMN id SET NOT NULL  

// 添加一个顶级列  
table.UpdateSchema()  
    .addColumn("toplevel", Types.DecimalType.of(9, 2))  
    .commit()  

// 添加一个列到嵌套类型里面  
table.UpdateSchema()  
    .addColumn("points", "z", Types.LongType.get(), "z axis")  
    .commit()  

// 更新列类型  
table.UpdateSchema()  
    .updateColumn("id", Types.LongType.get())  
    .commit()  

// 更新嵌套类里面子列类型  
table.UpdateSchema()  
    .updateColumn("locations.lat", Types.DoubleType.get())  
    .commit()  

table.UpdateSchema()  
    .updateColumn("locations.long", Types.DoubleType.get())  
    .commit()  

// 重命名顶级列  
table.UpdateSchema()  
    .renameColumn("data", "json")  
    .commit()  

// 重命名嵌套类型子列  
table.UpdateSchema()  
    .renameColumn("preferences.feature2", "newfeature")  
    .commit()  

table.UpdateSchema()  
    .renameColumn("locations.lat", "latitude")  
    .commit()  

table.UpdateSchema()  
    .renameColumn("points.x", "X")  
    .commit()  

// 删除列  
table.UpdateSchema()  
    .deleteColumn("points.z")  
    .commit();  

// 从Spark表获取Schema并转成Iceberg Schema  
Schema schema = SparkSchemaUtil  
    .schemaForTable(sparkSession, tableName);  

// 从Spark表获取Partition信息，转成Iceberg PartitionSpec  
PartitionSpec spc = SparkSchemaUtil  
    .specForTable(spark, tableName);  

// Spark StructType和Iceberg Schema互相转换  
Schema schema = SparkSchemaUtil.convert(structType);  
StructType struct = SparkSchemaUtil.convert(schema);  

import org.apache.iceberg.expressions.Expressions  
import org.apache.iceberg.Table  

val table: Table = null  

// =========Partition更新==========  
table.updateSpec()  
    // 添加分区字段user_id，且一个分区内分桶数量为10  
    .addField(Expressions.bucket("user_id", 10))  
    // 删除分区字段country  
    .removeField("country")  
    // .commit()  

// 可以通过 table = (BaseTable)origTable的方式获取内部table实例  
TableMetadata current = table.operations().current();  
PartitionSpec newSpec = PartitionSpec.builderFor(table.schema())  
    .hour("event_time")  
    .withSpecId(1)  
    .build();  
table.ops().commit(current, current.updatePartitionSpec(newSpec));  

Table sampleTable = ...;  
sampleTable.replaceSortOrder()  
    .asc("id", NullOrder.NULLS_LAST)  
    .dec("category", NullOrder.NULL_FIRST)  
    .commit();  

write.metadata.delete-after-commit.enabled=true  
write.metadata.previous-versions-max=5  

call spark_catalog.system.remove_orphan_files(table => 'db.sample');  

-- 删除孤立文件  
table.refresh()  
val orphanFilesRemoved = table.optimize("RemoveOrphanFiles")  
println(s"Orphan files removed: $orphanFilesRemoved")  

spark.sql(s"""  
    |CALL $catalogName.system.rewrite_data_files('${metaTable.schema}.${metaTable.tableName}', 'sort', '$specName DESC NULLS LAST',  
    | map('min-input-files', '1', 'target-file-size-bytes', '${128 * 1024 * 1024}', 'partial-progress.enabled', 'true', 'partial-progress.max-commits', '100'),  
    | '$specName >= "${dateFormat.format(tsToExpire)}" and $specName <= "${dateFormat.format(currentTimeMillis)}"')  
    |""".stripMargin)  

spark.sql(  
    s"""  
    |CALL $catalogName.system.rewrite_manifests('${metaTable.schema}.${metaTable.tableName}', false)  
    |""".stripMargin)  

spark.sql(  
    s"""  
    |CALL $catalogName.system.expire_snapshots('${metaTable.schema}.${metaTable.tableName}',  
    | TIMESTAMP '${dateFormat.format(tsToExpire)}', 12, 10)  
    |""".stripMargin)  

spark.sql(  
    s"""  
    |CALL $catalogName.system.remove_orphan_files('${metaTable.schema}.${metaTable.tableName}',  
    | TIMESTAMP '${dateFormat.format(tsToExpire)}', '${table.location()}/data', false, 10)  
    |""".stripMargin)  

spark.sql(  
    s"""  
    |CREATE TABLE IF NOT EXISTS hadoop_prod.$schemaName.$tableName (  
    |    id bigint,  
    |    data string,  
    |    jydw_no string,  
    |    ts timestamp)  
    |USING iceberg  
    |PARTITIONED BY (bucket(5, id), days(ts))  
    |TBLPROPERTIES ('write.distribution-mode'='hash',  
    |'write.metadata.delete-after-commit.enabled'='true',  
    |'write.metadata.previous-versions-max'='9')  
    |""".stripMargin)  

-- 清理快照  
ALTER TABLE test_table EXECUTE remove_orphan_files(retention_threshold => '7d')  

-- 合并小文件  
ALTER TABLE test_table EXECUTE optimize(file_size_threshold => '10MB')  

// 1. 定期合并新增分区的小文件：  
rewriteDataFilesAction.execute(); 仅合并小文件，不会删除旧文件。  

// 2. 删除过期的 snapshot，清理元数据及数据文件：  
table.expireSnapshots().expireOlderThan(timestamp).commit();  

// 3. 清理 orphan 文件，默认清理 3 天前，且无法触及的文件：  
removeOrphanFilesAction.olderThan(timestamp).execute();  

int maxSize = getMaxParallelSize();  
System.out.println("start to removeOrphanFile before " + today + ", " + end);  
ActionsWithParallel.forTable(table)  
    .removeOrphanFilesWithParallel()  
    .olderThan(end) // 谨慎操作，避免误删当前写入还未成功commit的datafile  
    .execute(maxSize, 10000);  

// tablelocation 改为具体的表地址，会清理不用的文件，  
CALL hive_iceberg_catalog.system.remove_orphan_files(table => 'ods_base.xx_table_name', location => 'tablelocation/data')