refactor: we don't need to check_iceberg_source since the schema is loaded

since #15415

Signed-off-by: xxchan <[email protected]>

Author: xxchan

src/common/src/catalog/mod.rs

@@ -98,6 +98,7 @@ pub fn default_key_column_name_version_mapping(version: &ColumnDescVersion) -> &
 /// [this rfc](https://github.com/risingwavelabs/rfcs/pull/20).
 pub const KAFKA_TIMESTAMP_COLUMN_NAME: &str = "_rw_kafka_timestamp";
 
+/// RisingWave iceberg table engine will create the column `_risingwave_iceberg_row_id` in the iceberg table..
 pub const RISINGWAVE_ICEBERG_ROW_ID: &str = "_risingwave_iceberg_row_id";
 pub fn is_system_schema(schema_name: &str) -> bool {
     SYSTEM_SCHEMAS.iter().any(|s| *s == schema_name)

src/frontend/src/handler/create_source.rs

@@ -19,7 +19,6 @@ use std::sync::LazyLock;
 use anyhow::{anyhow, Context};
 use either::Either;
 use external_schema::debezium::extract_debezium_avro_table_pk_columns;
-use external_schema::iceberg::check_iceberg_source;
 use external_schema::nexmark::check_nexmark_schema;
 use itertools::Itertools;
 use maplit::{convert_args, hashmap, hashset};
@@ -28,8 +27,7 @@ use rand::Rng;
 use risingwave_common::array::arrow::{arrow_schema_iceberg, IcebergArrowConvert};
 use risingwave_common::bail_not_implemented;
 use risingwave_common::catalog::{
-    debug_assert_column_ids_distinct, ColumnCatalog, ColumnDesc, ColumnId, Schema, TableId,
-    ICEBERG_FILE_PATH_COLUMN_NAME, ICEBERG_FILE_POS_COLUMN_NAME, ICEBERG_SEQUENCE_NUM_COLUMN_NAME,
+    debug_assert_column_ids_distinct, ColumnCatalog, ColumnDesc, ColumnId, TableId,
     INITIAL_SOURCE_VERSION_ID, KAFKA_TIMESTAMP_COLUMN_NAME, ROWID_PREFIX,
 };
 use risingwave_common::license::Feature;
@@ -576,7 +574,7 @@ pub(super) fn bind_source_watermark(
 ///
 /// One should only call this function after all properties of all columns are resolved, like
 /// generated column descriptors.
-pub(super) async fn check_format_encode(
+pub(super) fn check_format_encode(
     props: &WithOptionsSecResolved,
     row_id_index: Option<usize>,
     columns: &[ColumnCatalog],
@@ -587,10 +585,6 @@ pub(super) async fn check_format_encode(
 
     if connector == NEXMARK_CONNECTOR {
         check_nexmark_schema(props, row_id_index, columns)
-    } else if connector == ICEBERG_CONNECTOR {
-        Ok(check_iceberg_source(props, columns)
-            .await
-            .map_err(|err| ProtocolError(err.to_report_string()))?)
     } else {
         Ok(())
     }
@@ -839,7 +833,7 @@ pub async fn bind_create_source_or_table_with_connector(
         sql_columns_defs.to_vec(),
         &pk_col_ids,
     )?;
-    check_format_encode(&with_properties, row_id_index, &columns).await?;
+    check_format_encode(&with_properties, row_id_index, &columns)?;
 
     let definition = handler_args.normalized_sql.clone();
 

src/frontend/src/handler/create_source/external_schema/iceberg.rs

@@ -53,55 +53,3 @@ pub async fn extract_iceberg_columns(
         )))
     }
 }
-
-pub async fn check_iceberg_source(
-    props: &WithOptionsSecResolved,
-    columns: &[ColumnCatalog],
-) -> anyhow::Result<()> {
-    let props = ConnectorProperties::extract(props.clone(), true)?;
-    let ConnectorProperties::Iceberg(properties) = props else {
-        return Err(anyhow!(format!(
-            "Invalid properties for iceberg source: {:?}",
-            props
-        )));
-    };
-
-    let schema = Schema {
-        fields: columns
-            .iter()
-            .filter(|&c| {
-                c.column_desc.name != ICEBERG_SEQUENCE_NUM_COLUMN_NAME
-                    && c.column_desc.name != ICEBERG_FILE_PATH_COLUMN_NAME
-                    && c.column_desc.name != ICEBERG_FILE_POS_COLUMN_NAME
-            })
-            .cloned()
-            .map(|c| c.column_desc.into())
-            .collect(),
-    };
-
-    let table = properties.load_table().await?;
-
-    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) {
-            return Err(anyhow::anyhow!(format!(
-                "Column {} not found in iceberg table",
-                f1.name
-            )));
-        }
-    }
-
-    let new_iceberg_field = iceberg_schema
-        .fields
-        .iter()
-        .filter(|f1| schema.fields.iter().any(|f2| f1.name() == &f2.name))
-        .cloned()
-        .collect::<Vec<_>>();
-    let new_iceberg_schema = arrow_schema_iceberg::Schema::new(new_iceberg_field);
-
-    risingwave_connector::sink::iceberg::try_matches_arrow_schema(&schema, &new_iceberg_schema)?;
-
-    Ok(())
-}