Catch and return sink writer error ERROR: column \“some_column\” is of type some_pg_type but expression is of type character varying` to user

[kwannoel]

Describe the bug

ERROR: column \“some_column\” is of type some_pg_type but expression is of type character varying.

 Hint: You will need to rewrite or cast the expression.
  Position: 271  Call getNextException to see other errors in the batch.
  at io.grpc.Status.asRuntimeException(Status.java:530)
  at com.risingwave.connector.JDBCSink.write(JDBCSink.java:196)
  at com.risingwave.connector.SinkWriterStreamObserver.onNext(SinkWriterStreamObserver.java:132)
  com.risingwave.connector.JniSinkWriterHandler.runJniSinkWriterThread(JniSinkWriterHandler.java:40)",

Can we propagate this error back to RW, on sink create?

Perhaps we can delay reporting the first barrier, write some records downstream first then:

1. No error => yield first barrier, along with offsets.
2. Error => Propagate error back.

Error message/log

No response

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[kwannoel]

Related #13035

[lmatz]

Is this because it simply does not check whether the data types in RW and PostgreSQL mismatch or not,
or there is data type incompatibility issue as we test here: https://github.com/risingwavelabs/risingwave/blob/main/integration_tests/postgres-sink/create_source.sql#L46-L69

If it does not check the mismatch, I suppose we should add such test cases to the e2e test where the XXX in statement error XXX can be tested.

[StrikeW]

We did not check column types with downstream table right now. We can add the check during creation of a sink.

[lmatz]

Got it, thanks
I am going through recent incidents and just want to make sure that tests cover them