CREATE TABLE doris_sink_table (
  id INT,
  name VARCHAR
) WITH (
  'connector' = 'doris', -- Fixed value 'doris'.
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  'table.identifier' = 'test.sales_order', -- Doris table name. Format: db.tbl.
  username' = 'root', -- Username for accessing Doris, with write permission to the database.
  password' = 'password', -- Password for accessing Doris.
  'sink.batch.size' = '500', -- Maximum number of rows per BE write.
  'sink.batch.interval' = '1s' -- Flush interval. After this time, the asynchronous thread will write data in the cache to the backend. The default value is 1s, and supported time units include ms, s, min, h, and d. Setting to 0 means disabling periodic writes.
);

CREATE TABLE doris_sink_table (
  id INT,
  name VARCHAR
) WITH (
  'connector' = 'doris', -- Fixed value 'doris'.
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  'table.identifier' = 'test.sales_order', -- Doris table name. Format: db.tbl.
  username' = 'root', -- Username for accessing Doris, with write permission to the database.
  password' = 'password' -- Password for accessing Doris.
);

-- Note: 2PC is enabled by default.

CREATE CATALOG doris_catalog WITH (
  'type' = 'doris',
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  username' = 'root', -- Username for accessing Doris, with write permission to the database.
  password' = 'password', -- Password for accessing Doris.
  'default-database' = 'default'
  )

----- Flink 1.13 -----
CREATE TABLE datagen_source_table ( 
    id INT,
    name STRING
) WITH ( 
  'connector' = 'datagen',
  'rows-per-second'='1' -- Number of data rows generated per second.
);

CREATE TABLE doris_sink_table (
  id INT,
  name STRING
) WITH (
  'connector' = 'doris', -- Fixed value 'doris'.
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  'table.identifier' = 'test.sales_order', -- Doris table name. Format: db.tbl.
  username' = 'root', -- Username for accessing Doris, with write permission to the database.
  password' = 'password', -- Password for accessing Doris.
  'sink.batch.size' = '500', -- Maximum number of rows per BE write.
  sink.batch.interval' = '1s' -- Flush interval. After this time, the asynchronous thread will write data in the cache to the backend. The default value is 1s, and the supported time units include ms, s, min, h, and d. Setting to 0 means disabling periodic writes.
);

INSERT INTO doris_sink_table select * from datagen_source_table;


CREATE TABLE datagen_source_table ( 
    id INT,
    name STRING
) WITH ( 
  'connector' = 'datagen',
  'rows-per-second'='1' -- Number of data rows generated per second.
);

CREATE TABLE doris_sink_table (
  id INT,
  name STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = 'FE_IP:HTTP_PORT',
  'table.identifier' = 'db.table',
  'username' = 'root',
  'password' = 'password',
  'sink.label-prefix' = 'doris_label'
);

INSERT INTO doris_sink_table select * from datagen_source_table;

CREATE CATALOG doris_catalog WITH (
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  username' = 'root', -- Username for accessing Doris, with write permission to the database.
  password' = 'password', -- Password for accessing Doris.
  'default-database' = 'default'
);

CREATE TABLE datagen_source_table (
    id INT,
    name STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second'='1' -- Number of data rows generated per second.
);

INSERT INTO doris_catalog.my_database.my_table SELECT * FROM.datagen_source_table;

--mysql cdc source table.
CREATE TABLE mysql_cdc_source_table (
  id     INT NOT NULL,
  name   VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc' -- Fixed value 'mysql-cdc'.
  'hostname' = 'YourHostName', -- IP address of the database.
  port' = '3306', -- Port for database access.
  'username' = 'YourUserName', -- Username for database access (SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, and RELOAD permissions are required).
  password' = 'YourPassword', -- Password for database access.
  'database-name' = 'YourDatabase', -- Database to be synchronized.
  'table-name' = 'YourTable' -- Name of the data table to be synchronized.     
);

-- Write to Doris table.
CREATE TABLE print_table (
  id       INT,
  name     STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'doris', -- Fixed value 'doris'.
  'fenodes' = 'FE_IP:FE_HTTP_PORT', -- Doris FE HTTP address.
  'table.identifier' = 'dbName.tableName', -- Doris table name. Format: db.tbl.
  username' = 'YourUserName', -- Username for accessing Doris, with write permission to the database.
  password' = 'YourPassword', -- Password for accessing Doris.
);

insert into print_table
select id,name from mysql_cdc_source_table;

-- Table creation statement for the Uniq model.
CREATE TABLE doris_sink_table (
    id   int(11),
    name varchar(1024)
) 
UNIQUE KEY(id) 
DISTRIBUTED BY HASH(id) BUCKETS 10
PROPERTIES("replication_num" = "3");

-- Table creation statement for the Aggregate model.
CREATE TABLE doris_sink_table (
    id   int(11),
    name varchar(1024) REPLACE DEFAULT '0'
) 
AGGREGATE KEY('id')
DISTRIBUTED BY HASH(id) BUCKETS 10
PROPERTIES("replication_num" = "3"); -- Note that if the backend nodes are insufficient, the error Failed to find enough host in all backends will occur. You can reduce this value as needed.

CREATE USER 'test' IDENTIFIED BY 'test_passwd';
GRANT ALL ON test TO test;