refactor: remove unnessary check_iceberg_source

- Originally added in #14971 (When we can specify columns)
- After #15415, derive columns from iceberg source automatically and the check becomes useless.

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

Author: xxchan

.git-blame-ignore-revs

@@ -48,3 +48,6 @@ c583e2c6c054764249acf484438c7bf7197765f4
 
 # refactor: split catalog to smaller files (#19870)
 d6341b74be3f1913cc93993a95c147999df1ff74
+
+# refactor: add 2 pedantic clippy lints for strings (#19614)
+119e17eee482b6add5d5cb6ceb3b84326bf5ce68

src/common/src/array/data_chunk.rs

@@ -43,7 +43,7 @@ use crate::util::value_encoding::{
 };
 
 /// [`DataChunk`] is a collection of Columns,
-/// a with visibility mask for each row.
+/// with a visibility mask for each row.
 /// For instance, we could have a [`DataChunk`] of this format.
 ///
 /// | v1 | v2 | v3 |

src/common/src/catalog/column.rs

@@ -401,12 +401,6 @@ impl ColumnCatalog {
         ]
     }
 
-    pub fn is_iceberg_hidden_col(&self) -> bool {
-        self.column_desc.name == ICEBERG_SEQUENCE_NUM_COLUMN_NAME
-            || self.column_desc.name == ICEBERG_FILE_PATH_COLUMN_NAME
-            || self.column_desc.name == ICEBERG_FILE_POS_COLUMN_NAME
-    }
-
     /// Note: these columns are added in `SourceStreamChunkRowWriter::do_action`.
     /// May also look for the usage of `SourceColumnType`.
     pub fn debezium_cdc_source_cols() -> [Self; 3] {

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,7 +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,
+    debug_assert_column_ids_distinct, ColumnCatalog, ColumnDesc, ColumnId, TableId,
     INITIAL_SOURCE_VERSION_ID, KAFKA_TIMESTAMP_COLUMN_NAME, ROW_ID_COLUMN_NAME,
 };
 use risingwave_common::license::Feature;
@@ -620,7 +619,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],
@@ -631,10 +630,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(())
     }
@@ -892,7 +887,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

@@ -14,6 +14,7 @@
 
 use super::*;
 
+/// TODO: make hidden columns additional columns, instead of normal columns?
 pub async fn extract_iceberg_columns(
     with_properties: &WithOptionsSecResolved,
 ) -> anyhow::Result<Vec<ColumnCatalog>> {
@@ -51,51 +52,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.is_iceberg_hidden_col())
-            .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(())
-}