[Improve][Connector-V2][SelectDB Cloud]Refactor some SelectDB Cloud Sink code as well as support copy into batch and async flush and cdc

[zy-kkk]

Purpose of this pull request

Refactor some SelectDB Cloud Sink code as well as support copy into batch and async flush and cdc

Test procedure

SelectDB Table Schema

CREATE  TABLE `TypeTest` (
    c_map varchar(1000),
    c_array array<int>,
    c_string String,
    c_boolean boolean,
    c_tinyint tinyint,
    c_smallint smallint,
    c_int int,
    c_bigint bigint,
    c_float float,
    c_double double,
    c_decimal  decimal(27, 9),
    c_null String,
    c_bytes String,
    c_date date,
    c_timestamp datetime
) ENGINE=OLAP
DUPLICATE KEY(`c_map`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`c_map`) BUCKETS 4

1. ST Engine
   • config

   env {
      execution.parallelism = 1
      job.mode = "BATCH"
      checkpoint.interval = 5000
    }

      source {
        FakeSource {
          row.num = 10
          parallelism = 1
          schema = {
            fields {
              c_map = "map<string, string>"
              c_array = "array<int>"
              c_string = string
              c_boolean = boolean
              c_tinyint = tinyint
              c_smallint = smallint
              c_int = int
              c_bigint = bigint
              c_float = float
              c_double = double
              c_decimal = "decimal(27, 9)"
              c_null = "null"
              c_bytes = bytes
              c_date = date
              c_timestamp = timestamp
            }
          }
          result_table_name = "fake"
        }
      }
      
      transform {
      }
      
      sink {
        SelectDBCloud {
          load-url="ip:port"
          jdbc-url="ip:port"
          cluster-name="SeaTunnel"
          table.identifier="test.TypeTest"
          username="admin"
          password="******"
          selectdb.config {
              file.type="json"
          }
        }
      }

• result
  
  

2. Flink
   • config

   env {
        execution.parallelism = 1
        job.mode = "BATCH"
    }

      source {
        FakeSource {
          row.num = 10
          parallelism = 1
          schema = {
            fields {
              c_map = "map<string, string>"
              c_array = "array<int>"
              c_string = string
              c_boolean = boolean
              c_tinyint = tinyint
              c_smallint = smallint
              c_int = int
              c_bigint = bigint
              c_float = float
              c_double = double
              c_decimal = "decimal(27, 9)"
              c_null = "null"
              c_bytes = bytes
              c_date = date
              c_timestamp = timestamp
            }
          }
          result_table_name = "fake"
        }
      }
      
      transform {
      }
      
      sink {
        SelectDBCloud {
          load-url="ip:port"
          jdbc-url="ip:port"
          cluster-name="SeaTunnel"
          table.identifier="test.TypeTest"
          username="admin"
          password="******"
          selectdb.config {
              file.type="json"
          }
        }
      }

• result
  
  

3. Spark
   • Config

  env {
        job.name = "SeaTunnel"
        spark.executor.instances = 1
        spark.executor.cores = 1
        spark.executor.memory = "1g"
        spark.master = local
      }
      
      source {
        FakeSource {
          row.num = 10
          parallelism = 1
          schema = {
            fields {
              c_map = "map<string, string>"
              c_array = "array<int>"
              c_string = string
              c_boolean = boolean
              c_tinyint = tinyint
              c_smallint = smallint
              c_int = int
              c_bigint = bigint
              c_float = float
              c_double = double
              c_decimal = "decimal(27, 9)"
              c_null = "null"
              c_bytes = bytes
              c_date = date
              c_timestamp = timestamp
            }
          }
          result_table_name = "fake"
        }
      }
      
      transform {
      }
      
      sink {
        SelectDBCloud {
          load-url="ip:port"
          jdbc-url="ip:port"
          cluster-name="SeaTunnel"
          table.identifier="test.TypeTest"
          username="admin"
          password="******"
          selectdb.config {
              file.type="json"
          }
        }
      }

• result

Check list

• Code changed are covered with tests, or it does not need tests for reason:
• 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/incubator-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.

[TyrantLucifer]

cc @CalvinKirs