[Feature] [InfluxDB Source] add read by chunk

[15767714253]

Purpose of this pull request

This pull request enhances the InfluxDB source connector by introducing chunked data retrieval. The feature allows the connector to efficiently process large datasets by querying and fetching data in chunks.

Does this PR introduce any user-facing change?

No,The existing functionality of the InfluxDB source connector remains unchanged. Users who upgrade to the version containing this patch will have the additional option to enable chunked data retrieval for performance improvements, but this is not a breaking change, and the feature is opt-in.

How was this patch tested?

I have conducted multiple synchronizations in our company's production environment, with a single table containing more than 300 million data. However, this feature puts a significant load on the resources of InfluxDB itself, which is related to the size of the data read.

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 change log that in connector document. For more details you can refer to connector-v2
  2. Update plugin-mapping.properties and add new connector information in it
  3. Update the pom file of seatunnel-dist
• Update the release-note.

[Hisoka-X]

cc @hailin0 @EricJoy2048

[zhilinli123]

#4572
Change the document to v2 format
reference：
https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/source/Mysql.md
https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Mysql.md

@15767714253

[zhilinli123]

Key Features

[zhilinli123]

Key Features

[zhilinli123]

Refer to mysql Documentation

[zhilinli123]

identical?

[15767714253]

Using split_column is for shard querying, using chunk_size is for chunk querying, and using neither is for regular querying.

[EricJoy2048]

Repeated

[EricJoy2048]

log.error?

[15767714253]

log.info

[EricJoy2048]

1. Even if users set split_key, they can still use Chunk's read mode. Can this be understood as follows?
2. I think using Chunk mode to read should be a better way. A better approach is to give a default value if the user has not set chunk_size, and then use the readBychunkSize method uniformly to read the data.

[15767714253]

I think this is not advisable, because I believe that chunked reading puts a lot of performance pressure on InfluxDB, and some users would still prefer to shift the pressure to SeaTunnel.

[EricJoy2048]

I think this is not advisable, because I believe that chunked reading puts a lot of performance pressure on InfluxDB, and some users would still prefer to shift the pressure to SeaTunnel.

Can the pressure on InfluxDB be resolved by reducing parallelism (e.g. setting parallelism to 1) or config speed limit config in env https://seatunnel.apache.org/docs/2.3.5/concept/speed-limit?

[15767714253]

The parallelism was originally 1; it has nothing to do with the speed of reading, but rather the consumption of InfluxDB's own chunk operation.

[EricJoy2048]

Use CountDownLatch is not a good idea. You can update the code like this: (Then method readByChunkSize will become a sync methods and will not return until the reading thread is completed )

final CompletableFuture<Void> queryCompleteFeature = new CompletableFuture<>();

influxdb.query(
                new Query(split.getQuery(), config.getDatabase()),
config.getChunkSize(),
                (cancellable, queryResult) -> {
                    if (cancellable.isCanceled()) {
                        log.info("this chunk reader influxDB is canceled");
                        queryCompleteFeature.complete();
                        return;
                    }
                    if (queryResult.hasError()) {
                        log.error(
                                "this chunk reader influxDB result has error [{}]",
                                queryResult.getError());
                        queryCompleteFeature.completeExceptionally(new InfluxdbConnectorException)
                        return;
                    }
                    for (QueryResult.Result result : queryResult.getResults()) {
                        List<QueryResult.Series> serieList = result.getSeries();
                        if (CollectionUtils.isNotEmpty(serieList)) {
                            for (QueryResult.Series series : serieList) {
                                for (List<Object> values : series.getValues()) {
                                    SeaTunnelRow row =
                                            InfluxDBRowConverter.convert(
                                                    values, seaTunnelRowType, columnsIndexList);
                                    output.collect(row);
                                }
                            }
                        } else {
                            log.info("this chunk reader influxDB series is empty");
                        }
                    }
                },
                () -> {
                    log.info("this chunk reader influxDB complete");
                    queryCompleteFeature.complete();
                },
                throwable -> {
                    log.error(
                            "this chunk reader influxDB result has error [{}]",
                            throwable.getMessage());
                    queryCompleteFeature.completeExceptionally(new InfluxdbConnectorException(throwable))
                });
    }

   queryCompleteFeature.get();

[15767714253]

Yes, I've already tried to switch to CompletableFuture yesterday, but I haven't got it working yet due to limited time. I know CountDownLatch is not a good choice.

[15767714253]

But merely changing this method to a synchronous one is not enough, as it occupies the lock in the pollNext method.

[EricJoy2048]

But merely changing this method to a synchronous one is not enough, as it occupies the lock in the pollNext method.

It is normal to occupy a lock, and we must ensure that the lock(output.getCheckpointLock()) cannot be released until a split read is completed