From 9c4984d4ec2d17351a741cab8ac512b39dd2261a Mon Sep 17 00:00:00 2001 From: Dylan Date: Thu, 22 Aug 2024 19:15:39 +0800 Subject: [PATCH] refactor(iceberg): use load table v2 to extract iceberg columns (#18196) --- src/frontend/src/handler/create_source.rs | 17 +++++------------ 1 file changed, 5 insertions(+), 12 deletions(-) diff --git a/src/frontend/src/handler/create_source.rs b/src/frontend/src/handler/create_source.rs index c4048dd7cac01..09d5b8232e386 100644 --- a/src/frontend/src/handler/create_source.rs +++ b/src/frontend/src/handler/create_source.rs @@ -1313,12 +1313,9 @@ pub async fn extract_iceberg_columns( let props = ConnectorProperties::extract(with_properties.clone(), true)?; if let ConnectorProperties::Iceberg(properties) = props { let iceberg_config: IcebergConfig = properties.to_iceberg_config(); - let table = iceberg_config.load_table().await?; - let iceberg_schema: arrow_schema_iceberg::Schema = table - .current_table_metadata() - .current_schema()? - .clone() - .try_into()?; + let table = iceberg_config.load_table_v2().await?; + let iceberg_schema: arrow_schema_iceberg::Schema = + iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema())?; let columns = iceberg_schema .fields() @@ -1368,13 +1365,9 @@ pub async fn check_iceberg_source( .collect(), }; - let table = iceberg_config.load_table().await?; + let table = iceberg_config.load_table_v2().await?; - let iceberg_schema: arrow_schema_iceberg::Schema = table - .current_table_metadata() - .current_schema()? - .clone() - .try_into()?; + let iceberg_schema = iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema())?; for f1 in schema.fields() { if !iceberg_schema.fields.iter().any(|f2| f2.name() == &f1.name) {