Tracking: Postgres sink issues

[StrikeW]

• Migrate postgres sink from jdbc to rust implementation using tokio_postgres #19213

• Support sink to uuid array (uuid[]) column Support more array types in remote sinks #14616
  @StrikeW: I find that it might be difficult to support this via JDBC interface, since java.sql.Types doesn't have a type corresponding to UUID array, so we cannot get the uuid array type from the prepared statement. I am thinking about rewriting the Postgres sink with Rust tokio_postgres and leverage the ScalarAdapter (refactor(cdc): refactor parsing of non-builtin Postgres data types #16589 ) to convert our types to PG types.

• Check data type mismatch of downstream table 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 #16687
  We can check the data types after prepared the SQL statement

• bug(postgres-sink): delete doesn't work for timestamp with time zone #18179

• jdbc sink stuck on jvm side #18372

• Investigate if we can support Connection pool

[fuyufjh]

I am thinking about rewriting the Postgres sink with Rust tokio_postgres and leverage the ScalarAdapter

+1 for migrating from JDBC to native implementation. My intuitive is that JDBC adds another abstraction layer and makes it more difficult for us to make all data type conversion consistent among source/sink systems. Let alone the performance loss.

[xiangjinwu]

• Agree to implement the sink in rust and avoid going thru JDBC.

• I find that it might be difficult to support this via JDBC interface, since java.sql.Types doesn't have a type corresponding to UUID array, so we cannot get the uuid array type from the prepared statement.

  It is doable. The following snippet outputs _uuid, which means array of uuid in PostgreSQL (just like _float8 is double precision[]).

    PreparedStatement stp = conn.prepareStatement("select array['018f761c-a819-748a-863f-52204f67aa2a'::uuid] = ?;"
    System.out.println(stp.getParameterMetaData().getParameterTypeName(1));
  

  Just sharing it is possible. But for a better future I am also in favor of the rust implementation.

[fuyufjh]

Today we met a stability issue in PostgreSQL sink, causing the barriers stuck forever.

[Actor 445607]
Actor 445607: `<redacted>` [972.924s]
  Epoch 6619587039723520 [!!! 937.694s]
    Sink 6CCA700000002 [!!! 924.654s]
      Consume Log: sink_id: 46746 actor_id: 445607, executor_id: 1913867491868674 [!!! 972.924s]
        Wait Response Stream [!!! 937.694s]
        Wait Next Item: 6619587039723520 [!!! 923.734s]

Here the Wait Response Stream indicates the barrier was not passed back from Java connector.

Even worse, there are no logs telling what's happening in Java side. No helpful logs found.

[StrikeW]

Today we met a stability issue in PostgreSQL sink, causing the barriers stuck forever.

[Actor 445607]
Actor 445607: `<redacted>` [972.924s]
  Epoch 6619587039723520 [!!! 937.694s]
    Sink 6CCA700000002 [!!! 924.654s]
      Consume Log: sink_id: 46746 actor_id: 445607, executor_id: 1913867491868674 [!!! 972.924s]
        Wait Response Stream [!!! 937.694s]
        Wait Next Item: 6619587039723520 [!!! 923.734s]

Here the Wait Response Stream indicates the barrier was not passed back from Java connector.

Even worse, there are no logs telling what's happening in Java side. No helpful logs found.

I think we should prioritize #17095 to enable sink decouple by default. After that, issues happen in the Sink would not affect the whole cluster.

[fuyufjh]

I think we should prioritize #17095 to enable sink decouple by default. After that, issues happen in the Sink would not affect the whole cluster.

They are independent problems, I think. Even with sink_decouple=true, if this unknown bug recurs causing the JDBC connector to hang forever, the streaming job will eventually be stuck when log store become full.

[lmatz]

One issue mentioned by a user lately is that if different sinks point to the same destination, PG in this case, can they share the same connection (pool)?

[StrikeW]

One issue mentioned by a user lately is that if different sinks point to the same destination, PG in this case, can they share the same connection (pool)?

As I mentioned in the Notion, in current schedule policy it is hard to share connection for sinks. And embed a connection pool in CN will make it become stateful which may introduce limitation to scaling.