sink with temporal filter does not delete the row in the downstream database

[lmatz]

Describe the bug

CREATE TABLE person (
  id INT,
  name VARCHAR,
  city VARCHAR,
  account_id BIGINT DEFAULT NULL,
  event_date_time TIMESTAMP WITHOUT TIME ZONE
);

INSERT INTO person
(id, name, city, account_id, event_date_time)
VALUES
(6, 'Rahul', 'Goa', 1, Now());

CREATE MATERIALIZED VIEW rw_person_info_mv  AS 
SELECT name AS person_name, city, account_id, event_date_time FROM person
WHERE event_date_time > (NOW() - INTERVAL '1 minute');

CREATE SINK rw_person_info
FROM rw_person_info_mv
WITH (
  connector='jdbc',
  jdbc.url= 'jdbc:mysql://host.docker.internal:3306/public?user=root',type= 'upsert',
table.name= 'person_info',primary_key='person_name,account_id,event_date_time');

After one minute
MySQL still:

mysql> select * from person_info;
+-------------+------+------------+---------------------+
| person_name | city | account_id | event_date_time     |
+-------------+------+------------+---------------------+
| Rahul       | Goa  |          1 | 2024-05-14 05:28:24 |
+-------------+------+------------+---------------------+
1 row in set (0.00 sec)

RW:

dev=> select * from rw_person_info_mv;
 person_name | city | account_id | event_date_time 
-------------+------+------------+-----------------
(0 rows)

Error message/log

No error found in the log

To Reproduce

see above

Expected behavior

The record in the downstream MySQL table should be deleted as well

How did you deploy RisingWave?

1.8.1 standalone mode by docker-compose is used

The version of RisingWave

1.8.1

[lmatz]

During debugging on the branch:https://github.com/risingwavelabs/risingwave/tree/lz/sink (essentially main + a few new logs)

another problem occurred

MySQL:

CREATE TABLE person_info (
  id INT,
  person_name VARCHAR(255),
  city VARCHAR(255),
  account_id BIGINT NOT NULL,
  event_date_time TIMESTAMP ,
  PRIMARY KEY (person_name,account_id,event_date_time)
);

RW:

CREATE TABLE person (
  id INT,
  name VARCHAR,
  city VARCHAR,
  account_id BIGINT DEFAULT NULL,
  event_date_time TIMESTAMP WITHOUT TIME ZONE
);

INSERT INTO person
(id, name, city, account_id, event_date_time)
VALUES
(6, 'Rahul', 'Goa', 1, Now());

CREATE MATERIALIZED VIEW rw_person_info_mv  AS 
SELECT id, name AS person_name, city, account_id, event_date_time FROM person
WHERE event_date_time > (NOW() - INTERVAL '1 minute');

CREATE SINK rw_person_info
FROM rw_person_info_mv
WITH (
  connector='jdbc',
  jdbc.url= 'jdbc:mysql://127.0.0.1:3306/public?user=root',type= 'upsert',
table.name= 'person_info',primary_key='person_name,account_id,event_date_time');

Log:

 2024-05-15T15:20:21.550948+08:00  INFO risingwave_connector_node: Bind delete statement:com.mysql.cj.jdbc.       ClientPreparedStatement: DELETE FROM `person_info` WHERE `account_id` = ** NOT SPECIFIED ** AND                  `event_date_time` = ** NOT SPECIFIED ** AND `person_name` = ** NOT SPECIFIED **, row:com.risingwave.connector.   deserializer.StreamChunkDeserializer$StreamChunkRowWrapper@d859b73 thread="Thread-145" class="com.risingwave.    connector.JDBCSink"
 2024-05-15T15:20:21.551026+08:00  INFO risingwave_connector_node: pkIndices:[1, 3, 4] thread="Thread-145"        class="com.risingwave.connector.jdbc.MySqlDialect"
 2024-05-15T15:20:21.551073+08:00  INFO risingwave_connector_node: pkColumnSqlTypes:[12, -5, 93] thread="Thread-  145" class="com.risingwave.connector.jdbc.MySqlDialect"
 2024-05-15T15:20:21.551353+08:00 ERROR risingwave_connector_node: sink writer error: : java.lang.                ArrayIndexOutOfBoundsException: Index 3 out of bounds for length 3
     at com.risingwave.connector.jdbc.MySqlDialect.bindDeleteStatement(MySqlDialect.java:131)
     at com.risingwave.connector.JDBCSink$JdbcStatements.prepareDelete(JDBCSink.java:315)
     at com.risingwave.connector.JDBCSink.write(JDBCSink.java:164)
     at com.risingwave.connector.SinkWriterStreamObserver.onNext(SinkWriterStreamObserver.java:132)
     at com.risingwave.connector.JniSinkWriterHandler.runJniSinkWriterThread(JniSinkWriterHandler.java:40)
  thread="Thread-145" class="com.risingwave.connector.SinkWriterStreamObserver"
 2024-05-15T15:20:21.552221+08:00 ERROR risingwave_connector_node: JniSinkWriterHandler onError: : java.lang.     ArrayIndexOutOfBoundsException: Index 3 out of bounds for length 3
     at com.risingwave.connector.jdbc.MySqlDialect.bindDeleteStatement(MySqlDialect.java:131)
     at com.risingwave.connector.JDBCSink$JdbcStatements.prepareDelete(JDBCSink.java:315)
     at com.risingwave.connector.JDBCSink.write(JDBCSink.java:164)
     at com.risingwave.connector.SinkWriterStreamObserver.onNext(SinkWriterStreamObserver.java:132)
     at com.risingwave.connector.JniSinkWriterHandler.runJniSinkWriterThread(JniSinkWriterHandler.java:40)
  thread="Thread-145" class="com.risingwave.connector.JniSinkWriterResponseObserver"
 2024-05-15T15:20:21.55246+08:00 ERROR risingwave_connector_node: sink writer finishes with error: : java.lang.   RuntimeException: fail to sendSinkWriterResponseToChannel
     at com.risingwave.connector.JniSinkWriterHandler.runJniSinkWriterThread(JniSinkWriterHandler.java:43)
  thread="Thread-145" class="com.risingwave.connector.SinkWriterStreamObserver"

2024-05-15T15:20:21.552896+08:00 ERROR risingwave_stream::task::stream_manager: actor exit with error            actor_id=14 error=Executor error: Sink error: Remote sink error: Index 3 out of bounds for length 3
 
 Backtrace:
    0: std::backtrace_rs::backtrace::libunwind::trace
              at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/../../backtrace/src/backtrace/   libunwind.rs:104:5
    1: std::backtrace_rs::backtrace::trace_unsynchronized
              at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/../../backtrace/src/backtrace/   mod.rs:66:5
    2: std::backtrace::Backtrace::create
              at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/backtrace.rs:331:13
    3: anyhow::kind::Adhoc::new
              at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/anyhow-1.0.81/src/kind.rs:73: 36
    4: risingwave_jni_core::Java_com_risingwave_java_binding_Binding_sendSinkWriterErrorToChannel::{{closure}}
              at ./src/jni_core/src/lib.rs:1154:58
    5: risingwave_jni_core::execute_and_catch::{{closure}}
              at ./src/jni_core/src/lib.rs:209:59
    6: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
              at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panic/unwind_safe.rs:272:9
    7: std::panicking::try::do_call
              at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/panicking.rs:552:40

And because of the error, the one row in the MV rw_person_info_mv won't expire and get deleted.
select * from rw_person_info_mv will always return the row we inserted in.
(It works fine if I don't create the sink)

[lmatz]

I guess these could be two separate problems....

the first one is 1.8.1 and the second one is main + new logs