CREATE TABLE source_opengauss
(
     std_id      integer
    ,std_name    varchar
    ,std_sex     varchar
    ,std_birth   timestamp(0)
    ,std_in      timestamp(0)
    ,std_address varchar
) WITH (
    'connector' = 'opengauss-x',
    'url' = 'jdbc:opengauss://x.x.x.x:5432/school',
    'table-name' = 'student',
    'username' = 'test',
    'password' = 'xxx',
--     'scan.fetch-size' = '2',
    'scan.query-timeout' = '10'
    ,'scan.custom-sql' = 'SELECT * from student where std_id>40'
);

CREATE TABLE sink_opengauss (
     std_id      integer
    ,std_name    varchar
    ,std_sex     varchar
    ,std_birth   timestamp(0)
    ,std_in      timestamp(0)
    ,std_address varchar
) WITH (
      'connector' = 'opengauss-x',
      'url' = 'jdbc:opengauss://x.x.x.x:5432/school',
      'table-name' = 'di_test',
      'username' = 'test',
      'password' = 'xxxxx',
--       'sink.buffer-flush.max-rows' = '1',
      'sink.all-replace' = 'true'
      );

CREATE TABLE sink_stream (
    std_id      integer
    ,std_name    varchar
    ,std_sex     varchar
    ,std_birth   timestamp(0)
    ,std_in      timestamp(0)
    ,std_address varchar
) WITH (
    'connector' = 'stream-x'
);

-- INSERT INTO sink_stream SELECT *  FROM source_opengauss;
INSERT INTO sink_opengauss SELECT * FROM source_opengauss;