feat: skip rw's generated column when validating source

e2e_test/source/cdc/cdc.validate.postgres.slt

@@ -168,6 +168,9 @@ create table shipments (
  slot.name = 'shipments'
 ) format canal encode csv;
 
+statement ok
+drop table shipments;
+
 statement ok
 explain create table numeric_to_rw_int256 (
   id int,
@@ -217,3 +220,28 @@ explain create table enum_to_varchar (
  table.name = 'enum_table',
  slot.name = 'enum_to_varchar'
 );
+
+# generated column
+statement ok
+explain create table shipments (
+  shipment_id INTEGER,
+  order_id INTEGER,
+  origin STRING,
+  destination STRING,
+  is_arrived boolean,
+  generated_c INGEGER AS shipment_id + order_id,
+ PRIMARY KEY (shipment_id)
+) with (
+ connector = 'postgres-cdc',
+ hostname = 'db',
+ port = '5432',
+ username = 'postgres',
+ password = 'postgres',
+ database.name = 'cdc_test',
+ schema.name = 'public',
+ table.name = 'shipments',
+ slot.name = 'shipments'
+);
+
+statement ok
+drop table shipments;

src/connector/src/source/cdc/mod.rs

@@ -23,9 +23,9 @@ use std::marker::PhantomData;
 
 pub use enumerator::*;
 use itertools::Itertools;
-use risingwave_common::catalog::{ColumnDesc, Field, Schema};
 use risingwave_pb::catalog::PbSource;
 use risingwave_pb::connector_service::{PbSourceType, PbTableSchema, SourceType, TableSchema};
+use risingwave_pb::plan_common::column_desc::GeneratedOrDefaultColumn;
 use risingwave_pb::plan_common::ExternalTableDesc;
 use simd_json::prelude::ArrayTrait;
 pub use source::*;
@@ -144,6 +144,12 @@ where
                 .columns
                 .iter()
                 .flat_map(|col| &col.column_desc)
+                .filter(|col| {
+                    !matches!(
+                        col.generated_or_default_column,
+                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
+                    )
+                })
                 .cloned()
                 .collect(),
             pk_indices,
@@ -159,7 +165,17 @@ where
             table_desc.connect_properties.clone().into_iter().collect();
 
         let table_schema = TableSchema {
-            columns: table_desc.columns.clone(),
+            columns: table_desc
+                .columns
+                .iter()
+                .filter(|col| {
+                    !matches!(
+                        col.generated_or_default_column,
+                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
+                    )
+                })
+                .cloned()
+                .collect(),
             pk_indices: table_desc.stream_key.clone(),
         };
 
@@ -181,16 +197,4 @@ impl<T: CdcSourceTypeTrait> CdcProperties<T> {
     pub fn get_source_type_pb(&self) -> SourceType {
         SourceType::from(T::source_type())
     }
-
-    pub fn schema(&self) -> Schema {
-        Schema {
-            fields: self
-                .table_schema
-                .columns
-                .iter()
-                .map(ColumnDesc::from)
-                .map(Field::from)
-                .collect(),
-        }
-    }
 }

src/rpc_client/src/connector_client.rs

@@ -30,6 +30,7 @@ use risingwave_pb::connector_service::sink_writer_stream_request::{
 };
 use risingwave_pb::connector_service::sink_writer_stream_response::CommitResponse;
 use risingwave_pb::connector_service::*;
+use risingwave_pb::plan_common::column_desc::GeneratedOrDefaultColumn;
 use thiserror_ext::AsReport;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::transport::{Channel, Endpoint};
@@ -238,6 +239,15 @@ impl ConnectorClient {
         is_source_job: bool,
         is_backfill_table: bool,
     ) -> Result<()> {
+        let table_schema = table_schema.map(|mut table_schema| {
+            table_schema.columns.retain(|c| {
+                !matches!(
+                    c.generated_or_default_column,
+                    Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
+                )
+            });
+            table_schema
+        });
         let response = self
             .rpc_client
             .clone()