[Feature][Connector-V2] Starrocks implements multi table sink

[jw-itq]

Purpose of this pull request

#8455

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[hailin0]

Please add e2e and update docs

[jw-itq]

Please add e2e and update docs

ok,thanks!

[hailin0]

Why not reuse the testcase below:

https://github.com/apache/seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/fake-to-starrocks.conf

or

https://github.com/apache/seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/mysqlcdc_to_starrocks_with_schema_change.conf

[jw-itq]

Thanks, let me modify this. I'm not using fake-to-starrocks.conf because I want to test incremental data. I'm not using mysqlcdc_to_starrocks_with_schema_change.conf because it uses shop.sql from mysql-cdc, which doesn't contain multiple tables. I'm going to modify this now.

[jw-itq]

Sorry, there was a problem with my local code. I have fixed it now. Please help me take a look again. Thank you！

[hailin0]

LGTM cc @Hisoka-X

[ASDFSA13]

这个pr有问题, 其实还是没解决
对于mysql8.0.28版本的数据库
PRIMARY KEY (id),
UNIQUE KEY uni_str (str_value_sha256,seq,ledger_date,is_delete) USING BTREE,
KEY idx_status (seller_id,market_place_id,is_delete) USING BTREE,
KEY idx_created_time (create_time) USING BTREE,
KEY idx_ledger_date (ledger_date,event_type,sync_status) USING BTREE

表结构如上，这时候代码，对于表结构如果主键id为 PRIMARY KEY 的时候, 他是只会从 UNIQUE KEY 取字段生效, 而不从 PRIMARY KEY 取字段生效

String tableSc = splitColumnsConfig.getOrDefault(tableId.catalog() + "." + tableId.table(), null);

if (StringUtils.isNotEmpty(tableSc)) {
AtomicBoolean isUniqueKey = new AtomicBoolean(false);
dialect.getUniqueKeys(jdbc, tableId).forEach(ck -> ck.getColumnNames().forEach(ckc -> {
if (tableSc.equals(ckc.getColumnName())) {
isUniqueKey.set(true);
}
}));

if (isUniqueKey.get()) { return column; }
else { log.warn("Config snapshotSplitColumn not unique key for table {}", tableId); }

}