refine

Signed-off-by: xxchan <xxchan22f@gmail.com>

src/common/src/catalog/column.rs

@@ -451,6 +451,20 @@ pub fn debug_assert_column_ids_distinct(columns: &[ColumnCatalog]) {
     );
 }
 
+/// FIXME: perhapts we should use sth like `ColumnIdGenerator::new_alter`,
+/// However, the `SourceVersion` is problematic: It doesn't contain `next_col_id`.
+/// (But for now this isn't a large problem, since drop column is not allowed for source yet..)
+///
+/// Besides, the logic of column id handling is a mess.
+/// In some places, we use `ColumnId::placeholder()`, and use `col_id_gen` to fill it at the end;
+/// In other places, we create column id ad-hoc.
+pub fn max_column_id(columns: &[ColumnCatalog]) -> ColumnId {
+    // XXX: should we check the column IDs of struct fields here?
+    columns
+        .iter()
+        .fold(ColumnId::first_user_column(), |a, b| a.max(b.column_id()))
+}
+
 #[cfg(test)]
 pub mod tests {
     use risingwave_pb::plan_common::PbColumnDesc;

src/connector/src/parser/additional_columns.rs

@@ -16,7 +16,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::LazyLock;
 
 use risingwave_common::bail;
-use risingwave_common::catalog::{ColumnCatalog, ColumnDesc, ColumnId};
+use risingwave_common::catalog::{max_column_id, ColumnCatalog, ColumnDesc, ColumnId};
 use risingwave_common::types::{DataType, StructType};
 use risingwave_pb::data::data_type::TypeName;
 use risingwave_pb::data::DataType as PbDataType;
@@ -29,7 +29,6 @@ use risingwave_pb::plan_common::{
 };
 
 use crate::error::ConnectorResult;
-use crate::parser::max_column_id;
 use crate::source::cdc::MONGODB_CDC_CONNECTOR;
 use crate::source::{
     GCS_CONNECTOR, KAFKA_CONNECTOR, KINESIS_CONNECTOR, OPENDAL_S3_CONNECTOR, PULSAR_CONNECTOR,
@@ -281,11 +280,6 @@ pub fn source_add_partition_offset_cols(
     connector_name: &str,
 ) -> ([bool; 2], [ColumnCatalog; 2]) {
     let mut columns_exist = [false; 2];
-    // let mut last_column_id = columns
-    //     .iter()
-    //     .map(|c| c.column_desc.column_id)
-    //     .max()
-    //     .unwrap_or(ColumnId::placeholder());
     let mut last_column_id = max_column_id(columns);
 
     let additional_columns: Vec<_> = {

src/connector/src/parser/mod.rs

@@ -88,7 +88,6 @@ mod util;
 pub use debezium::DEBEZIUM_IGNORE_KEY;
 use risingwave_common::buffer::BitmapBuilder;
 pub use unified::{AccessError, AccessResult};
-pub use util::max_column_id;
 
 /// A builder for building a [`StreamChunk`] from [`SourceColumnDesc`].
 pub struct SourceStreamChunkBuilder {

src/connector/src/parser/util.rs

@@ -17,7 +17,6 @@ use anyhow::Context;
 use bytes::Bytes;
 use reqwest::Url;
 use risingwave_common::bail;
-use risingwave_common::catalog::{ColumnCatalog, ColumnId};
 use risingwave_common::types::{Datum, DatumCow, DatumRef};
 use risingwave_pb::data::DataType as PbDataType;
 
@@ -173,17 +172,3 @@ pub fn extract_header_inner_from_meta<'a>(
         _ => None,
     }
 }
-
-/// FIXME: perhapts we should use sth like `ColumnIdGenerator::new_alter`,
-/// However, the `SourceVersion` is problematic: It doesn't contain `next_col_id`.
-/// (But for now this isn't a large problem, since drop column is not allowed for source yet..)
-///
-/// Besides, the logic of column id handling is a mess.
-/// In some places, we use `ColumnId::placeholder()`, and use `col_id_gen` to fill it at the end;
-/// In other places, we create column id ad-hoc.
-pub fn max_column_id(columns: &[ColumnCatalog]) -> ColumnId {
-    // XXX: should we check the column IDs of struct fields here?
-    columns
-        .iter()
-        .fold(ColumnId::new(i32::MIN), |a, b| a.max(b.column_id()))
-}

src/frontend/src/handler/alter_source_column.rs

@@ -14,7 +14,7 @@
 
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
-use risingwave_connector::parser::max_column_id;
+use risingwave_common::catalog::max_column_id;
 use risingwave_connector::source::{extract_source_struct, SourceEncode, SourceStruct};
 use risingwave_sqlparser::ast::{
     AlterSourceOperation, ColumnDef, CreateSourceStatement, ObjectName, Statement,

src/frontend/src/handler/alter_source_with_sr.rs

@@ -18,8 +18,7 @@ use anyhow::Context;
 use itertools::Itertools;
 use pgwire::pg_response::StatementType;
 use risingwave_common::bail_not_implemented;
-use risingwave_common::catalog::ColumnCatalog;
-use risingwave_connector::parser::max_column_id;
+use risingwave_common::catalog::{max_column_id, ColumnCatalog};
 use risingwave_connector::WithPropertiesExt;
 use risingwave_pb::catalog::StreamSourceInfo;
 use risingwave_pb::plan_common::{EncodeType, FormatType};