diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index c656a2f32..e61a2fc7a 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -40,6 +40,16 @@ const CREDENTIAL_VENDING_UNSUPPORTED: &str = "unsupported"; const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces"; const TABLE_CATALOG_TABLE_RESOURCE_ROOT: &str = "tables"; const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ + "GET /v1/{prefix}/namespaces", + "POST /v1/{prefix}/namespaces", + "GET /v1/{prefix}/namespaces/{namespace}", + "DELETE /v1/{prefix}/namespaces/{namespace}", + "GET /v1/{prefix}/namespaces/{namespace}/tables", + "POST /v1/{prefix}/namespaces/{namespace}/tables", + "POST /v1/{prefix}/namespaces/{namespace}/register", + "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}", + "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}", + "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}", "PUT /buckets/{warehouse}", "GET /buckets/{warehouse}", "GET /{warehouse}/namespaces", @@ -133,6 +143,8 @@ struct CreateTableRequest { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct RestCommitTableRequest { + #[serde(default, rename = "identifier")] + _identifier: Option, #[serde(default, rename = "commit-id")] commit_id: Option, #[serde(default, rename = "idempotency-key")] @@ -2434,6 +2446,12 @@ mod tests { Some(&TABLE_CATALOG_COMPAT_PREFIX) ); assert!(response.overrides.is_empty()); + assert!(response.endpoints.contains(&"GET /v1/{prefix}/namespaces")); + assert!( + response + .endpoints + .contains(&"GET /v1/{prefix}/namespaces/{namespace}/tables/{table}") + ); assert!(response.endpoints.contains(&"GET /{warehouse}/namespaces")); assert!(response.endpoints.contains(&"POST /{warehouse}/namespaces")); assert!( @@ -4020,6 +4038,7 @@ mod tests { new_metadata_location: Some(next_metadata_location.to_string()), requirements: Vec::new(), updates: Vec::new(), + _identifier: None, writer: Some("pyiceberg".to_string()), }, ) @@ -4756,6 +4775,7 @@ mod tests { new_metadata_location: Some(mismatched_location.to_string()), requirements: Vec::new(), updates: Vec::new(), + _identifier: None, writer: Some("pyiceberg".to_string()), }, ) diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index b11ddd581..432affa69 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -38,7 +38,7 @@ use rustfs_ecstore::disk::RUSTFS_META_BUCKET; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::{ set_disk::get_lock_acquire_timeout, - store_api::{HTTPPreconditions, ListOperations, NamespaceLocking, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader}, + store_api::{HTTPPreconditions, ObjectOptions, PutObjReader, StorageAPI}, }; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use time::{Duration, OffsetDateTime}; @@ -400,6 +400,10 @@ pub(crate) enum TableCatalogPutPrecondition { pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static { async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult>; + async fn read_object_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult> { + self.read_object(bucket, object).await + } + async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult; async fn put_object( @@ -410,6 +414,16 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static { precondition: TableCatalogPutPrecondition, ) -> TableCatalogStoreResult<()>; + async fn put_object_unlocked( + &self, + bucket: &str, + object: &str, + data: Vec, + precondition: TableCatalogPutPrecondition, + ) -> TableCatalogStoreResult<()> { + self.put_object(bucket, object, data, precondition).await + } + async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()>; async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult>; @@ -548,7 +562,39 @@ where where T: DeserializeOwned, { - let Some(object_data) = self.backend.read_object(bucket, object).await? else { + self.read_entry_with(bucket, object, |backend, bucket, object| { + Box::pin(async move { backend.read_object(bucket, object).await }) + }) + .await + } + + async fn read_entry_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult)>> + where + T: DeserializeOwned, + { + self.read_entry_with(bucket, object, |backend, bucket, object| { + Box::pin(async move { backend.read_object_unlocked(bucket, object).await }) + }) + .await + } + + async fn read_entry_with<'a, T, F>( + &'a self, + bucket: &'a str, + object: &'a str, + read_object: F, + ) -> TableCatalogStoreResult)>> + where + T: DeserializeOwned, + F: FnOnce( + &'a B, + &'a str, + &'a str, + ) -> std::pin::Pin< + Box>> + Send + 'a>, + >, + { + let Some(object_data) = read_object(&self.backend, bucket, object).await? else { return Ok(None); }; @@ -572,6 +618,21 @@ where self.backend.put_object(bucket, object, data, precondition).await } + async fn write_entry_unlocked( + &self, + bucket: &str, + object: &str, + entry: &T, + precondition: TableCatalogPutPrecondition, + ) -> TableCatalogStoreResult<()> + where + T: Serialize, + { + let data = serde_json::to_vec(entry) + .map_err(|err| TableCatalogStoreError::Internal(format!("failed to serialize catalog entry {object}: {err}")))?; + self.backend.put_object_unlocked(bucket, object, data, precondition).await + } + async fn require_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> { if self.get_table_bucket(table_bucket).await?.is_none() { return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); @@ -595,6 +656,25 @@ where Ok(Some((entry, etag))) } + async fn read_table_with_etag_unlocked( + &self, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + ) -> TableCatalogStoreResult> { + let table_path = self.paths.table_entry_path(table_bucket, namespace, table); + let Some((entry, etag)) = self + .read_entry_unlocked::(self.catalog_bucket(), &table_path) + .await? + else { + return Ok(None); + }; + let Some(etag) = etag else { + return Err(TableCatalogStoreError::Internal(format!("catalog table entry has no etag: {table_path}"))); + }; + Ok(Some((entry, etag))) + } + async fn write_table_entry( &self, entry: TableEntry, @@ -967,7 +1047,7 @@ where let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; - let Some((entry, _)) = self.read_table_with_etag(table_bucket, &namespace, &table).await? else { + let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else { return Err(TableCatalogStoreError::NotFound(format!( "table {}/{}/{}", table_bucket, @@ -1163,7 +1243,10 @@ where let table_path = self.paths.table_entry_path(&request.table_bucket, &namespace, &table); let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; - let Some((current, current_etag)) = self.read_table_with_etag(&request.table_bucket, &namespace, &table).await? else { + let Some((current, current_etag)) = self + .read_table_with_etag_unlocked(&request.table_bucket, &namespace, &table) + .await? + else { return Err(TableCatalogStoreError::NotFound(format!( "table {}/{}/{}", request.table_bucket, request.namespace, request.table @@ -1289,7 +1372,7 @@ where .await?; } - self.write_entry( + self.write_entry_unlocked( self.catalog_bucket(), &table_path, &next, @@ -1364,7 +1447,7 @@ impl Clone for EcStoreTableCatalogObjectBackend { impl EcStoreTableCatalogObjectBackend where - S: ObjectIO + ObjectOperations + ListOperations + NamespaceLocking, + S: StorageAPI, { pub fn new(store: Arc) -> Self { Self { store } @@ -1376,34 +1459,22 @@ pub(crate) type EcStoreTableCatalogStore = ObjectTableCatalogStore TableCatalogObjectBackend for EcStoreTableCatalogObjectBackend where - S: ObjectIO + ObjectOperations + ListOperations + NamespaceLocking, + S: StorageAPI, { async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult> { - let info = match self.store.get_object_info(bucket, object, &ObjectOptions::default()).await { - Ok(info) => info, - Err(err) if is_missing_storage_error(&err) => return Ok(None), - Err(err) => return Err(storage_error_to_catalog("read catalog object info", err)), - }; - let mut reader = match self - .store - .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) - .await - { - Ok(reader) => reader, - Err(err) if is_missing_storage_error(&err) => return Ok(None), - Err(err) => return Err(storage_error_to_catalog("read catalog object", err)), - }; - let mut data = Vec::new(); - reader - .stream - .read_to_end(&mut data) - .await - .map_err(|err| TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}")))?; - Ok(Some(TableCatalogObject { - data, - etag: info.etag, - mod_time: info.mod_time, - })) + self.read_object_with_options(bucket, object, ObjectOptions::default()).await + } + + async fn read_object_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult> { + self.read_object_with_options( + bucket, + object, + ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await } async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult { @@ -1421,16 +1492,17 @@ where data: Vec, precondition: TableCatalogPutPrecondition, ) -> TableCatalogStoreResult<()> { - let mut reader = PutObjReader::from_vec(data); - let opts = ObjectOptions { - http_preconditions: http_preconditions_for_catalog_put(precondition), - ..Default::default() - }; - self.store - .put_object(bucket, object, &mut reader, &opts) - .await - .map(|_| ()) - .map_err(|err| storage_error_to_catalog("write catalog object", err)) + self.put_object_with_options(bucket, object, data, precondition, false).await + } + + async fn put_object_unlocked( + &self, + bucket: &str, + object: &str, + data: Vec, + precondition: TableCatalogPutPrecondition, + ) -> TableCatalogStoreResult<()> { + self.put_object_with_options(bucket, object, data, precondition, true).await } async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> { @@ -1484,6 +1556,65 @@ where } } +impl EcStoreTableCatalogObjectBackend +where + S: StorageAPI, +{ + async fn read_object_with_options( + &self, + bucket: &str, + object: &str, + opts: ObjectOptions, + ) -> TableCatalogStoreResult> { + let info = match self.store.get_object_info(bucket, object, &opts).await { + Ok(info) => info, + Err(err) if is_missing_storage_error(&err) => return Ok(None), + Err(err) => return Err(storage_error_to_catalog("read catalog object info", err)), + }; + let mut reader = match self + .store + .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) + .await + { + Ok(reader) => reader, + Err(err) if is_missing_storage_error(&err) => return Ok(None), + Err(err) => return Err(storage_error_to_catalog("read catalog object", err)), + }; + let mut data = Vec::new(); + reader + .stream + .read_to_end(&mut data) + .await + .map_err(|err| TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}")))?; + Ok(Some(TableCatalogObject { + data, + etag: info.etag, + mod_time: info.mod_time, + })) + } + + async fn put_object_with_options( + &self, + bucket: &str, + object: &str, + data: Vec, + precondition: TableCatalogPutPrecondition, + no_lock: bool, + ) -> TableCatalogStoreResult<()> { + let mut reader = PutObjReader::from_vec(data); + let opts = ObjectOptions { + http_preconditions: http_preconditions_for_catalog_put(precondition), + no_lock, + ..Default::default() + }; + self.store + .put_object(bucket, object, &mut reader, &opts) + .await + .map(|_| ()) + .map_err(|err| storage_error_to_catalog("write catalog object", err)) + } +} + fn parse_namespace_for_store(namespace: &str) -> TableCatalogStoreResult { Namespace::parse(namespace).map_err(|err| TableCatalogStoreError::Invalid(format!("invalid namespace: {err}"))) } diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py new file mode 100755 index 000000000..b645a5133 --- /dev/null +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -0,0 +1,356 @@ +#!/usr/bin/env python3 +"""Manual PyIceberg smoke test for the RustFS Iceberg REST catalog.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import ssl +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class RuntimeDeps: + boto3: Any + botocore_client_error: Any + botocore_config: Any + botocore_credentials: Any + botocore_auth: Any + botocore_awsrequest: Any + pyarrow: Any + load_catalog: Any + + +def parse_args() -> argparse.Namespace: + run_id = str(int(time.time())) + parser = argparse.ArgumentParser(description="Run a PyIceberg smoke test against RustFS table catalog APIs.") + parser.add_argument("--endpoint", default=os.getenv("RUSTFS_ENDPOINT", "http://127.0.0.1:9000")) + parser.add_argument("--access-key", default=os.getenv("RUSTFS_ACCESS_KEY", "rustfsadmin")) + parser.add_argument("--secret-key", default=os.getenv("RUSTFS_SECRET_KEY", "rustfsadmin")) + parser.add_argument("--region", default=os.getenv("RUSTFS_REGION", os.getenv("AWS_REGION", "us-east-1"))) + parser.add_argument("--bucket", default=os.getenv("RUSTFS_TABLE_BUCKET", "rustfs-s3table-smoke")) + parser.add_argument("--namespace", default=os.getenv("RUSTFS_TABLE_NAMESPACE", f"smoke{run_id}")) + parser.add_argument("--table", default=os.getenv("RUSTFS_TABLE_NAME", f"events{run_id}")) + parser.add_argument("--catalog-name", default=os.getenv("RUSTFS_TABLE_CATALOG_NAME", "rustfs")) + parser.add_argument("--rest-signing-name", default=os.getenv("RUSTFS_TABLE_REST_SIGNING_NAME", "s3")) + parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20"))) + parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.") + parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.") + parser.add_argument("--insecure", action="store_true", help="Disable TLS verification for HTTPS endpoints.") + return parser.parse_args() + + +def load_runtime_deps() -> RuntimeDeps: + missing: list[str] = [] + try: + import boto3 + except ModuleNotFoundError: + boto3 = None + missing.append("boto3") + try: + import pyarrow as pyarrow + except ModuleNotFoundError: + pyarrow = None + missing.append("pyarrow") + try: + from botocore.auth import SigV4Auth + from botocore.awsrequest import AWSRequest + from botocore.exceptions import ClientError + from botocore.config import Config + from botocore.credentials import Credentials + except ModuleNotFoundError: + ClientError = Config = Credentials = SigV4Auth = AWSRequest = None + missing.append("botocore") + try: + from pyiceberg.catalog import load_catalog + except ModuleNotFoundError: + load_catalog = None + missing.append("pyiceberg") + + if missing: + unique_missing = ", ".join(sorted(set(missing))) + raise RuntimeError( + f"missing Python dependencies: {unique_missing}\n" + "Install them with: python -m pip install 'pyiceberg[pyarrow]' boto3" + ) + + return RuntimeDeps( + boto3=boto3, + botocore_client_error=ClientError, + botocore_config=Config, + botocore_credentials=Credentials, + botocore_auth=SigV4Auth, + botocore_awsrequest=AWSRequest, + pyarrow=pyarrow, + load_catalog=load_catalog, + ) + + +def normalized_endpoint(endpoint: str) -> str: + return endpoint.rstrip("/") + + +def ensure_local_proxy_bypass(endpoint: str) -> None: + host = urllib.parse.urlparse(endpoint).hostname + if host not in {"127.0.0.1", "localhost", "::1"}: + return + existing = os.environ.get("NO_PROXY") or os.environ.get("no_proxy") or "" + entries = {entry.strip() for entry in existing.split(",") if entry.strip()} + entries.update({"127.0.0.1", "localhost", "::1"}) + os.environ["NO_PROXY"] = ",".join(sorted(entries)) + + +def ensure_aws_env(access_key: str, secret_key: str, region: str) -> None: + os.environ.setdefault("AWS_ACCESS_KEY_ID", access_key) + os.environ.setdefault("AWS_SECRET_ACCESS_KEY", secret_key) + os.environ.setdefault("AWS_REGION", region) + os.environ.setdefault("AWS_DEFAULT_REGION", region) + + +def unsigned_ssl_context(insecure: bool) -> ssl.SSLContext | None: + if not insecure: + return None + return ssl._create_unverified_context() + + +def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]: + endpoint = normalized_endpoint(args.endpoint) + url = f"{endpoint}{path}" + payload = b"" if body is None else json.dumps(body, separators=(",", ":")).encode("utf-8") + headers = { + "host": urllib.parse.urlparse(url).netloc, + "x-amz-content-sha256": hashlib.sha256(payload).hexdigest(), + } + if body is not None: + headers["content-type"] = "application/json" + + aws_request = deps.botocore_awsrequest(method=method, url=url, data=payload, headers=headers) + credentials = deps.botocore_credentials(args.access_key, args.secret_key) + deps.botocore_auth(credentials, args.rest_signing_name, args.region).add_auth(aws_request) + prepared = aws_request.prepare() + + request = urllib.request.Request( + prepared.url, + data=payload if method not in {"GET", "HEAD"} else None, + headers=dict(prepared.headers.items()), + method=method, + ) + try: + with urllib.request.urlopen(request, timeout=args.timeout, context=unsigned_ssl_context(args.insecure)) as response: + response_data = response.read() + if not response_data: + return {} + return json.loads(response_data.decode("utf-8")) + except urllib.error.HTTPError as error: + response_body = error.read().decode("utf-8", errors="replace") + raise RuntimeError(f"{method} {path} failed with HTTP {error.code}: {response_body}") from error + + +def ensure_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: + client = deps.boto3.client( + "s3", + endpoint_url=normalized_endpoint(args.endpoint), + aws_access_key_id=args.access_key, + aws_secret_access_key=args.secret_key, + region_name=args.region, + verify=not args.insecure, + config=deps.botocore_config(signature_version="s3v4", s3={"addressing_style": "path"}), + ) + try: + client.head_bucket(Bucket=args.bucket) + return + except deps.botocore_client_error as error: + status_code = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode") + error_code = error.response.get("Error", {}).get("Code") + if status_code != 404 and error_code not in {"404", "NoSuchBucket", "NotFound"}: + raise + + if args.region == "us-east-1": + client.create_bucket(Bucket=args.bucket) + else: + client.create_bucket(Bucket=args.bucket, CreateBucketConfiguration={"LocationConstraint": args.region}) + + +def enable_table_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: + encoded_bucket = urllib.parse.quote(args.bucket, safe="") + signed_rest_request(args, deps, "PUT", f"/iceberg/v1/buckets/{encoded_bucket}") + + +def catalog_properties(args: argparse.Namespace) -> dict[str, str]: + endpoint = normalized_endpoint(args.endpoint) + properties = { + "type": "rest", + "uri": f"{endpoint}/iceberg", + "warehouse": args.bucket, + "prefix": args.bucket, + "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO", + "s3.endpoint": endpoint, + "s3.access-key-id": args.access_key, + "s3.secret-access-key": args.secret_key, + "s3.region": args.region, + "s3.path-style-access": "true", + "rest.sigv4-enabled": "true", + "rest.signing-region": args.region, + "rest.signing-name": args.rest_signing_name, + } + if args.insecure: + properties["s3.verify-ssl"] = "false" + return properties + + +def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, deps: RuntimeDeps) -> None: + from urllib import parse + + from requests.adapters import HTTPAdapter + + class RustfsSigV4Adapter(HTTPAdapter): + def add_headers(self, request: Any, **kwargs: Any) -> None: + body = request.body or b"" + if isinstance(body, str): + body = body.encode("utf-8") + request.headers["x-amz-content-sha256"] = hashlib.sha256(body).hexdigest() + + if "connection" in request.headers: + del request.headers["connection"] + + url = str(request.url).split("?")[0] + query = str(parse.urlsplit(request.url).query) + params = dict(parse.parse_qsl(query)) + credentials = deps.botocore_credentials(args.access_key, args.secret_key) + aws_request = deps.botocore_awsrequest( + method=request.method, + url=url, + params=params, + data=body, + headers=dict(request.headers), + ) + deps.botocore_auth(credentials, args.rest_signing_name, args.region).add_auth(aws_request) + request.headers.update(aws_request.headers) + + catalog._session.mount(catalog.uri, RustfsSigV4Adapter()) + + +def table_identifier(args: argparse.Namespace) -> tuple[str, str]: + return (args.namespace, args.table) + + +def table_exists(catalog: Any, identifier: tuple[str, str]) -> bool: + try: + return bool(catalog.table_exists(identifier)) + except AttributeError: + try: + catalog.load_table(identifier) + return True + except Exception: + return False + + +def ensure_namespace(catalog: Any, namespace: str) -> None: + try: + catalog.create_namespace_if_not_exists(namespace) + return + except AttributeError: + pass + except Exception as error: + if "already exists" in str(error).lower(): + return + raise + + try: + catalog.create_namespace(namespace) + except Exception as error: + if "already exists" not in str(error).lower(): + raise + + +def drop_table_if_present(catalog: Any, identifier: tuple[str, str]) -> None: + if table_exists(catalog, identifier): + catalog.drop_table(identifier) + + +def cleanup_catalog(catalog: Any, identifier: tuple[str, str]) -> None: + drop_table_if_present(catalog, identifier) + try: + catalog.drop_namespace(identifier[0]) + except Exception as error: + message = str(error).lower() + if "not found" not in message and "does not exist" not in message: + print(f"warning: failed to drop namespace {identifier[0]}: {error}", file=sys.stderr) + + +def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: + endpoint = normalized_endpoint(args.endpoint) + ensure_local_proxy_bypass(endpoint) + ensure_aws_env(args.access_key, args.secret_key, args.region) + + print(f"[1/7] ensuring S3 bucket {args.bucket}") + ensure_bucket(args, deps) + + print(f"[2/7] enabling RustFS table bucket {args.bucket}") + enable_table_bucket(args, deps) + + print(f"[3/7] loading PyIceberg REST catalog at {endpoint}/iceberg") + catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args)) + install_rustfs_rest_sigv4_adapter(catalog, args, deps) + identifier = table_identifier(args) + + if args.replace: + print(f"[4/7] replacing existing table {'.'.join(identifier)} if present") + drop_table_if_present(catalog, identifier) + else: + print(f"[4/7] table {'.'.join(identifier)} is available") + + print(f"[5/7] creating namespace and table {'.'.join(identifier)}") + ensure_namespace(catalog, args.namespace) + arrow_schema = deps.pyarrow.schema( + [ + deps.pyarrow.field("id", deps.pyarrow.int64(), nullable=False), + deps.pyarrow.field("payload", deps.pyarrow.string(), nullable=False), + ] + ) + table = catalog.create_table(identifier, schema=arrow_schema) + + print(f"[6/7] appending rows through PyIceberg") + rows = deps.pyarrow.Table.from_pylist( + [ + {"id": 1, "payload": "alpha"}, + {"id": 2, "payload": "beta"}, + ], + schema=arrow_schema, + ) + table.append(rows) + + print(f"[7/7] reloading and scanning table") + loaded = catalog.load_table(identifier) + scanned = loaded.scan().to_arrow() + if scanned.num_rows != 2: + raise RuntimeError(f"expected 2 rows after append, got {scanned.num_rows}") + + if args.cleanup: + print(f"cleanup: dropping table and namespace {'.'.join(identifier)}") + cleanup_catalog(catalog, identifier) + + print(f"PASS: PyIceberg smoke test completed for {'.'.join(identifier)}") + + +def main() -> int: + args = parse_args() + try: + deps = load_runtime_deps() + run_smoke(args, deps) + return 0 + except Exception as error: + print(f"ERROR: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main())