feat(sink): enable delta lake sink

[wenym1]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Previously, the delta lake java connector is implemented, but user cannot yet create the delta lake sink via SQL. In this PR, we enable user create delta lake sink via SQL.

Following is a demo to run the delta lake sink. Delta lake data are stored in minio. But it should be able to be stored in AWS S3.

1. Start a RisingWave cluster with minio via ./risedev d ci-iceberg-test. Minio endpoint is at localhost:9301.
2. Create a bucket named deltalake in the minio portal. Portal endpoint is at http://localhost:9400.
3. Download spark 3.3.1 distribution from https://archive.apache.org/dist/spark/spark-3.3.1/ and launch a spark sql shell with the following command

./spark-3.3.1-bin-hadoop3/bin/spark-sql --packages io.delta:delta-core_2.12:2.2.0,org.apache.hadoop:hadoop-aws:3.3.2 \
--conf 'spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog' \
--conf 'spark.hadoop.fs.s3a.access.key=hummockadmin' \
--conf 'spark.hadoop.fs.s3a.secret.key=hummockadmin' \
--conf 'spark.hadoop.fs.s3a.endpoint=http://localhost:9301' \
--conf 'spark.hadoop.fs.s3a.path.style.access=true' 

4. In spark sql shell, execute the following SQL to create a deltalake table.

create table delta.`s3a://deltalake/delta`(id int, name string) using delta

5. Connect to RisingWave via psql -h localhost -p 4566 -d dev -U root and then create source and sink. Data are randomly generated with the datagen source.

CREATE SOURCE source (id int, name varchar)
WITH (
     connector = 'datagen',
     fields.id.kind = 'sequence',
     fields.id.start = '1',
     fields.id.end = '10000',
     fields.name.kind = 'random',
     fields.name.length = '10',
     datagen.rows.per.second = '200'
 ) ROW FORMAT JSON;

create sink delta_lake_sink from source
with (
    connector = 'deltalake',
    type = 'append-only', 
    location = 's3a://deltalake/delta',
    s3.access.key = 'hummockadmin',
    s3.secret.key = 'hummockadmin',
    s3.endpoint = 'http://localhost:9301'
);

6. Wait several seconds and type flush in the psql client to ensure that some data are flushed to the sink. Execute the following SQL from spark sql shell to query deltalake data from Spark.

select * from delta.`s3a://deltalake/delta` limit 10;

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

Documentation

• My PR contains user-facing changes.

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• Installation and deployment
• Connector (sources & sinks)
• SQL commands, functions, and operators
• RisingWave cluster configuration changes
• Other (please specify in the release note below)

Release note

[tabVersion]

should we mark it as user-facing?

[codecov]

Codecov Report

Merging #10374 (3a22ec9) into main (5c6b25c) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@           Coverage Diff           @@
##             main   #10374   +/-   ##
=======================================
  Coverage   70.48%   70.48%           
=======================================
  Files        1257     1257           
  Lines      214185   214185           
=======================================
+ Hits       150959   150960    +1     
+ Misses      63226    63225    -1     

Flag	Coverage Δ
rust	70.48% <100.00%> (+<0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/connector/src/sink/remote.rs	57.07% <ø> (ø)
...rc/frontend/src/optimizer/plan_node/stream_sink.rs	70.70% <100.00%> (ø)

... and 1 file with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more