Merge remote-tracking branch 'origin/main' into bz/transaction-test

ci/scripts/connector-node-integration-test.sh

@@ -86,8 +86,9 @@ tar xf java-binding-integration-test.tar.zst bin
 
 echo "--- prepare integration tests"
 cd ${RISINGWAVE_ROOT}/java/connector-node
-pip3 install grpcio grpcio-tools psycopg2 psycopg2-binary pyspark==3.3
-cd python-client && bash gen-stub.sh
+pip3 install grpcio grpcio-tools psycopg2 psycopg2-binary pyspark==3.3 black
+cd python-client && bash gen-stub.sh && bash format-python.sh --check
+export PYTHONPATH=proto
 
 echo "--- running streamchunk data format integration tests"
 cd ${RISINGWAVE_ROOT}/java/connector-node/python-client

ci/scripts/release.sh

@@ -11,7 +11,7 @@ if [ "${BUILDKITE_SOURCE}" != "schedule" ] && [ "${BUILDKITE_SOURCE}" != "webhoo
 fi
 
 echo "--- Install java and maven"
-yum install -y java-11-openjdk wget python3
+yum install -y java-11-openjdk wget python3 cyrus-sasl-devel
 pip3 install toml-cli
 wget https://ci-deps-dist.s3.amazonaws.com/apache-maven-3.9.3-bin.tar.gz && tar -zxvf apache-maven-3.9.3-bin.tar.gz
 export PATH="${REPO_ROOT}/apache-maven-3.9.3/bin:$PATH"

ci/workflows/integration-tests.yml

@@ -93,7 +93,7 @@ steps:
         testcase:
           - "twitter"
           - "twitter-pulsar"
-          - "debezium-mongo"
+        # - "debezium-mongo"
           - "debezium-postgres"
           - "tidb-cdc-sink"
           - "debezium-sqlserver"
@@ -105,10 +105,10 @@ steps:
             testcase: "twitter-pulsar"
             format: "protobuf"
           skip: true
-        - with:
-            testcase: "debezium-mongo"
-            format: "protobuf"
-          skip: true
+      # - with:
+      #     testcase: "debezium-mongo"
+      #     format: "protobuf"
+      #   skip: true
         - with:
             testcase: "debezium-postgres"
             format: "protobuf"

e2e_test/streaming/temporal_join/temporal_join_with_index.slt

@@ -0,0 +1,84 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table stream(id1 int, a1 int, b1 int) APPEND ONLY;
+
+statement ok
+create table version(id2 int, a2 int, b2 int, primary key (id2));
+
+statement ok
+create index idx on version (a2);
+
+statement ok
+create materialized view v as select id1, a1, id2, a2 from stream left join idx FOR SYSTEM_TIME AS OF PROCTIME() on b1 = b2 and a1 = a2;
+
+statement ok
+insert into stream values(1, 11, 111);
+
+statement ok
+insert into version values(1, 11, 111);
+
+statement ok
+insert into version values(9, 11, 111);
+
+statement ok
+insert into stream values(1, 11, 111);
+
+statement ok
+delete from version;
+
+query IIII rowsort
+select * from v;
+----
+1 11 1 11
+1 11 9 11
+1 11 NULL NULL
+
+statement ok
+insert into version values(2, 22, 222);
+
+statement ok
+insert into stream values(2, 22, 222);
+
+statement ok
+insert into version values(8, 22, 222);
+
+statement ok
+insert into stream values(2, 22, 222);
+
+query IIII rowsort
+select * from v;
+----
+1 11 1 11
+1 11 9 11
+1 11 NULL NULL
+2 22 2 22
+2 22 2 22
+2 22 8 22
+
+statement ok
+update version set b2 = 333 where id2 = 2;
+
+statement ok
+insert into stream values(2, 22, 222);
+
+query IIII rowsort
+select * from v;
+----
+1 11 1 11
+1 11 9 11
+1 11 NULL NULL
+2 22 2 22
+2 22 2 22
+2 22 8 22
+2 22 8 22
+
+statement ok
+drop materialized view v;
+
+statement ok
+drop table stream;
+
+statement ok
+drop table version;

java/connector-node/README.md

@@ -62,13 +62,17 @@ Navigate to the `python-client` directory and run the following command:
 ```
 bash build-venv.sh
 bash gen-stub.sh
-python3 integration_tests.py
+PYTHONPATH=proto python3 integration_tests.py
 ```
 
 Or you can use conda and install the necessary package `grpcio grpcio-tools psycopg2 psycopg2-binary`. 
 
 The connector service is the server and Python integration test is a client, which will send gRPC request and get response from the connector server. So when running integration_tests, remember to launch the connector service in advance. You can get the gRPC response and check messages or errors in client part. And check the detailed exception information on server side.
 
+### Python file format
+
+We use `black` as the python file formatter. We can run `format-python.sh` to format the python files. 
+
 ### JDBC test
 
 We have integration tests that involve the use of several sinks, including file sink, jdbc sink, iceberg sink, and deltalake sink. If you wish to run these tests locally, you will need to configure both MinIO and PostgreSQL. 

java/connector-node/python-client/.gitignore

@@ -1,3 +1 @@
-*.py
-
-!integration_tests.py
+.vscode

java/connector-node/python-client/build-venv.sh

@@ -1,3 +1,3 @@
 virtualenv sink-client-venv
 source sink-client-venv/bin/activate
-pip3 install grpcio grpcio-tools psycopg2 psycopg2-binary
+pip3 install grpcio grpcio-tools psycopg2 psycopg2-binary black

java/connector-node/python-client/format-python.sh

@@ -0,0 +1,2 @@
+set -ex
+black $@ integration_tests.py pyspark-util.py

java/connector-node/python-client/gen-stub.sh

@@ -1 +1 @@
-python3 -m grpc_tools.protoc -I../../../proto/   --python_out=. --grpc_python_out=. ../../../proto/*.proto
+python3 -m grpc_tools.protoc -I../../../proto/   --python_out=./proto --grpc_python_out=./proto ../../../proto/*.proto