RW data type conversion to Postgres should respect the original PG data type

[yuhao-su]

RisingWave doesn't have exactly the same data type as PG. Some PG data types like uuid was converted to RW type varchar. It usually works well when we ingest data from PG.

But in some cases the data stored in RW will be converted to PG type and applied to the PG data source. One example is when backfilling, start_pk_row is originated from PG (e.g. uuid) but converted to RW varchar. When converting the RW varchar back to use as a SQL param, we will get this incorrect binary data format in bind parameter 1

risingwave/src/connector/src/source/cdc/external/postgres.rs

Lines 208 to 234 in b5ec2ae

	let sql = if start_pk_row.is_none() {
	format!(
	"SELECT {} FROM {} ORDER BY {} LIMIT {limit}",
	self.field_names,
	self.get_normalized_table_name(&table_name),
	order_key,
	)
	} else {
	let filter_expr = Self::filter_expression(&primary_keys);
	format!(
	"SELECT {} FROM {} WHERE {} ORDER BY {} LIMIT {limit}",
	self.field_names,
	self.get_normalized_table_name(&table_name),
	filter_expr,
	order_key,
	)
	};
	let client = self.client.lock().await;
	client.execute("set time zone '+00:00'", &[]).await?;
	let params: Vec<DatumRef<'_>> = match start_pk_row {
	Some(ref pk_row) => pk_row.iter().collect_vec(),
	None => Vec::new(),
	};
	let stream = client.query_raw(&sql, &params).await?;