diff --git a/README.md b/README.md index 879a6538..c58af61d 100644 --- a/README.md +++ b/README.md @@ -368,6 +368,8 @@ rc du local/photos --fallback --incomplete ## Command Overview +For Iceberg catalog and administrative workflows, see [table catalog commands](docs/usage/table-catalog.md). + For full command documentation, see the [`rc` command reference](docs/reference/rc/README.md). | Command | Description | diff --git a/crates/cli/src/commands/admin/mod.rs b/crates/cli/src/commands/admin/mod.rs index 4e7011f6..84a19f1e 100644 --- a/crates/cli/src/commands/admin/mod.rs +++ b/crates/cli/src/commands/admin/mod.rs @@ -40,6 +40,9 @@ use rc_s3::AdminClient; /// Admin subcommands for IAM and cluster management #[derive(Subcommand, Debug)] pub enum AdminCommands { + /// Manage table catalog maintenance, recovery and migration + #[command(subcommand)] + Table(super::table::AdminTableCommands), /// Manage the identity this alias authenticates as #[command(subcommand)] Account(account::AccountCommands), @@ -140,6 +143,7 @@ pub async fn execute(cmd: AdminCommands, output_config: OutputConfig) -> ExitCod let formatter = Formatter::new(output_config); match cmd { + AdminCommands::Table(cmd) => super::table::execute_admin(cmd, &formatter).await, AdminCommands::Account(account_cmd) => account::execute(account_cmd, &formatter).await, AdminCommands::Capabilities(args) => capabilities::execute(args, &formatter).await, AdminCommands::Diagnostics(command) => diagnostics::execute(command, &formatter).await, diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index 23340dbe..cd7ab3c6 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -58,6 +58,7 @@ mod rm; mod share; mod sql; mod stat; +mod table; mod tag; mod transfer_fidelity; mod tree; @@ -224,6 +225,9 @@ impl GlobalOutputOptions { #[derive(Subcommand, Debug)] pub enum Commands { + /// Manage RustFS Iceberg tables and namespaces + #[command(subcommand)] + Table(table::TableCommands), /// Manage storage service aliases #[command(subcommand)] Alias(alias::AliasCommands), @@ -391,6 +395,15 @@ pub async fn execute(cli: Cli) -> ExitCode { }; match cli.command { + Commands::Table(cmd) => { + table::execute( + cmd, + &crate::output::Formatter::new( + output_options.resolve(OutputBehavior::StructuredDefault), + ), + ) + .await + } Commands::Alias(cmd) => { alias::execute(cmd, output_options.resolve(OutputBehavior::HumanDefault)).await } diff --git a/crates/cli/src/commands/table/contract_tests.rs b/crates/cli/src/commands/table/contract_tests.rs new file mode 100644 index 00000000..8087a162 --- /dev/null +++ b/crates/cli/src/commands/table/contract_tests.rs @@ -0,0 +1,314 @@ +use clap::Parser; + +use super::*; +use crate::commands::{Cli, Commands}; +use crate::output::OutputConfig; +use async_trait::async_trait; + +struct Api { + expected: Op, + fail: bool, +} +#[async_trait] +impl TableCatalogApi for Api { + async fn catalog(&self, request: &CatalogRequest) -> Result { + assert_eq!(request.operation, self.expected); + if self.fail { + Err(Error::Auth("denied".into())) + } else { + Ok(json!({"metadata":{"snapshots":[{"snapshot-id":1}]}})) + } + } +} +fn prepare_command(args: Vec) -> Result { + match Cli::try_parse_from(args) + .map_err(|e| Error::Config(e.to_string()))? + .command + { + Commands::Table(command) => prepare_table(command), + Commands::Admin(super::super::admin::AdminCommands::Table(command)) => { + prepare_admin(command) + } + _ => panic!("not a table command"), + } +} +#[tokio::test] +async fn catalog_commands_dispatch_success_and_permission_failure() { + let cases = [ + ("table config a/b", Op::Config), + ("table warehouse show a/b", Op::WarehouseShow), + ("admin table warehouse enable a/b", Op::WarehouseEnable), + ( + "table namespace create a/b/n --property owner=team", + Op::NamespaceCreate, + ), + ("table namespace list a/b", Op::NamespaceList), + ("table namespace show a/b/n", Op::NamespaceShow), + ("table namespace exists a/b/n", Op::NamespaceExists), + ( + "table namespace update a/b/n --set owner=team --remove obsolete", + Op::NamespaceUpdate, + ), + ("table namespace remove a/b/n", Op::NamespaceRemove), + ("table create a/b/n/t --schema-file FILE", Op::TableCreate), + ( + "table register a/b/n/t --metadata-location s3://b/m.json", + Op::TableRegister, + ), + ( + "table list a/b/n --page-size 1 --page-token opaque --no-paginate", + Op::TableList, + ), + ("table show a/b/n/t", Op::TableShow), + ("table exists a/b/n/t", Op::TableExists), + ("table rename a/b/n/t a/b/n/u", Op::TableRename), + ("table remove a/b/n/t", Op::TableRemove), + ("table metadata show a/b/n/t", Op::MetadataShow), + ( + "table snapshot list a/b/n/t --snapshots refs", + Op::TableShow, + ), + ("table snapshot show a/b/n/t 1", Op::TableShow), + ("table ref list a/b/n/t", Op::RefList), + ( + "table ref set a/b/n/t release --type tag --snapshot-id 1 --expected-snapshot-id null --commit-id ref-1", + Op::RefSet, + ), + ( + "table ref remove a/b/n/t release --expected-snapshot-id 1 --commit-id ref-2", + Op::RefRemove, + ), + ( + "table commit a/b/n/t --file COMMIT --commit-id edit-1", + Op::Commit, + ), + ("table view create a/b/n/t --file FILE", Op::ViewCreate), + ("table view list a/b/n", Op::ViewList), + ("table view show a/b/n/t", Op::ViewShow), + ("table view exists a/b/n/t", Op::ViewExists), + ("table view replace a/b/n/t --file FILE", Op::ViewReplace), + ("table view remove a/b/n/t", Op::ViewRemove), + ( + "admin table maintenance plan a/b/n/t --file FILE", + Op::MaintenancePlan, + ), + ( + "admin table maintenance run a/b/n/t --file FILE --yes", + Op::MaintenanceRun, + ), + ( + "admin table maintenance config show a/b/n/t", + Op::MaintenanceConfigShow, + ), + ( + "admin table maintenance config set a/b/n/t --file FILE --yes", + Op::MaintenanceConfigSet, + ), + ( + "admin table maintenance job show a/b/n/t job-1", + Op::MaintenanceJobShow, + ), + ( + "admin table maintenance job heartbeat a/b/n/t job-1 --file FILE --yes", + Op::JobHeartbeat, + ), + ( + "admin table maintenance job quarantine a/b/n/t job-1 --file FILE --yes", + Op::JobQuarantine, + ), + ( + "admin table maintenance scheduler show a/b/n/t", + Op::SchedulerShow, + ), + ( + "admin table maintenance scheduler run a/b/n/t --yes", + Op::SchedulerRun, + ), + ( + "admin table maintenance worker run a/b/n/t --file FILE --yes", + Op::WorkerRun, + ), + ("admin table catalog diagnostics a/b/n/t", Op::Diagnostics), + ("admin table catalog export a/b/n/t", Op::Export), + ( + "admin table catalog import a/b/n/t --file FILE --yes", + Op::Import, + ), + ("admin table catalog recover a/b/n/t --yes", Op::Recover), + ( + "admin table catalog rollback a/b/n/t --file FILE --yes", + Op::Rollback, + ), + ( + "admin table catalog metadata-update a/b/n/t --file FILE --yes", + Op::MetadataUpdate, + ), + ( + "admin table catalog external show a/b/n/t", + Op::ExternalShow, + ), + ( + "admin table catalog external set a/b/n/t --file FILE --yes", + Op::ExternalSet, + ), + ( + "admin table catalog external sync a/b/n/t --file FILE --yes", + Op::ExternalSync, + ), + ("admin table migration status a/b", Op::MigrationStatus), + ("admin table migration start a/b --yes", Op::MigrationStart), + ( + "admin table migration cancel a/b --yes", + Op::MigrationCancel, + ), + ]; + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("request.json"); + let commit = dir.path().join("commit.json"); + std::fs::write(&file,r#"{"expected-version-token":"v1","expected-metadata-location":"s3://b/m1","version-token":"v1"}"#).unwrap(); + std::fs::write(&commit,r#"{"requirements":[{"type":"assert-ref-snapshot-id","ref":"main","snapshot-id":1}],"updates":[]}"#).unwrap(); + let formatter = Formatter::new(OutputConfig { + quiet: true, + ..Default::default() + }); + for (command, expected) in cases { + for fail in [false, true] { + let args = std::iter::once("rc".to_string()) + .chain(command.split_whitespace().map(|s| { + match s { + "FILE" => file.to_str().unwrap(), + "COMMIT" => commit.to_str().unwrap(), + s => s, + } + .to_string() + })) + .collect(); + let prepared = prepare_command(args).unwrap_or_else(|e| panic!("{command}: {e}")); + let code = execute_with_api(prepared, &Api { expected, fail }, &formatter).await; + assert_eq!( + code, + if fail { + ExitCode::AuthError + } else { + ExitCode::Success + }, + "{command}" + ); + } + } +} +#[test] +fn catalog_output_matches_v3_schema() { + let schema: Value = + serde_json::from_str(include_str!("../../../../../schemas/output_v3.json")).unwrap(); + let validator = jsonschema::validator_for(&schema).unwrap(); + for value in [ + success_output(Op::TableList, json!({"identifiers":[]})), + success_output(Op::MaintenanceConfigShow, Value::Null), + error_output(&Error::Conflict("stale".into())), + error_output(&Error::UnsupportedFeature("backing".into())), + ] { + assert!(validator.is_valid(&value), "{value}"); + } +} +#[test] +fn catalog_standard_commit_rejects_ignored_version_guards() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("commit.json"); + std::fs::write(&file,r#"{"requirements":[{"type":"assert-current-schema-id","current-schema-id":0}],"updates":[]}"#).unwrap(); + let args = vec![ + "rc", + "table", + "commit", + "a/b/n/t", + "--file", + file.to_str().unwrap(), + "--commit-id", + "c1", + "--expected-version-token", + "v1", + "--expected-metadata-location", + "s3://b/m1", + ] + .into_iter() + .map(str::to_string) + .collect(); + assert!( + prepare_command(args) + .err() + .unwrap() + .to_string() + .contains("Standard updates") + ); +} + +#[test] +fn catalog_pointer_commit_requires_conditions_and_rejects_ignored_updates() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("pointer.json"); + for (body, valid) in [ + (json!({"new-metadata-location":"s3://b/m2"}), false), + ( + json!({"new-metadata-location":"s3://b/m2","expected-version-token":"v1","expected-metadata-location":"s3://b/m1"}), + true, + ), + ( + json!({"new-metadata-location":"s3://b/m2","expected-version-token":"v1","expected-metadata-location":"s3://b/m1","updates":[{"action":"set-properties","updates":{"owner":"new"}}]}), + false, + ), + ] { + std::fs::write(&file, body.to_string()).unwrap(); + let args = [ + "rc", + "table", + "commit", + "a/b/n/t", + "--file", + file.to_str().unwrap(), + "--commit-id", + "c1", + ] + .into_iter() + .map(str::to_string) + .collect(); + assert_eq!(prepare_command(args).is_ok(), valid, "{body}"); + } +} + +#[test] +fn catalog_namespace_list_accepts_root_and_parent_targets() { + for (target, namespace) in [ + ("a/warehouse", Vec::::new()), + ("a/warehouse/sales.eu", vec!["sales".into(), "eu".into()]), + ] { + let args = [ + "rc", + "table", + "namespace", + "list", + target, + "--page-token", + "opaque", + "--no-paginate", + ] + .into_iter() + .map(str::to_string) + .collect(); + let prepared = prepare_command(args).unwrap(); + assert_eq!(prepared.request.operation, Op::NamespaceList); + assert_eq!(prepared.request.target.namespace, namespace); + assert_eq!(prepared.request.page_token.as_deref(), Some("opaque")); + assert!(prepared.request.single_page); + } + for target in [ + "a/warehouse/sales..eu", + "a/warehouse/sales%2Feu", + "a/warehouse/sales/table", + ] { + let args = ["rc", "table", "namespace", "list", target] + .into_iter() + .map(str::to_string) + .collect(); + assert!(prepare_command(args).is_err()); + } +} diff --git a/crates/cli/src/commands/table/mod.rs b/crates/cli/src/commands/table/mod.rs new file mode 100644 index 00000000..53018182 --- /dev/null +++ b/crates/cli/src/commands/table/mod.rs @@ -0,0 +1,907 @@ +//! Catalog commands. Data-file reads and writes remain query-engine operations. + +use crate::{exit_code::ExitCode, output::Formatter}; +use clap::{Args, Subcommand, ValueEnum}; +use rc_core::catalog::{ + CatalogOperation as Op, CatalogRequest, CatalogTarget, ResourceKind as Kind, TableCatalogApi, +}; +use rc_core::{Error, Result}; +use serde_json::{Value, json}; +use std::{collections::BTreeMap, io::Read, path::PathBuf}; + +#[derive(Debug, Args)] +pub struct TargetArgs { + /// Catalog resource: alias/warehouse[/namespace[/table]] + pub target: String, +} +#[derive(Debug, Args)] +pub struct ListArgs { + #[command(flatten)] + pub target: TargetArgs, + /// Number of entries per server page (all pages are fetched by default) + #[arg(long, default_value_t=1000, value_parser=clap::value_parser!(u16).range(1..=1000))] + pub page_size: u16, + /// Opaque continuation token from an earlier single-page result + #[arg(long)] + pub page_token: Option, + /// Return one page and its next-page-token + #[arg(long)] + pub no_paginate: bool, +} +#[derive(Debug, Args)] +pub struct FileArgs { + #[command(flatten)] + pub target: TargetArgs, + /// JSON request object file; '-' reads stdin (maximum 8 MiB) + #[arg(long)] + pub file: PathBuf, +} +#[derive(Debug, Args)] +pub struct AdminFileArgs { + #[command(flatten)] + pub input: FileArgs, + /// Confirm this administrative mutation + #[arg(long, required = true)] + pub yes: bool, +} +#[derive(Debug, Args)] +pub struct AdminMutationArgs { + #[command(flatten)] + pub target: TargetArgs, + /// Confirm this administrative mutation + #[arg(long, required = true)] + pub yes: bool, +} +#[derive(Debug, Args)] +pub struct CreateArgs { + #[command(flatten)] + pub target: TargetArgs, + #[arg(long)] + pub schema_file: PathBuf, + #[arg(long)] + pub partition_spec_file: Option, + #[arg(long)] + pub sort_order_file: Option, + /// Optional location within this warehouse + #[arg(long)] + pub location: Option, + #[arg(long, default_value_t=2, value_parser=clap::value_parser!(u8).range(1..=2))] + pub format_version: u8, + /// Table property, repeatable as key=value + #[arg(long)] + pub property: Vec, +} +#[derive(Debug, Args)] +pub struct RegisterArgs { + #[command(flatten)] + pub target: TargetArgs, + #[arg(long)] + pub metadata_location: String, +} +#[derive(Debug, Args)] +pub struct RenameArgs { + pub source: String, + /// Destination in the same alias and warehouse + pub destination: String, +} +#[derive(Debug, Args)] +pub struct NamespaceCreateArgs { + #[command(flatten)] + pub target: TargetArgs, + #[arg(long)] + pub property: Vec, +} +#[derive(Debug, Args)] +pub struct NamespaceUpdateArgs { + #[command(flatten)] + pub target: TargetArgs, + #[arg(long)] + pub set: Vec, + #[arg(long)] + pub remove: Vec, +} +#[derive(Debug, Clone, Copy, ValueEnum)] +pub enum Snapshots { + All, + Refs, +} +#[derive(Debug, Args)] +pub struct ShowArgs { + #[command(flatten)] + pub target: TargetArgs, + #[arg(long, value_enum, default_value_t=Snapshots::All)] + pub snapshots: Snapshots, +} +#[derive(Debug, Args)] +pub struct SnapshotShowArgs { + #[command(flatten)] + pub target: TargetArgs, + pub snapshot_id: i64, +} +#[derive(Debug, Args)] +pub struct CommitArgs { + #[command(flatten)] + pub input: FileArgs, + /// Version token read before preparing the mutation + #[arg(long)] + pub expected_version_token: Option, + /// Metadata location read before preparing the mutation + #[arg(long)] + pub expected_metadata_location: Option, + /// Stable identifier for this logical commit; reuse after an uncertain outcome + #[arg(long)] + pub commit_id: String, +} +#[derive(Debug, Clone, Copy, ValueEnum)] +pub enum RefType { + Branch, + Tag, +} +#[derive(Debug, Args)] +pub struct RefSetArgs { + #[command(flatten)] + pub target: TargetArgs, + pub name: String, + #[arg(long)] + pub snapshot_id: i64, + #[arg(long = "type", value_enum)] + pub ref_type: RefType, + /// Current snapshot ID, or 'null' to require an absent reference + #[arg(long)] + pub expected_snapshot_id: String, + #[arg(long)] + pub commit_id: String, + #[arg(long)] + pub min_snapshots_to_keep: Option, + #[arg(long)] + pub max_snapshot_age_ms: Option, + #[arg(long)] + pub max_ref_age_ms: Option, +} +#[derive(Debug, Args)] +pub struct RefRemoveArgs { + #[command(flatten)] + pub target: TargetArgs, + pub name: String, + #[arg(long)] + pub expected_snapshot_id: i64, + #[arg(long)] + pub commit_id: String, + /// Allow removing a ref with explicit retention; never bypasses main protection + #[arg(long)] + pub force: bool, +} +#[derive(Debug, Args)] +pub struct JobArgs { + #[command(flatten)] + pub target: TargetArgs, + pub job: String, +} +#[derive(Debug, Args)] +pub struct JobMutationArgs { + #[command(flatten)] + pub input: AdminFileArgs, + pub job: String, +} + +#[derive(Debug, Subcommand)] +pub enum TableCommands { + /// Discover the catalog configuration for a warehouse + Config(TargetArgs), + /// Inspect or enable a table warehouse + #[command(subcommand)] + Warehouse(WarehouseCommands), + /// Manage namespaces and their properties + #[command(subcommand)] + Namespace(NamespaceCommands), + /// Create an Iceberg v1/v2 table; does not write rows + Create(CreateArgs), + /// Register an existing metadata file without overwriting a table + Register(RegisterArgs), + /// List tables in a namespace + List(ListArgs), + /// Load table metadata, without credential bundles + Show(ShowArgs), + /// Return success if present, exit 5 if absent + Exists(TargetArgs), + /// Rename a table within the same warehouse + Rename(RenameArgs), + /// Drop the catalog entry, preserving data files (purge is unsupported) + Remove(TargetArgs), + /// Read the current metadata pointer and version token + #[command(subcommand)] + Metadata(MetadataCommands), + /// Inspect snapshot metadata without querying rows + #[command(subcommand)] + Snapshot(SnapshotCommands), + /// Manage snapshot branches and tags + #[command(subcommand)] + Ref(RefCommands), + /// Submit metadata updates with explicit optimistic concurrency conditions + Commit(CommitArgs), + /// Manage Iceberg view definitions + #[command(subcommand)] + View(ViewCommands), +} +#[derive(Debug, Subcommand)] +pub enum WarehouseCommands { + Show(TargetArgs), +} +#[derive(Debug, Subcommand)] +pub enum NamespaceCommands { + Create(NamespaceCreateArgs), + /// List root namespaces for a warehouse, or direct children of a namespace + List(ListArgs), + Show(TargetArgs), + Exists(TargetArgs), + Update(NamespaceUpdateArgs), + Remove(TargetArgs), +} +#[derive(Debug, Subcommand)] +pub enum MetadataCommands { + Show(TargetArgs), +} +#[derive(Debug, Subcommand)] +pub enum SnapshotCommands { + List(ShowArgs), + Show(SnapshotShowArgs), +} +#[derive(Debug, Subcommand)] +pub enum RefCommands { + List(TargetArgs), + Set(RefSetArgs), + Remove(RefRemoveArgs), +} +#[derive(Debug, Subcommand)] +pub enum ViewCommands { + /// JSON contains schema and view-version; name is derived from target + Create(FileArgs), + List(ListArgs), + Show(TargetArgs), + Exists(TargetArgs), + /// JSON contains view requirements/updates plus expected-metadata-location + Replace(FileArgs), + Remove(TargetArgs), +} +#[derive(Debug, Subcommand)] +pub enum AdminTableCommands { + /// Inspect or enable a table warehouse + #[command(subcommand)] + Warehouse(AdminWarehouseCommands), + /// Inspect and run guarded maintenance operations + #[command(subcommand)] + Maintenance(MaintenanceCommands), + /// Inspect, export, import and recover catalog state + #[command(subcommand)] + Catalog(CatalogCommands), + /// Inspect and control backing migration + #[command(subcommand)] + Migration(MigrationCommands), +} +#[derive(Debug, Subcommand)] +pub enum AdminWarehouseCommands { + /// Enable catalog on an existing bucket; create the bucket separately + Enable(TargetArgs), +} +#[derive(Debug, Subcommand)] +pub enum MaintenanceCommands { + /// Preview maintenance; overrides all deletion and commit flags to false + Plan(FileArgs), + /// Execute only the deletion/commit actions explicitly selected in the JSON + Run(AdminFileArgs), + /// Read or update maintenance configuration + #[command(subcommand)] + Config(MaintenanceConfigCommands), + /// Inspect jobs and control leases or quarantine + #[command(subcommand)] + Job(JobCommands), + /// Inspect or run one scheduling pass + #[command(subcommand)] + Scheduler(SchedulerCommands), + /// Run one maintenance worker pass + #[command(subcommand)] + Worker(WorkerCommands), +} +#[derive(Debug, Subcommand)] +pub enum MaintenanceConfigCommands { + Show(TargetArgs), + Set(AdminFileArgs), +} +#[derive(Debug, Subcommand)] +pub enum JobCommands { + Show(JobArgs), + Heartbeat(JobMutationArgs), + Quarantine(JobMutationArgs), +} +#[derive(Debug, Subcommand)] +pub enum SchedulerCommands { + Show(TargetArgs), + Run(AdminMutationArgs), +} +#[derive(Debug, Subcommand)] +pub enum WorkerCommands { + Run(AdminFileArgs), +} +#[derive(Debug, Subcommand)] +pub enum CatalogCommands { + Diagnostics(TargetArgs), + Export(TargetArgs), + Import(AdminFileArgs), + Recover(AdminMutationArgs), + Rollback(AdminFileArgs), + /// Update metadata pointer using a JSON request with version-token + MetadataUpdate(AdminFileArgs), + /// Manage operator-supplied external catalog pointers + #[command(subcommand)] + External(ExternalCommands), +} +#[derive(Debug, Subcommand)] +pub enum ExternalCommands { + Show(TargetArgs), + Set(AdminFileArgs), + Sync(AdminFileArgs), +} +#[derive(Debug, Subcommand)] +pub enum MigrationCommands { + Status(TargetArgs), + Start(AdminMutationArgs), + Cancel(AdminMutationArgs), +} + +struct Prepared { + request: CatalogRequest, + projection: Projection, +} +enum Projection { + None, + Snapshots, + Snapshot(i64), +} +fn prepare(op: Op, args: TargetArgs, kind: Kind) -> Result { + Ok(Prepared { + request: CatalogRequest::new(op, CatalogTarget::parse(&args.target, kind)?), + projection: Projection::None, + }) +} +fn list(op: Op, args: ListArgs, kind: Kind) -> Result { + let mut prepared = prepare(op, args.target, kind)?; + prepared.request.page_size = args.page_size; + prepared.request.page_token = args.page_token; + prepared.request.single_page = args.no_paginate; + Ok(prepared) +} +fn read_json(path: &PathBuf) -> Result { + let mut bytes = Vec::new(); + let reader: Box = if path.as_os_str() == "-" { + Box::new(std::io::stdin()) + } else { + Box::new(std::fs::File::open(path)?) + }; + reader.take(8 * 1024 * 1024 + 1).read_to_end(&mut bytes)?; + if bytes.len() > 8 * 1024 * 1024 { + return Err(Error::Config("JSON input exceeds 8 MiB".into())); + } + serde_json::from_slice(&bytes).map_err(|_| Error::Config("Invalid JSON input".into())) +} +fn file(op: Op, args: FileArgs) -> Result { + let mut prepared = prepare(op, args.target, Kind::Table)?; + let value = read_json(&args.file)?; + if !value.is_object() { + return Err(Error::Config("Request must be a JSON object".into())); + } + prepared.request.body = Some(value); + Ok(prepared) +} +fn properties(values: Vec) -> Result> { + let mut result = BTreeMap::new(); + for item in values { + let (key, value) = item + .split_once('=') + .filter(|(key, _)| !key.is_empty()) + .ok_or_else(|| Error::Config("Properties must be key=value".into()))?; + if result.insert(key.into(), value.into()).is_some() { + return Err(Error::Config("Duplicate property".into())); + } + } + Ok(result) +} +fn require_string(body: &Value, field: &str) -> Result<()> { + if !body + .get(field) + .and_then(Value::as_str) + .is_some_and(|s| !s.trim().is_empty()) + { + return Err(Error::Config(format!("Request requires nonempty {field}"))); + } + Ok(()) +} +fn guard_commit(body: &Value) -> Result<()> { + require_string(body, "expected-version-token")?; + require_string(body, "expected-metadata-location") +} +fn show(args: ShowArgs) -> Result { + let mut p = prepare(Op::TableShow, args.target, Kind::Table)?; + p.request.snapshots = Some( + match args.snapshots { + Snapshots::All => "all", + Snapshots::Refs => "refs", + } + .into(), + ); + Ok(p) +} + +fn prepare_table(command: TableCommands) -> Result { + match command { + TableCommands::Config(a) => prepare(Op::Config, a, Kind::Warehouse), + TableCommands::Warehouse(WarehouseCommands::Show(a)) => { + prepare(Op::WarehouseShow, a, Kind::Warehouse) + } + TableCommands::Namespace(command) => match command { + NamespaceCommands::List(a) => { + let kind = if a.target.target.split('/').count() == 2 { + Kind::Warehouse + } else { + Kind::Namespace + }; + list(Op::NamespaceList, a, kind) + } + NamespaceCommands::Show(a) => prepare(Op::NamespaceShow, a, Kind::Namespace), + NamespaceCommands::Exists(a) => prepare(Op::NamespaceExists, a, Kind::Namespace), + NamespaceCommands::Remove(a) => prepare(Op::NamespaceRemove, a, Kind::Namespace), + NamespaceCommands::Create(a) => { + let mut p = prepare(Op::NamespaceCreate, a.target, Kind::Namespace)?; + p.request.body = Some( + json!({"namespace": p.request.target.namespace, "properties": properties(a.property)?}), + ); + Ok(p) + } + NamespaceCommands::Update(a) => { + let mut p = prepare(Op::NamespaceUpdate, a.target, Kind::Namespace)?; + let updates = properties(a.set)?; + if a.remove.iter().any(|key| updates.contains_key(key)) { + return Err(Error::Config( + "Cannot set and remove the same property".into(), + )); + } + p.request.body = Some(json!({"updates": updates, "removals": a.remove})); + Ok(p) + } + }, + TableCommands::Create(a) => { + let mut p = prepare(Op::TableCreate, a.target, Kind::Table)?; + let mut props = properties(a.property)?; + if props.contains_key("format-version") { + return Err(Error::Config( + "Use --format-version instead of a format-version property".into(), + )); + } + props.insert("format-version".into(), a.format_version.to_string()); + let mut body = json!({"name": p.request.target.name, "schema": read_json(&a.schema_file)?, "properties": props}); + if let Some(path) = a.partition_spec_file { + body["partition-spec"] = read_json(&path)?; + } + if let Some(path) = a.sort_order_file { + body["write-order"] = read_json(&path)?; + } + if let Some(location) = a.location { + body["location"] = json!(location); + } + p.request.body = Some(body); + Ok(p) + } + TableCommands::Register(a) => { + let mut p = prepare(Op::TableRegister, a.target, Kind::Table)?; + p.request.body = Some( + json!({"name": p.request.target.name, "metadata-location": a.metadata_location}), + ); + Ok(p) + } + TableCommands::List(a) => list(Op::TableList, a, Kind::Namespace), + TableCommands::Show(a) => show(a), + TableCommands::Exists(a) => prepare(Op::TableExists, a, Kind::Table), + TableCommands::Remove(a) => prepare(Op::TableRemove, a, Kind::Table), + TableCommands::Rename(a) => { + let mut p = prepare( + Op::TableRename, + TargetArgs { target: a.source }, + Kind::Table, + )?; + let dest = CatalogTarget::parse(&a.destination, Kind::Table)?; + let source = &p.request.target; + if source.alias != dest.alias || source.warehouse != dest.warehouse { + return Err(Error::Config( + "Rename requires the same alias and warehouse".into(), + )); + } + p.request.body = Some( + json!({"source": {"namespace": source.namespace, "name": source.name}, "destination": {"namespace": dest.namespace, "name": dest.name}}), + ); + Ok(p) + } + TableCommands::Metadata(MetadataCommands::Show(a)) => { + prepare(Op::MetadataShow, a, Kind::Table) + } + TableCommands::Snapshot(SnapshotCommands::List(a)) => { + let mut p = show(a)?; + p.projection = Projection::Snapshots; + Ok(p) + } + TableCommands::Snapshot(SnapshotCommands::Show(a)) => { + let mut p = prepare(Op::TableShow, a.target, Kind::Table)?; + p.projection = Projection::Snapshot(a.snapshot_id); + Ok(p) + } + TableCommands::Commit(a) => { + let mut p = file(Op::Commit, a.input)?; + let body = p.request.body.as_mut().expect("file request has a body"); + if body + .get("commit-id") + .is_some_and(|old| old != &json!(a.commit_id)) + { + return Err(Error::Config( + "Conflicting commit-id in file and flags".into(), + )); + } + body["commit-id"] = json!(a.commit_id); + require_string(body, "commit-id")?; + if body.get("new-metadata-location").is_some() { + for (key, value) in [ + ("expected-version-token", a.expected_version_token), + ("expected-metadata-location", a.expected_metadata_location), + ] { + if let Some(value) = value { + if body.get(key).is_some_and(|old| old != &json!(value)) { + return Err(Error::Config(format!( + "Conflicting {key} in file and flags" + ))); + } + body[key] = json!(value); + } + } + if body + .get("updates") + .is_some_and(|v| !v.as_array().is_some_and(Vec::is_empty)) + { + return Err(Error::Config( + "Pointer commits cannot include metadata updates".into(), + )); + } + guard_commit(body)?; + require_string(body, "new-metadata-location")?; + } else { + if a.expected_version_token.is_some() + || a.expected_metadata_location.is_some() + || body.get("expected-version-token").is_some() + || body.get("expected-metadata-location").is_some() + { + return Err(Error::Config("Standard updates use Iceberg requirements; version/location guards require new-metadata-location".into())); + } + if !body + .get("requirements") + .and_then(Value::as_array) + .is_some_and(|v| !v.is_empty()) + { + return Err(Error::Config( + "Standard commit requires explicit Iceberg requirements".into(), + )); + } + if !body.get("updates").is_some_and(Value::is_array) { + return Err(Error::Config( + "Standard commit requires an updates array".into(), + )); + } + } + Ok(p) + } + TableCommands::Ref(command) => match command { + RefCommands::List(a) => prepare(Op::RefList, a, Kind::Table), + RefCommands::Set(a) => { + let mut p = prepare(Op::RefSet, a.target, Kind::Table)?; + let expected: Value = + serde_json::from_str(&a.expected_snapshot_id).map_err(|_| { + Error::Config("expected-snapshot-id must be an integer or null".into()) + })?; + if !expected.is_null() && expected.as_i64().is_none() { + return Err(Error::Config( + "expected-snapshot-id must be an integer or null".into(), + )); + } + let mut body = json!({"snapshot-id":a.snapshot_id,"type":match a.ref_type { RefType::Branch=>"branch", RefType::Tag=>"tag" },"expected-snapshot-id":expected,"commit-id":a.commit_id}); + for (key, value) in [ + ("min-snapshots-to-keep", a.min_snapshots_to_keep), + ("max-snapshot-age-ms", a.max_snapshot_age_ms), + ("max-ref-age-ms", a.max_ref_age_ms), + ] { + if let Some(value) = value { + body[key] = json!(value); + } + } + require_string(&body, "commit-id")?; + p.request.child = Some(a.name); + p.request.body = Some(body); + Ok(p) + } + RefCommands::Remove(a) => { + let mut p = prepare(Op::RefRemove, a.target, Kind::Table)?; + p.request.child = Some(a.name); + p.request.body = Some( + json!({"expected-snapshot-id":a.expected_snapshot_id,"commit-id":a.commit_id,"force":a.force}), + ); + require_string(p.request.body.as_ref().expect("body"), "commit-id")?; + Ok(p) + } + }, + TableCommands::View(command) => match command { + ViewCommands::Create(a) => { + let mut p = file(Op::ViewCreate, a)?; + let body = p.request.body.as_mut().expect("body"); + if body + .get("name") + .is_some_and(|name| name != &json!(p.request.target.name)) + { + return Err(Error::Config("View name differs from target".into())); + } + body["name"] = json!(p.request.target.name); + Ok(p) + } + ViewCommands::List(a) => list(Op::ViewList, a, Kind::Namespace), + ViewCommands::Show(a) => prepare(Op::ViewShow, a, Kind::Table), + ViewCommands::Exists(a) => prepare(Op::ViewExists, a, Kind::Table), + ViewCommands::Remove(a) => prepare(Op::ViewRemove, a, Kind::Table), + ViewCommands::Replace(a) => { + let p = file(Op::ViewReplace, a)?; + require_string( + p.request.body.as_ref().expect("body"), + "expected-metadata-location", + )?; + Ok(p) + } + }, + } +} + +fn admin_file(op: Op, args: AdminFileArgs) -> Result { + if !args.yes { + return Err(Error::Config( + "Administrative mutation requires --yes".into(), + )); + } + file(op, args.input) +} +fn admin_mutation(op: Op, args: AdminMutationArgs, kind: Kind) -> Result { + if !args.yes { + return Err(Error::Config( + "Administrative mutation requires --yes".into(), + )); + } + prepare(op, args.target, kind) +} +fn prepare_admin(command: AdminTableCommands) -> Result { + match command { + AdminTableCommands::Warehouse(AdminWarehouseCommands::Enable(a)) => { + prepare(Op::WarehouseEnable, a, Kind::Warehouse) + } + AdminTableCommands::Migration(c) => match c { + MigrationCommands::Status(a) => prepare(Op::MigrationStatus, a, Kind::Warehouse), + MigrationCommands::Start(a) => admin_mutation(Op::MigrationStart, a, Kind::Warehouse), + MigrationCommands::Cancel(a) => admin_mutation(Op::MigrationCancel, a, Kind::Warehouse), + }, + AdminTableCommands::Catalog(c) => match c { + CatalogCommands::Diagnostics(a) => prepare(Op::Diagnostics, a, Kind::Table), + CatalogCommands::Export(a) => prepare(Op::Export, a, Kind::Table), + CatalogCommands::Import(a) => admin_file(Op::Import, a), + CatalogCommands::Recover(a) => admin_mutation(Op::Recover, a, Kind::Table), + CatalogCommands::Rollback(a) => { + let p = admin_file(Op::Rollback, a)?; + require_string(p.request.body.as_ref().expect("body"), "version-token")?; + Ok(p) + } + CatalogCommands::MetadataUpdate(a) => { + let p = admin_file(Op::MetadataUpdate, a)?; + require_string(p.request.body.as_ref().expect("body"), "version-token")?; + Ok(p) + } + CatalogCommands::External(c) => match c { + ExternalCommands::Show(a) => prepare(Op::ExternalShow, a, Kind::Table), + ExternalCommands::Set(a) => admin_file(Op::ExternalSet, a), + ExternalCommands::Sync(a) => admin_file(Op::ExternalSync, a), + }, + }, + AdminTableCommands::Maintenance(c) => match c { + MaintenanceCommands::Plan(a) => { + let mut p = file(Op::MaintenancePlan, a)?; + let body = p.request.body.as_mut().expect("body"); + for key in ["delete", "commit-snapshot-expiration", "commit-compaction"] { + body[key] = json!(false); + } + Ok(p) + } + MaintenanceCommands::Run(a) => admin_file(Op::MaintenanceRun, a), + MaintenanceCommands::Config(c) => match c { + MaintenanceConfigCommands::Show(a) => { + prepare(Op::MaintenanceConfigShow, a, Kind::Table) + } + MaintenanceConfigCommands::Set(a) => admin_file(Op::MaintenanceConfigSet, a), + }, + MaintenanceCommands::Scheduler(c) => match c { + SchedulerCommands::Show(a) => prepare(Op::SchedulerShow, a, Kind::Table), + SchedulerCommands::Run(a) => admin_mutation(Op::SchedulerRun, a, Kind::Table), + }, + MaintenanceCommands::Worker(WorkerCommands::Run(a)) => admin_file(Op::WorkerRun, a), + MaintenanceCommands::Job(c) => match c { + JobCommands::Show(a) => { + let mut p = prepare(Op::MaintenanceJobShow, a.target, Kind::Table)?; + p.request.child = Some(a.job); + Ok(p) + } + JobCommands::Heartbeat(a) => { + let mut p = admin_file(Op::JobHeartbeat, a.input)?; + p.request.child = Some(a.job); + Ok(p) + } + JobCommands::Quarantine(a) => { + let mut p = admin_file(Op::JobQuarantine, a.input)?; + p.request.child = Some(a.job); + Ok(p) + } + }, + }, + } +} + +pub async fn execute(command: TableCommands, formatter: &Formatter) -> ExitCode { + run(prepare_table(command), formatter).await +} +pub async fn execute_admin(command: AdminTableCommands, formatter: &Formatter) -> ExitCode { + run(prepare_admin(command), formatter).await +} +async fn run(prepared: Result, formatter: &Formatter) -> ExitCode { + let prepared = match prepared { + Ok(p) => p, + Err(e) => return emit_error(&e, formatter), + }; + let client = rc_core::AliasManager::new() + .and_then(|aliases| aliases.get(&prepared.request.target.alias)) + .and_then(|alias| rc_s3::AdminClient::new(&alias)); + let client = match client { + Ok(client) => client, + Err(error) => return emit_error(&error, formatter), + }; + execute_with_api(prepared, &client, formatter).await +} +async fn execute_with_api( + prepared: Prepared, + api: &dyn TableCatalogApi, + formatter: &Formatter, +) -> ExitCode { + let result = async { + let value = api.catalog(&prepared.request).await?; + match prepared.projection { + Projection::None => Ok(value), + Projection::Snapshots => { + let snapshots = value + .pointer("/metadata/snapshots") + .and_then(Value::as_array) + .ok_or_else(|| Error::General("LoadTable response missing snapshots".into()))?; + Ok(json!({"snapshots":snapshots})) + } + Projection::Snapshot(id) => { + let snapshots = value + .pointer("/metadata/snapshots") + .and_then(Value::as_array) + .ok_or_else(|| Error::General("LoadTable response missing snapshots".into()))?; + snapshots + .iter() + .find(|v| v.get("snapshot-id").and_then(Value::as_i64) == Some(id)) + .cloned() + .ok_or_else(|| Error::NotFound(format!("Snapshot {id}"))) + } + } + } + .await; + match result { + Ok(data) => { + if formatter.is_json() { + formatter.json(&success_output(prepared.request.operation, data)); + } else { + formatter.println( + &formatter + .sanitize_text(&serde_json::to_string_pretty(&data).unwrap_or_default()), + ); + } + ExitCode::Success + } + Err(error) => emit_error(&error, formatter), + } +} +fn success_output(operation: Op, result: Value) -> Value { + json!({"schema_version":3,"type":"table_catalog","status":"success","data":{"operation":operation,"result":result}}) +} +fn error_output(error: &Error) -> Value { + let kind = match error.exit_code() { + 2 => "usage_error", + 3 => "network_error", + 4 => "auth_error", + 5 => "not_found", + 6 => "conflict", + 7 => "unsupported_feature", + 130 => "interrupted", + _ => "general_error", + }; + let mut detail = json!({"type":kind,"message":error.to_string(),"retryable":false}); + if error.exit_code() == 7 { + detail["capability"] = json!("table_catalog"); + detail["server"] = Value::Null; + } + json!({"schema_version":3,"type":"table_catalog","status":"error","error":detail}) +} +fn emit_error(error: &Error, formatter: &Formatter) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + if formatter.is_json() { + formatter.json_error(&error_output(error)); + } else { + formatter.error_with_code(code, &error.to_string()); + } + code +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + #[derive(Parser)] + struct TestCli { + #[command(subcommand)] + command: TableCommands, + } + #[test] + fn catalog_rejects_unavailable_flags_and_missing_guards() { + for args in [ + vec!["rc", "remove", "a/b/n/t", "--purge"], + vec![ + "rc", + "create", + "a/b/n/t", + "--schema-file", + "s.json", + "--format-version", + "3", + ], + vec!["rc", "commit", "a/b/n/t", "--file", "x.json"], + ] { + assert!(TestCli::try_parse_from(args).is_err()); + } + } + #[test] + fn catalog_rename_cannot_cross_warehouses() { + let command = TestCli::parse_from(["rc", "rename", "a/b/n/t", "a/c/n/u"]).command; + assert!(prepare_table(command).is_err()); + } + #[test] + fn catalog_preview_clears_mutations() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("request.json"); + std::fs::write( + &path, + r#"{"delete":true,"commit-compaction":true,"commit-snapshot-expiration":true}"#, + ) + .unwrap(); + let p = prepare_admin(AdminTableCommands::Maintenance(MaintenanceCommands::Plan( + FileArgs { + target: TargetArgs { + target: "a/b/n/t".into(), + }, + file: path, + }, + ))) + .unwrap(); + let body = p.request.body.unwrap(); + assert_eq!(body["delete"], false); + assert_eq!(body["commit-compaction"], false); + assert_eq!(body["commit-snapshot-expiration"], false); + } +} + +#[cfg(test)] +mod contract_tests; diff --git a/crates/cli/tests/table_catalog.rs b/crates/cli/tests/table_catalog.rs new file mode 100644 index 00000000..c7ba1c13 --- /dev/null +++ b/crates/cli/tests/table_catalog.rs @@ -0,0 +1,126 @@ +#![cfg(not(windows))] + +mod admin_support; + +use admin_support::{rc_binary, rc_host_alias, start_admin_sequence_test_server}; +use serde_json::{Value, json}; +use std::process::Command; +use std::time::Duration; + +#[test] +fn catalog_binary_paginates_signed_requests_and_redacts_credentials() { + let config = tempfile::tempdir().unwrap(); + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![ + ( + "200 OK", + r#"{"identifiers":[{"name":"one"}],"next-page-token":"a+/="}"#, + ), + ( + "200 OK", + r#"{"identifiers":[{"name":"two"}],"storage-credentials":[{"secret-access-key":"hidden"}]}"#, + ), + ]); + let output = Command::new(rc_binary()) + .args([ + "table", + "list", + "myalias/warehouse.bucket/sales.eu", + "--json", + ]) + .env("RC_CONFIG_DIR", config.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .env("NO_PROXY", "*") + .env("no_proxy", "*") + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let data: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(data["type"], "table_catalog"); + assert_eq!(data["data"]["operation"], "table_list"); + assert_eq!( + data["data"]["result"]["identifiers"], + json!([{"name":"one"},{"name":"two"}]) + ); + assert!(!String::from_utf8_lossy(&output.stdout).contains("hidden")); + let first = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + let second = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + first.target, + "/iceberg/v1/warehouse.bucket/namespaces/sales%1Feu/tables?pageSize=1000" + ); + assert!( + first + .headers + .to_ascii_lowercase() + .contains("authorization: aws4-hmac-sha256") + ); + assert!(second.target.ends_with("pageToken=a%2B%2F%3D")); + server.join().unwrap(); +} + +#[test] +fn catalog_binary_conflict_preserves_pointer_guards_and_exit_code() { + let config = tempfile::tempdir().unwrap(); + let file = config.path().join("commit.json"); + std::fs::write(&file, r#"{"new-metadata-location":"s3://warehouse/m2"}"#).unwrap(); + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![( + "409 Conflict", + r#"{"error":{"message":"stale metadata"}}"#, + )]); + let output = Command::new(rc_binary()) + .args(["table", "commit", "myalias/warehouse/ns/table", "--file"]) + .arg(file) + .args([ + "--expected-version-token", + "v1", + "--expected-metadata-location", + "s3://warehouse/m1", + "--commit-id", + "attempt-1", + "--json", + ]) + .env("RC_CONFIG_DIR", config.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .env("NO_PROXY", "*") + .env("no_proxy", "*") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(6)); + assert!(output.stdout.is_empty()); + let data: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!(data["status"], "error"); + assert_eq!(data["error"]["retryable"], false); + let request = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(request.method, "POST"); + let body: Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!(body["expected-version-token"], "v1"); + assert_eq!(body["expected-metadata-location"], "s3://warehouse/m1"); + assert_eq!(body["commit-id"], "attempt-1"); + server.join().unwrap(); + assert!(requests.try_recv().is_err()); +} + +#[test] +fn catalog_binary_missing_alias_uses_catalog_error_envelope() { + let config = tempfile::tempdir().unwrap(); + let output = Command::new(rc_binary()) + .args([ + "table", + "show", + "missing-catalog-alias/warehouse/ns/table", + "--json", + ]) + .env("RC_CONFIG_DIR", config.path()) + .env_remove("RC_HOST_missing-catalog-alias") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(5)); + assert!(output.stdout.is_empty()); + let data: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!(data["type"], "table_catalog"); + assert_eq!(data["error"]["type"], "not_found"); +} diff --git a/crates/core/src/catalog.rs b/crates/core/src/catalog.rs new file mode 100644 index 00000000..f6bf28a3 --- /dev/null +++ b/crates/core/src/catalog.rs @@ -0,0 +1,184 @@ +//! Table catalog resources and operations, independent of HTTP and storage SDKs. + +use crate::{Error, Result}; +use async_trait::async_trait; +use serde_json::Value; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ResourceKind { + Warehouse, + Namespace, + Table, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CatalogTarget { + pub alias: String, + pub warehouse: String, + pub namespace: Vec, + pub name: Option, +} + +impl CatalogTarget { + pub fn parse(path: &str, kind: ResourceKind) -> Result { + let parts: Vec<_> = path.split('/').collect(); + let count = match kind { + ResourceKind::Warehouse => 2, + ResourceKind::Namespace => 3, + ResourceKind::Table => 4, + }; + if parts.len() != count || parts[0].is_empty() { + return Err(Error::InvalidPath( + "Expected alias/warehouse[/namespace[/table]]".into(), + )); + } + validate_warehouse(parts[1])?; + let namespace = if count >= 3 { + if parts[2].len() > 512 { + return Err(Error::InvalidPath("Namespace exceeds 512 bytes".into())); + } + parts[2] + .split('.') + .map(|segment| { + validate_segment(segment)?; + Ok(segment.to_owned()) + }) + .collect::>>()? + } else { + Vec::new() + }; + let name = if count == 4 { + validate_segment(parts[3])?; + Some(parts[3].to_owned()) + } else { + None + }; + Ok(Self { + alias: parts[0].into(), + warehouse: parts[1].into(), + namespace, + name, + }) + } +} + +/// Preserve existing bucket names while rejecting URL path syntax. +pub fn validate_warehouse(value: &str) -> Result<()> { + if value.is_empty() + || value.len() > 255 + || !value + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)) + || matches!(value, "." | "..") + { + return Err(Error::InvalidPath("Invalid warehouse bucket name".into())); + } + Ok(()) +} + +pub fn validate_segment(value: &str) -> Result<()> { + let boundary = |b: u8| b.is_ascii_lowercase() || b.is_ascii_digit(); + if value.is_empty() + || value.len() > 64 + || !value.bytes().all(|b| boundary(b) || b == b'_' || b == b'-') + || !value.bytes().next().is_some_and(boundary) + || !value.bytes().last().is_some_and(boundary) + { + return Err(Error::InvalidPath("Catalog names must be 1-64 lowercase ASCII letters, digits, '-' or '_', with alphanumeric boundaries".into())); + } + Ok(()) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CatalogOperation { + Config, + WarehouseShow, + WarehouseEnable, + NamespaceCreate, + NamespaceList, + NamespaceShow, + NamespaceExists, + NamespaceUpdate, + NamespaceRemove, + TableCreate, + TableRegister, + TableList, + TableShow, + TableExists, + TableRename, + TableRemove, + MetadataShow, + MetadataUpdate, + Commit, + RefList, + RefSet, + RefRemove, + ViewCreate, + ViewList, + ViewShow, + ViewExists, + ViewReplace, + ViewRemove, + MaintenancePlan, + MaintenanceRun, + MaintenanceConfigShow, + MaintenanceConfigSet, + MaintenanceJobShow, + SchedulerShow, + SchedulerRun, + WorkerRun, + JobHeartbeat, + JobQuarantine, + Diagnostics, + Export, + Import, + Recover, + Rollback, + ExternalShow, + ExternalSet, + ExternalSync, + MigrationStatus, + MigrationStart, + MigrationCancel, +} + +impl CatalogOperation { + pub const fn is_list(self) -> bool { + matches!(self, Self::NamespaceList | Self::TableList | Self::ViewList) + } +} + +#[derive(Clone, Debug)] +pub struct CatalogRequest { + pub operation: CatalogOperation, + pub target: CatalogTarget, + pub body: Option, + /// Reference or maintenance job identifier, encoded as one path segment. + pub child: Option, + pub page_size: u16, + pub page_token: Option, + pub single_page: bool, + pub snapshots: Option, +} + +impl CatalogRequest { + pub fn new(operation: CatalogOperation, target: CatalogTarget) -> Self { + Self { + operation, + target, + body: None, + child: None, + page_size: 1000, + page_token: None, + single_page: false, + snapshots: None, + } + } +} + +#[async_trait] +pub trait TableCatalogApi: Send + Sync { + /// Execute a catalog operation. Writes are never automatically retried. + async fn catalog(&self, request: &CatalogRequest) -> Result; +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 47d0c155..e70e2258 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -94,3 +94,5 @@ pub use undo::{ UndoAction, UndoObjectResult, UndoOutcome, UndoPlan, UndoPlanItem, plan_object_undo, }; pub use watch::{WatchApi, WatchEvent, WatchFrame, WatchRequest, WatchSource, WatchStream}; + +pub mod catalog; diff --git a/crates/core/tests/catalog.rs b/crates/core/tests/catalog.rs new file mode 100644 index 00000000..df5b527b --- /dev/null +++ b/crates/core/tests/catalog.rs @@ -0,0 +1,34 @@ +use rc_core::catalog::{CatalogTarget, ResourceKind}; + +#[test] +fn nested_namespace_is_not_an_object_key() { + let target = + CatalogTarget::parse("local/analytics/sales.eu/orders", ResourceKind::Table).unwrap(); + assert_eq!(target.namespace, vec!["sales", "eu"]); + assert_eq!(target.name.as_deref(), Some("orders")); +} + +#[test] +fn malformed_resources_are_rejected() { + for path in [ + "local/b/n/t/extra", + "local/b/n%2Fother/t", + "local/b/../t", + "local/b/n/", + "local/b/n/T", + ] { + assert!( + CatalogTarget::parse(path, ResourceKind::Table).is_err(), + "{path}" + ); + } +} + +#[test] +fn warehouse_preserves_bucket_dots_without_accepting_path_syntax() { + let target = CatalogTarget::parse("local/data.warehouse/ns/t", ResourceKind::Table).unwrap(); + assert_eq!(target.warehouse, "data.warehouse"); + for bucket in ["..", ".", "a?b", "a%2Fb", "a#b"] { + assert!(CatalogTarget::parse(&format!("local/{bucket}"), ResourceKind::Warehouse).is_err()); + } +} diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index ed4f8679..e7beecf4 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -11074,3 +11074,5 @@ mod tests { } } } + +mod catalog; diff --git a/crates/s3/src/admin/catalog.rs b/crates/s3/src/admin/catalog.rs new file mode 100644 index 00000000..c021a840 --- /dev/null +++ b/crates/s3/src/admin/catalog.rs @@ -0,0 +1,395 @@ +//! Iceberg REST transport using the existing alias TLS and SigV4 client. + +use super::{AdminClient, read_bounded_response_body}; +use async_trait::async_trait; +use rc_core::catalog::{CatalogOperation as Op, CatalogRequest, TableCatalogApi}; +use rc_core::{Error, Result}; +use reqwest::{Method, StatusCode}; +use serde_json::{Value, json}; +use std::collections::HashSet; + +#[cfg(test)] +mod tests { + use super::*; + use rc_core::catalog::{CatalogTarget, ResourceKind}; + #[test] + fn catalog_namespace_encoding_and_register_parent() { + let request = CatalogRequest::new( + Op::TableRegister, + CatalogTarget::parse("local/warehouse/a.b/table", ResourceKind::Table).unwrap(), + ); + let (method, path) = route(&request).unwrap(); + assert_eq!(method, Method::POST); + assert_eq!(path, "/warehouse/namespaces/a%1Fb/register"); + } + #[test] + fn catalog_ref_cannot_escape_path() { + let mut request = CatalogRequest::new( + Op::RefSet, + CatalogTarget::parse("local/warehouse/ns/table", ResourceKind::Table).unwrap(), + ); + request.child = Some("../bad".into()); + assert!(route(&request).is_err()); + } + #[test] + fn catalog_error_statuses_preserve_exit_classes() { + for (status, code) in [ + (400, 2), + (401, 4), + (403, 4), + (404, 5), + (406, 7), + (409, 6), + (412, 6), + (503, 3), + ] { + assert_eq!( + catalog_error(StatusCode::from_u16(status).unwrap(), "unsupported backing") + .exit_code(), + code + ); + } + } +} + +fn route(request: &CatalogRequest) -> Result<(Method, String)> { + let t = &request.target; + rc_core::catalog::validate_warehouse(&t.warehouse)?; + for segment in &t.namespace { + rc_core::catalog::validate_segment(segment)?; + } + if let Some(name) = &t.name { + rc_core::catalog::validate_segment(name)?; + } + let warehouse = format!("/{}", urlencoding::encode(&t.warehouse)); + let namespaces = format!("{warehouse}/namespaces"); + let ns = format!( + "{namespaces}/{}", + urlencoding::encode(&t.namespace.join("\u{1f}")) + ); + let tables = format!("{ns}/tables"); + let name = urlencoding::encode(t.name.as_deref().unwrap_or("")); + let table = format!("{tables}/{name}"); + let views = format!("{ns}/views"); + let view = format!("{views}/{name}"); + let child = request.child.as_deref().unwrap_or(""); + if matches!(request.operation, Op::RefSet | Op::RefRemove) { + rc_core::catalog::validate_segment(child)?; + } + if matches!( + request.operation, + Op::MaintenanceJobShow | Op::JobHeartbeat | Op::JobQuarantine + ) && (child.is_empty() + || child.len() > 256 + || child.contains(['/', '\\', '%']) + || child == "." + || child == "..") + { + return Err(Error::InvalidPath( + "Invalid maintenance job identifier".into(), + )); + } + let child = urlencoding::encode(child); + let (method, path) = match request.operation { + Op::Config => (Method::GET, "/config".into()), + Op::WarehouseShow => (Method::GET, format!("/buckets{}", warehouse)), + Op::WarehouseEnable => (Method::PUT, format!("/buckets{}", warehouse)), + Op::NamespaceCreate => (Method::POST, namespaces), + Op::NamespaceList => (Method::GET, namespaces), + Op::NamespaceShow => (Method::GET, ns), + Op::NamespaceExists => (Method::HEAD, ns), + Op::NamespaceUpdate => (Method::POST, format!("{ns}/properties")), + Op::NamespaceRemove => (Method::DELETE, ns), + Op::TableCreate => (Method::POST, tables), + Op::TableList => (Method::GET, tables), + Op::TableRegister => (Method::POST, format!("{ns}/register")), + Op::TableShow => (Method::GET, table), + Op::TableExists => (Method::HEAD, table), + Op::TableRename => (Method::POST, format!("{warehouse}/tables/rename")), + Op::TableRemove => (Method::DELETE, table), + Op::Commit => (Method::POST, table), + Op::MetadataShow => (Method::GET, format!("{table}/metadata-location")), + Op::MetadataUpdate => (Method::PUT, format!("{table}/metadata-location")), + Op::RefList => (Method::GET, format!("{table}/refs")), + Op::RefSet if ref_requires_absence(request) => (Method::POST, table), + Op::RefSet => (Method::PUT, format!("{table}/refs/{child}")), + Op::RefRemove => (Method::DELETE, format!("{table}/refs/{child}")), + Op::ViewCreate => (Method::POST, views), + Op::ViewList => (Method::GET, views), + Op::ViewShow => (Method::GET, view), + Op::ViewExists => (Method::HEAD, view), + Op::ViewReplace => (Method::POST, view), + Op::ViewRemove => (Method::DELETE, view), + Op::MaintenancePlan | Op::MaintenanceRun => { + (Method::POST, format!("{table}/maintenance/metadata")) + } + Op::MaintenanceConfigShow => (Method::GET, format!("{table}/maintenance/config")), + Op::MaintenanceConfigSet => (Method::PUT, format!("{table}/maintenance/config")), + Op::MaintenanceJobShow => (Method::GET, format!("{table}/maintenance/jobs/{child}")), + Op::JobHeartbeat => ( + Method::POST, + format!("{table}/maintenance/jobs/{child}/heartbeat"), + ), + Op::JobQuarantine => ( + Method::POST, + format!("{table}/maintenance/jobs/{child}/quarantine"), + ), + Op::SchedulerShow => (Method::GET, format!("{table}/maintenance/scheduler")), + Op::SchedulerRun => (Method::POST, format!("{table}/maintenance/scheduler/run")), + Op::WorkerRun => (Method::POST, format!("{table}/maintenance/worker/run")), + Op::Diagnostics => (Method::GET, format!("{table}/catalog/diagnostics")), + Op::Export => (Method::GET, format!("{table}/catalog/export")), + Op::Import => (Method::POST, format!("{table}/catalog/import")), + Op::Recover => (Method::POST, format!("{table}/catalog/recovery")), + Op::Rollback => (Method::POST, format!("{table}/catalog/rollback")), + Op::ExternalShow => (Method::GET, format!("{table}/catalog/external")), + Op::ExternalSet => (Method::PUT, format!("{table}/catalog/external")), + Op::ExternalSync => (Method::POST, format!("{table}/catalog/external/sync")), + Op::MigrationStatus => (Method::GET, format!("{warehouse}/catalog/migration")), + Op::MigrationStart => (Method::POST, format!("{warehouse}/catalog/migration")), + Op::MigrationCancel => (Method::DELETE, format!("{warehouse}/catalog/migration")), + }; + Ok((method, path)) +} + +fn ref_requires_absence(request: &CatalogRequest) -> bool { + request.operation == Op::RefSet + && request + .body + .as_ref() + .and_then(|v| v.get("expected-snapshot-id")) + .is_some_and(Value::is_null) +} +fn catalog_wire_body(request: &CatalogRequest) -> Option { + if !ref_requires_absence(request) { + return request.body.clone(); + } + // The ref endpoint deserializes null as None. A standard requirement keeps + // null explicit, so create-if-absent cannot overwrite an existing ref. + let mut update = request.body.clone()?; + let map = update.as_object_mut()?; + let commit_id = map.remove("commit-id"); + map.remove("expected-snapshot-id"); + map.insert("action".into(), json!("set-snapshot-ref")); + map.insert("ref-name".into(), json!(request.child)); + Some( + json!({"commit-id":commit_id,"requirements":[{"type":"assert-ref-snapshot-id","ref":request.child,"snapshot-id":null}],"updates":[update]}), + ) +} + +fn catalog_error(status: StatusCode, message: &str) -> Error { + let message = format!("Catalog HTTP {}: {}", status.as_u16(), message); + match status.as_u16() { + 400 | 422 => Error::Config(message), + 401 | 403 => Error::Auth(message), + 404 => Error::NotFound(message), + 406 | 501 => Error::UnsupportedFeature(message), + 409 | 412 => Error::Conflict(message), + 408 | 429 | 500 | 502 | 503 | 504 => Error::Network(message), + _ => Error::General(message), + } +} + +// Credentials belong to protocol configuration maps, not arbitrary metadata keys. +fn remove_credentials(value: &mut Value) { + let Some(response) = value.as_object_mut() else { + return; + }; + response.remove("storage-credentials"); + for field in ["config", "defaults", "overrides"] { + if let Some(config) = response.get_mut(field).and_then(Value::as_object_mut) { + for key in [ + "s3.access-key-id", + "s3.secret-access-key", + "s3.session-token", + ] { + config.remove(key); + } + } + } +} + +impl AdminClient { + async fn catalog_page(&self, request: &CatalogRequest) -> Result { + let (method, path) = route(request)?; + let mut url = url::Url::parse(&format!("{}/iceberg/v1{path}", self.endpoint))?; + { + let mut query = url.query_pairs_mut(); + if request.operation == Op::Config { + query.append_pair("warehouse", &request.target.warehouse); + } + if request.operation == Op::NamespaceList && !request.target.namespace.is_empty() { + query.append_pair("parent", &request.target.namespace.join("\u{1f}")); + } + if request.operation.is_list() { + query.append_pair("pageSize", &request.page_size.to_string()); + if let Some(token) = &request.page_token { + query.append_pair("pageToken", token); + } + } + if let Some(snapshots) = &request.snapshots { + query.append_pair("snapshots", snapshots); + } + if request.operation == Op::TableRemove { + query.append_pair("purgeRequested", "false"); + } + } + let wire_body = catalog_wire_body(request); + let body = wire_body + .as_ref() + .map(serde_json::to_vec) + .transpose()? + .unwrap_or_default(); + if body.len() > 8 * 1024 * 1024 { + return Err(Error::Config("Catalog request exceeds 8 MiB".into())); + } + let headers = self.request_headers(&body)?; + let headers = self + .sign_request(&method, url.as_str(), &headers, &body) + .await?; + let write = !matches!(method, Method::GET | Method::HEAD); + let response = self + .http_client + .request(method.clone(), url) + .headers(headers) + .body(body) + .send() + .await + .map_err(|_| { + Error::Network( + if write { + "Catalog mutation outcome unknown; request was not retried" + } else { + "Catalog request failed" + } + .into(), + ) + })?; + let status = response.status(); + if !status.is_success() { + let bytes = read_bounded_response_body(response, 64 * 1024, "Catalog error").await?; + let parsed = serde_json::from_slice::(&bytes).ok(); + let mut message = parsed + .as_ref() + .and_then(|v| v.pointer("/error/message")) + .and_then(Value::as_str) + .unwrap_or("Server rejected catalog operation") + .chars() + .take(2048) + .collect::(); + self.redact_admin_credentials(&mut message); + // Do not echo credential-bearing server diagnostics. + if ["secret", "credential", "authorization", "token="] + .iter() + .any(|word| message.to_ascii_lowercase().contains(word)) + { + message = + "Server rejected catalog operation; check permissions and configuration".into(); + } + return Err(catalog_error(status, &message)); + } + if method == Method::HEAD { + return Ok(json!({"exists": true})); + } + if status == StatusCode::NO_CONTENT { + return Ok(json!({})); + } + let bytes = + read_bounded_response_body(response, 64 * 1024 * 1024, "Catalog response").await?; + let mut value: Value = if bytes.is_empty() { + json!({}) + } else { + serde_json::from_slice(&bytes) + .map_err(|_| Error::General("Invalid catalog JSON response".into()))? + }; + if !value.is_object() + && !(request.operation == Op::MaintenanceConfigShow && value.is_null()) + { + return Err(Error::General("Catalog response must be an object".into())); + } + if request.operation.is_list() { + let field = if request.operation == Op::NamespaceList { + "namespaces" + } else { + "identifiers" + }; + if !value.get(field).is_some_and(Value::is_array) { + return Err(Error::General(format!("Catalog listing missing {field}"))); + } + if value + .get("next-page-token") + .is_some_and(|token| !token.is_null() && !token.is_string()) + { + return Err(Error::General("Invalid catalog pagination token".into())); + } + } + remove_credentials(&mut value); + self.redact_admin_credentials_in_value(&mut value); + Ok(value) + } +} + +#[async_trait] +impl TableCatalogApi for AdminClient { + async fn catalog(&self, request: &CatalogRequest) -> Result { + if !(1..=1000).contains(&request.page_size) { + return Err(Error::Config("page-size must be between 1 and 1000".into())); + } + let mut next = request.clone(); + let mut result = self.catalog_page(&next).await?; + if !request.operation.is_list() || request.single_page { + return Ok(result); + } + let field = if request.operation == Op::NamespaceList { + "namespaces" + } else { + "identifiers" + }; + let mut seen = HashSet::new(); + if let Some(token) = &request.page_token { + seen.insert(token.clone()); + } + let mut bytes = serde_json::to_vec(&result)?.len(); + loop { + if !result.get(field).is_some_and(Value::is_array) { + return Err(Error::General(format!("Catalog listing missing {field}"))); + } + let token = match result.get("next-page-token") { + None | Some(Value::Null) => break, + Some(Value::String(token)) if token.is_empty() => break, + Some(Value::String(token)) => token.clone(), + _ => return Err(Error::General("Invalid catalog pagination token".into())), + }; + if token.len() > 16 * 1024 || !seen.insert(token.clone()) || seen.len() > 10000 { + return Err(Error::General( + "Catalog pagination did not make bounded progress".into(), + )); + } + next.page_token = Some(token); + let page = self.catalog_page(&next).await?; + bytes = bytes.saturating_add(serde_json::to_vec(&page)?.len()); + if bytes > 64 * 1024 * 1024 { + return Err(Error::General( + "Catalog listing exceeds 64 MiB; use --no-paginate".into(), + )); + } + let items = page + .get(field) + .and_then(Value::as_array) + .ok_or_else(|| Error::General(format!("Catalog listing missing {field}")))?; + result + .get_mut(field) + .and_then(Value::as_array_mut) + .ok_or_else(|| Error::General("Invalid catalog listing".into()))? + .extend(items.iter().cloned()); + result["next-page-token"] = page.get("next-page-token").cloned().unwrap_or(Value::Null); + } + if let Some(map) = result.as_object_mut() { + map.remove("next-page-token"); + } + Ok(result) + } +} + +#[cfg(test)] +mod transport_tests; diff --git a/crates/s3/src/admin/catalog/transport_tests.rs b/crates/s3/src/admin/catalog/transport_tests.rs new file mode 100644 index 00000000..365586c0 --- /dev/null +++ b/crates/s3/src/admin/catalog/transport_tests.rs @@ -0,0 +1,285 @@ +use super::*; +use rc_core::{ + Alias, + catalog::{CatalogTarget, ResourceKind}, +}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +async fn server( + responses: Vec<(u16, Value)>, +) -> (AdminClient, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let mut requests = Vec::new(); + for (status, body) in responses { + let (mut stream, _) = + tokio::time::timeout(std::time::Duration::from_secs(10), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut bytes = Vec::new(); + loop { + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).await.unwrap(); + assert!(n > 0); + bytes.extend_from_slice(&buf[..n]); + if let Some(end) = bytes.windows(4).position(|v| v == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&bytes[..end]); + let length = headers + .lines() + .find_map(|l| { + l.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|s| s.trim().parse::().unwrap()) + }) + .unwrap_or(0); + if bytes.len() >= end + 4 + length { + break; + } + } + } + requests.push(String::from_utf8(bytes).unwrap()); + let body = if status == 204 { + String::new() + } else { + body.to_string() + }; + stream.write_all(format!("HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",body.len()).as_bytes()).await.unwrap(); + } + requests + }); + let mut client = + AdminClient::new(&Alias::new("a", &endpoint, "test-access", "test-secret")).unwrap(); + client.http_client = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(std::time::Duration::from_secs(5)) + .build() + .unwrap(); + (client, task) +} +fn request(op: Op) -> CatalogRequest { + CatalogRequest::new( + op, + CatalogTarget::parse("a/warehouse/sales.eu/orders", ResourceKind::Table).unwrap(), + ) +} + +#[tokio::test] +async fn catalog_transport_signs_pages_and_preserves_opaque_tokens() { + let (client, server) = server(vec![ + ( + 200, + json!({"identifiers":[{"name":"one"}],"next-page-token":"a+/="}), + ), + ( + 200, + json!({"identifiers":[{"name":"two"}],"next-page-token":null}), + ), + ]) + .await; + let result = client.catalog(&request(Op::TableList)).await.unwrap(); + assert_eq!(result["identifiers"].as_array().unwrap().len(), 2); + assert!(result.get("next-page-token").is_none()); + let requests = server.await.unwrap(); + assert!( + requests[0] + .starts_with("GET /iceberg/v1/warehouse/namespaces/sales%1Feu/tables?pageSize=1000 ") + ); + assert!(requests[1].contains("pageToken=a%2B%2F%3D")); + assert!( + requests[0] + .to_ascii_lowercase() + .contains("authorization: aws4-hmac-sha256") + ); + assert!(requests[0].contains("/s3/aws4_request")); +} +#[tokio::test] +async fn catalog_transport_single_page_and_later_failure() { + let (client, task) = server(vec![( + 200, + json!({"namespaces":[],"next-page-token":"next"}), + )]) + .await; + let mut req = request(Op::NamespaceList); + req.single_page = true; + assert_eq!( + client.catalog(&req).await.unwrap()["next-page-token"], + "next" + ); + task.await.unwrap(); + let (client, task) = server(vec![ + ( + 200, + json!({"identifiers":[{"name":"one"}],"next-page-token":"next"}), + ), + (403, json!({"error":{"message":"permission denied"}})), + ]) + .await; + assert_eq!( + client + .catalog(&request(Op::TableList)) + .await + .unwrap_err() + .exit_code(), + 4 + ); + task.await.unwrap(); +} +#[tokio::test] +async fn catalog_transport_rejects_repeated_and_missing_page_fields() { + let (client, task) = server(vec![ + (200, json!({"identifiers":[],"next-page-token":"same"})), + (200, json!({"identifiers":[],"next-page-token":"same"})), + ]) + .await; + assert!( + client + .catalog(&request(Op::TableList)) + .await + .unwrap_err() + .to_string() + .contains("progress") + ); + task.await.unwrap(); + let (client, task) = server(vec![(200, json!({"next-page-token":null}))]).await; + assert!(client.catalog(&request(Op::TableList)).await.is_err()); + task.await.unwrap(); +} +#[tokio::test] +async fn catalog_transport_conflict_is_not_retried_and_body_keeps_guards() { + let (client, task) = server(vec![(409, json!({"error":{"message":"stale version"}}))]).await; + let mut req = request(Op::Commit); + req.body = Some( + json!({"expected-version-token":"v1","expected-metadata-location":"s3://warehouse/m1","commit-id":"attempt-1","updates":[]}), + ); + assert_eq!(client.catalog(&req).await.unwrap_err().exit_code(), 6); + let requests = task.await.unwrap(); + assert_eq!(requests.len(), 1); + let body: Value = serde_json::from_str(requests[0].split("\r\n\r\n").nth(1).unwrap()).unwrap(); + assert_eq!(Some(body), req.body); +} +#[tokio::test] +async fn catalog_transport_empty_delete_and_credential_redaction() { + let (client, task) = server(vec![(204, Value::Null)]).await; + assert_eq!( + client.catalog(&request(Op::TableRemove)).await.unwrap(), + json!({}) + ); + assert!(task.await.unwrap()[0].contains("purgeRequested=false")); + let (client,task)=server(vec![(200,json!({"metadata":{"snapshots":[]},"storage-credentials":[{"access-key-id":"temporary"}],"config":{"s3.secret-access-key":"temporary-secret","s3.session-token":"temporary-token","visible":"test-secret"}}))]).await; + let data = client + .catalog(&request(Op::TableShow)) + .await + .unwrap() + .to_string(); + assert!(!data.contains("temporary")); + assert!(!data.contains("test-secret")); + task.await.unwrap(); +} +#[tokio::test] +async fn catalog_transport_preserves_permission_and_backing_errors() { + for (status, code, message) in [ + (403, 4, "table action denied"), + (406, 7, "requires object-backed catalog"), + (404, 5, "table not found"), + ] { + let (client, task) = server(vec![(status, json!({"error":{"message":message}}))]).await; + let e = client.catalog(&request(Op::TableShow)).await.unwrap_err(); + assert_eq!(e.exit_code(), code); + assert!(e.to_string().contains(message)); + task.await.unwrap(); + } +} + +#[tokio::test] +async fn catalog_ref_create_preserves_absence_requirement() { + let (client, task) = server(vec![(409, json!({"error":{"message":"ref exists"}}))]).await; + let mut req = request(Op::RefSet); + req.child = Some("release".into()); + req.body = Some( + json!({"snapshot-id":7,"expected-snapshot-id":null,"type":"tag","commit-id":"ref-create"}), + ); + assert_eq!(client.catalog(&req).await.unwrap_err().exit_code(), 6); + let requests = task.await.unwrap(); + assert!( + requests[0].starts_with("POST /iceberg/v1/warehouse/namespaces/sales%1Feu/tables/orders") + ); + let body: Value = serde_json::from_str(requests[0].split("\r\n\r\n").nth(1).unwrap()).unwrap(); + assert_eq!( + body["requirements"], + json!([{"type":"assert-ref-snapshot-id","ref":"release","snapshot-id":null}]) + ); + assert_eq!(body["updates"][0]["ref-name"], "release"); +} + +#[tokio::test] +async fn catalog_redaction_preserves_refs_schema_and_properties() { + let metadata = json!({ + "refs":{"password-audit":{"snapshot-id":7,"type":"tag"}}, + "schemas":[{"fields":[{"id":1,"name":"password","type":"string"}]}], + "properties":{"authorization-mode":"strict","password-policy":"required"} + }); + let (client, task) = server(vec![(200, json!({ + "metadata":metadata, + "refs":{"password-audit":{"snapshot-id":7,"type":"tag"}}, + "user-defined-ref-count":1, + "properties":{"authorization-mode":"strict"}, + "storage-credentials":[{"config":{"s3.secret-access-key":"vended-secret"}}], + "config":{"s3.access-key-id":"vended-access","s3.secret-access-key":"vended-secret","s3.session-token":"vended-token","s3.region":"us-east-1"}, + "defaults":{"s3.secret-access-key":"default-secret","warehouse":"warehouse"}, + "overrides":{"s3.session-token":"override-token","s3.endpoint":"http://localhost"} + }))]).await; + let result = client.catalog(&request(Op::TableShow)).await.unwrap(); + assert_eq!(result["metadata"], metadata); + assert_eq!(result["refs"]["password-audit"]["snapshot-id"], 7); + assert_eq!(result["properties"]["authorization-mode"], "strict"); + assert_eq!(result["config"], json!({"s3.region":"us-east-1"})); + assert_eq!(result["defaults"], json!({"warehouse":"warehouse"})); + assert_eq!( + result["overrides"], + json!({"s3.endpoint":"http://localhost"}) + ); + assert!(result.get("storage-credentials").is_none()); + task.await.unwrap(); +} + +#[tokio::test] +async fn catalog_namespace_pages_preserve_encoded_parent() { + let (client, task) = server(vec![ + ( + 200, + json!({"namespaces":[["sales","eu","one"]],"next-page-token":"a+/="}), + ), + (200, json!({"namespaces":[["sales","eu","two"]]})), + ]) + .await; + let req = CatalogRequest::new( + Op::NamespaceList, + CatalogTarget::parse("a/warehouse/sales.eu", ResourceKind::Namespace).unwrap(), + ); + let result = client.catalog(&req).await.unwrap(); + assert_eq!(result["namespaces"].as_array().unwrap().len(), 2); + for (index, raw) in task.await.unwrap().iter().enumerate() { + let target = raw.split_whitespace().nth(1).unwrap(); + let url = url::Url::parse(&format!("http://localhost{target}")).unwrap(); + assert_eq!(url.path(), "/iceberg/v1/warehouse/namespaces"); + assert_eq!( + url.query_pairs() + .find(|(key, _)| key == "parent") + .unwrap() + .1, + "sales\u{1f}eu" + ); + if index == 1 { + assert_eq!( + url.query_pairs() + .find(|(key, _)| key == "pageToken") + .unwrap() + .1, + "a+/=" + ); + } + } +} diff --git a/docs/usage/table-catalog.md b/docs/usage/table-catalog.md new file mode 100644 index 00000000..0682a091 --- /dev/null +++ b/docs/usage/table-catalog.md @@ -0,0 +1,154 @@ +# RustFS table catalog commands + +`rc table` manages the Iceberg catalog. `rc admin table` manages catalog enablement, maintenance, recovery and backing migration. These commands target the RustFS server contract at `22bff27aee9cfcb8054c9c2ad5b8b91a7376430a`; operations can still be restricted by permissions, configuration and catalog backing. They do not implement AIStor-specific APIs. + +## Connection and resource names + +Configure the usual alias with `rc alias set`. The catalog uses that alias's endpoint, credentials, region, TLS trust and client certificate settings. Requests use SigV4 service `s3` and append `/iceberg/v1` to the endpoint. Catalog access does not request or print vended credentials. There is no separate persistent catalog profile. + +Resources are `alias/warehouse`, `alias/warehouse/namespace`, and `alias/warehouse/namespace/table`. Namespace components are dot-separated at the command line, for example `local/analytics/sales.eu/orders`; the transport encodes the ordered components using the REST unit separator. Names are not object keys. Warehouse names are existing bucket names. Each namespace, table and ref component accepts up to 64 lowercase ASCII letters, digits, hyphens or underscores, with alphanumeric boundaries; namespace length is at most 512 bytes. Views use the same resource shape as tables. + +## Basic workflow + +```sh +rc bucket create local/analytics +rc admin table warehouse enable local/analytics +rc table warehouse show local/analytics +rc table config local/analytics +rc table namespace create local/analytics/sales --property owner=analytics +rc table create local/analytics/sales/orders --schema-file schema.json +rc table list local/analytics/sales +rc table show local/analytics/sales/orders --json +rc table metadata show local/analytics/sales/orders +rc table rename local/analytics/sales/orders local/analytics/sales/orders_v2 +rc table remove local/analytics/sales/orders_v2 +rc table namespace remove local/analytics/sales +``` + +Example `schema.json`: + +```json +{"type":"struct","fields":[{"id":1,"name":"id","type":"long","required":true}]} +``` + +`create` also accepts `--partition-spec-file`, `--sort-order-file`, `--location`, repeatable `--property key=value`, and `--format-version 1|2`. Locations must pass the server's warehouse validation. `register TARGET --metadata-location LOCATION` registers existing metadata without overwrite. Rename must stay within the same alias and warehouse. Drop preserves data files; there is no `--purge` or recursive namespace removal. Create a normal bucket before enabling catalog support; enablement does not call CreateBucket. + +Use a compatible engine such as PyIceberg, DuckDB or Spark to write and query rows. `rc sql` continues to mean S3 Select on an object. + +## Read and namespace operations + +- `namespace list WAREHOUSE` lists root namespaces; `namespace list NAMESPACE` lists its direct children, for example `rc table namespace list local/analytics/sales.eu`. `show NAMESPACE`, `exists NAMESPACE`, and `remove NAMESPACE` address an individual namespace. +- `namespace update NAMESPACE --set key=value --remove key` changes properties; the same key cannot occur in both sets. +- `table list NAMESPACE`, `show TABLE --snapshots all|refs`, `exists TABLE`. +- `snapshot list TABLE --snapshots all|refs` and `snapshot show TABLE SNAPSHOT_ID` project LoadTable metadata; they are not a separate server scan or SQL query. +- `metadata show TABLE` returns metadata location, version token and generation for pointer operations. +- `exists` succeeds when present and returns exit 5 for an absent resource. + +Namespace, table and view lists retrieve all server pages by default. `--page-size` accepts 1 through 1000. `--no-paginate` returns one page with its opaque `next-page-token`; pass that unchanged through `--page-token` to resume. A later page failure fails the whole command before printing partial results. Each response and accumulated list is limited to 64 MiB; use single-page mode for larger listings. Snapshot projections do not have server-side pagination. + +## Commits and refs + +There are two distinct commit contracts. The CLI never automatically retries a mutation or refreshes stale conditions to force it through. + +### Standard metadata updates + +```sh +rc table commit local/analytics/sales/orders --file update.json --commit-id property-edit-1 +``` + +```json +{ + "requirements": [{"type":"assert-ref-snapshot-id","ref":"main","snapshot-id":123}], + "updates": [{"action":"set-properties","updates":{"owner":"analytics"}}] +} +``` + +Supply Iceberg requirements that protect the state used to prepare the update. The command requires a nonempty requirements array. The current RustFS standard update path uses these requirements and its internal publication CAS; it does not enforce the optional external expected-version/location fields. Therefore the CLI rejects those fields or flags in this mode. A requirements failure is a conflict; review the current state before preparing a new operation. + +### Existing metadata pointer commit + +```sh +rc table commit local/analytics/sales/orders --file pointer.json \ + --expected-version-token CURRENT_TOKEN \ + --expected-metadata-location CURRENT_LOCATION \ + --commit-id pointer-edit-1 +``` + +```json +{"new-metadata-location":"s3://analytics/PATH/TO/VALIDATED/METADATA.json"} +``` + +Use metadata that already exists and satisfies the server's metadata-directory, schema, object-reference and warehouse checks. This mode requires a nonempty version token and old metadata location, either in the file or flags; conflicting values and nonempty metadata updates are rejected. Reuse the same commit ID and exact request after an uncertain outcome. This is not a promise of idempotency for every catalog API. + +```sh +rc table ref list local/analytics/sales/orders +rc table ref set local/analytics/sales/orders release --type tag \ + --snapshot-id 123 --expected-snapshot-id null --commit-id release-create-1 +rc table ref remove local/analytics/sales/orders release \ + --expected-snapshot-id 123 --commit-id release-remove-1 +``` + +`ref set` accepts optional `--min-snapshots-to-keep`, `--max-snapshot-age-ms`, `--max-ref-age-ms`. `null` requires the ref to be absent: the transport sends an explicit standard Iceberg requirement, because the dedicated RustFS ref endpoint treats JSON null as an omitted optional field. Numeric expectations use the ref endpoint. `ref remove --force` allows a retained ref to be removed but does not bypass the server's protection of `main`. + +## Views + +`view create TARGET --file view.json`, `view list NAMESPACE`, `view show TARGET`, `view exists TARGET`, `view replace TARGET --file update.json`, and `view remove TARGET` are available. The create target supplies the name. Example create body: + +```json +{ + "schema":{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","type":"long","required":true}]}, + "view-version":{"version-id":1,"schema-id":0,"timestamp-ms":1788652800000,"summary":{},"representations":[{"type":"sql","sql":"SELECT id FROM sales.orders","dialect":"spark"}],"default-namespace":["sales"]}, + "properties":{} +} +``` + +Replace takes the server's view `requirements`/`updates` body and requires `expected-metadata-location` in that file. Read the current view response before preparing it; LoadView does not expose a version token. The server checks the expected metadata location during conditional publication. View rename/register are not exposed because the referenced RustFS server has no corresponding route. + +## Administrative operations + +Administrative mutations below require `--yes`; they still enforce server authorization and backing-specific checks. `--file` reads a JSON object, or stdin with `--file -`, bounded to 8 MiB. Server JSON fields use their original spelling and are not converted from CLI flags. + +| Command below `rc admin table` | Request body / behavior | +| --- | --- | +| `maintenance plan TABLE --file FILE` | Metadata-maintenance request; forcibly sets `delete`, `commit-snapshot-expiration`, `commit-compaction` to false. | +| `maintenance run TABLE --file FILE --yes` | Sends the requested maintenance actions; only explicitly true flags apply changes. | +| `maintenance config show TABLE` | Loads current configuration, which may be absent. | +| `maintenance config set TABLE --file FILE --yes` | `TableMaintenanceConfig` object; inspect current configuration first. | +| `maintenance scheduler show TABLE` | Reads scheduler/job state. | +| `maintenance scheduler run TABLE --yes` | Runs one scheduling pass using the server's default scheduler identity. | +| `maintenance worker run TABLE --file FILE --yes` | `{}` or `{"worker-id":"operator-worker"}`; runs one worker pass. | +| `maintenance job show TABLE JOB` | Loads one report. | +| `maintenance job heartbeat TABLE JOB --file FILE --yes` | `{"lease-id":"CURRENT_LEASE","worker-id":"CURRENT_WORKER"}`. | +| `maintenance job quarantine TABLE JOB --file FILE --yes` | Server quarantine action and optional reason. | +| `catalog diagnostics TABLE` | Reads recovery/consistency state. | +| `catalog export TABLE` | Returns the server export document for inspection/backup. | +| `catalog import TABLE --file FILE --yes` | `{"metadata-location":"...","properties":{}}`; this is a metadata import request, not direct replay of the export envelope. | +| `catalog recover TABLE --yes` | Repairs commit finalization; no input file. | +| `catalog rollback TABLE --file FILE --yes` | `metadata-location`, `version-token`, optional stable `commit-id`; only forward-safe rollback is supported by the server. | +| `catalog metadata-update TABLE --file FILE --yes` | `metadata-location`, `version-token`, optional stable `commit-id`. | +| `catalog external show TABLE` | Reads external bridge configuration. | +| `catalog external set TABLE --file FILE --yes` | Server bridge configuration, including catalog, external-namespace and external-table. | +| `catalog external sync TABLE --file FILE --yes` | Server sync request including metadata-location and any required expected state; no vendor polling. | +| `migration status WAREHOUSE` | Reads migration readiness and blockers. | +| `migration start WAREHOUSE --yes` | Materializes the server-managed migration target. | +| `migration cancel WAREHOUSE --yes` | Requests cancellation; server can reject an advanced target. | + +Minimal maintenance preview input: + +```json +{"retain-recent-metadata-files":5,"snapshot-expiration":{"min-snapshots-to-keep":2,"max-snapshot-age-ms":604800000}} +``` + +Minimal maintenance config input: + +```json +{"version":1,"retain-recent-metadata-files":5,"delete-enabled":false,"background-enabled":false} +``` + +The `background-enabled` server configuration does not make this CLI a daemon or supply a built-in periodic server scheduler. Refer to the [pinned RustFS request definitions](https://github.com/rustfs/rustfs/blob/22bff27aee9cfcb8054c9c2ad5b8b91a7376430a/rustfs/src/admin/handlers/table_catalog/mod.rs) and [maintenance models](https://github.com/rustfs/rustfs/blob/22bff27aee9cfcb8054c9c2ad5b8b91a7376430a/rustfs/src/table_catalog/model.rs) for advanced request fields. Unsupported actions fail explicitly; the CLI does not emulate them by modifying reserved S3 objects. + +## Output and errors + +`--json` uses the existing output v3 envelope with `type: table_catalog`. Success is written to stdout and errors to stderr. Success `data` contains `operation` and the server `result` (or snapshot projection). Use `.data.result` when consuming JSON. Human output prints readable JSON for nested catalog documents. Top-level `storage-credentials` bundles and credential keys in the protocol `config`, `defaults`, and `overrides` maps are removed. Business properties, schema fields and ref names are preserved, including names containing words such as `password` or `authorization`. Known alias credential values remain redacted. + +Exit codes: 2 invalid arguments/request, 3 network/transient service error, 4 authentication/authorization, 5 not found, 6 conflict/precondition failure, 7 unsupported operation. Mutations with a lost response have an unknown outcome and are not retried automatically. Error envelopes conservatively report `retryable: false`; callers must decide whether a read retry or an exact commit replay is safe. No automatic cross-warehouse rename, purge, overwrite registration, staged create, v3, multi-table commit, table replication, table encryption policy or Delta Sharing is offered. diff --git a/schemas/output_v3.json b/schemas/output_v3.json index c7e7f072..d24710fa 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -25,6 +25,7 @@ "type": "string", "enum": [ "capabilities", + "table_catalog", "versioned_objects", "locks", "multipart_uploads", @@ -2046,6 +2047,12 @@ } ] }, + { + "allOf": [ + { "$ref": "#/definitions/successEnvelope" }, + { "properties": { "type": { "const": "table_catalog" }, "data": { "type": "object" } } } + ] + }, { "$ref": "#/definitions/errorOutput" } ] }