refactor(iceberg): extract IcebergCommon config

src/batch/src/executor/iceberg_scan.rs

@@ -26,7 +26,6 @@ use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::types::DataType;
 use risingwave_common::util::iter_util::ZipEqFast;
-use risingwave_connector::sink::iceberg::IcebergConfig;
 use risingwave_connector::source::iceberg::{IcebergProperties, IcebergSplit};
 use risingwave_connector::source::{ConnectorProperties, SplitImpl, SplitMetaData};
 use risingwave_connector::WithOptionsSecResolved;
@@ -38,7 +37,7 @@ use crate::executor::{DataChunk, Executor};
 use crate::task::BatchTaskContext;
 
 pub struct IcebergScanExecutor {
-    iceberg_config: IcebergConfig,
+    iceberg_config: IcebergProperties,
     #[allow(dead_code)]
     snapshot_id: Option<i64>,
     table_meta: TableMetadata,
@@ -65,7 +64,7 @@ impl Executor for IcebergScanExecutor {
 
 impl IcebergScanExecutor {
     pub fn new(
-        iceberg_config: IcebergConfig,
+        iceberg_config: IcebergProperties,
         snapshot_id: Option<i64>,
         table_meta: TableMetadata,
         data_file_scan_tasks: Vec<FileScanTask>,
@@ -277,7 +276,7 @@ impl BoxedExecutorBuilder for IcebergScanExecutorBuilder {
             let iceberg_properties: IcebergProperties = *iceberg_properties;
             let split: IcebergSplit = split.clone();
             Ok(Box::new(IcebergScanExecutor::new(
-                iceberg_properties.to_iceberg_config(),
+                iceberg_properties,
                 Some(split.snapshot_id),
                 split.table_meta.deserialize(),
                 split.files.into_iter().map(|x| x.deserialize()).collect(),

src/connector/src/connector_common/common.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::borrow::Cow;
 use std::collections::BTreeMap;
 use std::io::Write;
 use std::time::Duration;
@@ -38,8 +37,6 @@ use crate::deserialize_duration_from_string;
 use crate::error::ConnectorResult;
 use crate::sink::SinkError;
 use crate::source::nats::source::NatsOffset;
-// The file describes the common abstractions for each connector and can be used in both source and
-// sink.
 
 pub const PRIVATE_LINK_BROKER_REWRITE_MAP_KEY: &str = "broker.rewrite.endpoints";
 pub const PRIVATE_LINK_TARGETS_KEY: &str = "privatelink.targets";
@@ -563,13 +560,6 @@ impl KinesisCommon {
         Ok(KinesisClient::from_conf(builder.build()))
     }
 }
-#[derive(Debug, Deserialize)]
-pub struct UpsertMessage<'a> {
-    #[serde(borrow)]
-    pub primary_key: Cow<'a, [u8]>,
-    #[serde(borrow)]
-    pub record: Cow<'a, [u8]>,
-}
 
 #[serde_as]
 #[derive(Deserialize, Debug, Clone, WithOptions)]