yuanoOo commented on code in PR #4061:
URL: https://github.com/apache/flink-cdc/pull/4061#discussion_r2224819428
##########
docs/content/docs/connectors/flink-sources/oceanbase-cdc.md:
##########
@@ -433,434 +498,429 @@ Available Metadata
The following format metadata can be exposed as read-only (VIRTUAL) columns in
a table definition.
Review Comment:
This is copied from the mysql-cdc document. It is best to keep it consistent.
##########
docs/content/docs/connectors/flink-sources/oceanbase-cdc.md:
##########
@@ -433,434 +498,429 @@ Available Metadata
The following format metadata can be exposed as read-only (VIRTUAL) columns in
a table definition.
<table class="colwidths-auto docutils">
- <thead>
- <tr>
- <th class="text-left" style="width: 15%">Key</th>
- <th class="text-left" style="width: 30%">DataType</th>
- <th class="text-left" style="width: 55%">Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td>tenant_name</td>
- <td>STRING</td>
- <td>Name of the tenant that contains the row.</td>
- </tr>
- <tr>
- <td>database_name</td>
- <td>STRING</td>
- <td>Name of the database that contains the row.</td>
- </tr>
- <tr>
- <td>schema_name</td>
- <td>STRING</td>
- <td>Name of the schema that contains the row.</td>
- </tr>
- <tr>
- <td>table_name</td>
- <td>STRING NOT NULL</td>
- <td>Name of the table that contains the row.</td>
- </tr>
- <tr>
- <td>op_ts</td>
- <td>TIMESTAMP_LTZ(3) NOT NULL</td>
- <td>It indicates the time that the change was made in the
database. <br>
- If the record is read from snapshot of the table instead of
the change stream, the value is always 0.</td>
- </tr>
- </tbody>
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 15%">Key</th>
+ <th class="text-left" style="width: 30%">DataType</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>table_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the table that contain the row.</td>
+ </tr>
+ <tr>
+ <td>database_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the database that contain the row.</td>
+ </tr>
+ <tr>
+ <td>op_ts</td>
+ <td>TIMESTAMP_LTZ(3) NOT NULL</td>
+ <td>It indicates the time that the change was made in the database.
<br>If the record is read from snapshot of the table instead of the binlog, the
value is always 0.</td>
+ </tr>
+ <tr>
+ <td>row_kind</td>
+ <td>STRING NOT NULL</td>
+ <td>It indicates the row kind of the changelog,Note: The downstream SQL
operator may fail to compare due to this new added column when processing the
row retraction if
+the source operator chooses to output the 'row_kind' column for each record.
It is recommended to use this metadata column only in simple synchronization
jobs.
+<br>'+I' means INSERT message, '-D' means DELETE message, '-U' means
UPDATE_BEFORE message and '+U' means UPDATE_AFTER message.</td>
+ </tr>
+ </tbody>
</table>
The extended CREATE TABLE example demonstrates the syntax for exposing these
metadata fields:
```sql
-CREATE TABLE products (
- tenant_name STRING METADATA FROM 'tenant_name' VIRTUAL,
- db_name STRING METADATA FROM 'database_name' VIRTUAL,
- table_name STRING METADATA FROM 'table_name' VIRTUAL,
- operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
- order_id INT,
- order_date TIMESTAMP(0),
+CREATE TABLE products
+(
+ db_name STRING METADATA FROM 'database_name' VIRTUAL,
+ table_name STRING METADATA FROM 'table_name' VIRTUAL,
+ operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+ operation STRING METADATA FROM 'row_kind' VIRTUAL,
+ order_id INT,
+ order_date TIMESTAMP(0),
customer_name STRING,
- price DECIMAL(10, 5),
- product_id INT,
- order_status BOOLEAN,
- PRIMARY KEY(order_id) NOT ENFORCED
+ price DECIMAL(10, 5),
+ product_id INT,
+ order_status BOOLEAN,
+ PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
- 'connector' = 'oceanbase-cdc',
- 'scan.startup.mode' = 'initial',
- 'username' = 'user@test_tenant',
- 'password' = 'pswd',
- 'tenant-name' = 'test_tenant',
- 'database-name' = '^test_db$',
- 'table-name' = '^orders$',
- 'hostname' = '127.0.0.1',
- 'port' = '2881',
- 'rootserver-list' = '127.0.0.1:2882:2881',
- 'logproxy.host' = '127.0.0.1',
- 'logproxy.port' = '2983',
- 'working-mode' = 'memory'
-);
+ 'connector' = 'oceanbase-cdc',
+ 'hostname' = 'localhost',
+ 'port' = '2881',
+ 'username' = 'root',
+ 'password' = '123456',
+ 'database-name' = 'mydb',
+ 'table-name' = 'orders'
+ );
```
-Features
---------
-
-### At-Least-Once Processing
-
-The OceanBase CDC connector is a Flink Source connector which will read
database snapshot first and then continues to read change events with
**at-least-once processing**.
-
-OceanBase is a kind of distributed database whose log files are distributed on
different servers. As there is no position information like MySQL binlog
offset, we can only use timestamp as the position mark. In order to ensure the
completeness of reading data, `liboblog` (a C++ library to read OceanBase log
record) might read some log data before the given timestamp. So in this way we
may read duplicate data whose timestamp is around the start point, and only
'at-least-once' can be guaranteed.
-
-### Startup Reading Position
-
-The config option `scan.startup.mode` specifies the startup mode for OceanBase
CDC consumer. The valid enumerations are:
-
-- `initial`: Performs an initial snapshot on the monitored table upon first
startup, and continue to read the latest commit log.
-- `latest-offset`: Never to perform snapshot on the monitored table upon first
startup and just read the latest commit log since the connector is started.
-- `timestamp`: Never to perform snapshot on the monitored table upon first
startup and just read the commit log from the given `scan.startup.timestamp`.
-- `snapshot`: Only perform snapshot on the monitored table.
-
-### Consume Commit Log
+The extended CREATE TABLE example demonstrates the usage of regex to match
multi-tables:
-The OceanBase CDC Connector using
[oblogclient](https://github.com/oceanbase/oblogclient) to consume commit log
from OceanBase LogProxy.
-
-### DataStream Source
-
-The OceanBase CDC connector can also be a DataStream source. You can create a
SourceFunction as the following shows:
-
-```java
-import org.apache.flink.cdc.connectors.base.options.StartupOptions;
-import org.apache.flink.cdc.connectors.oceanbase.OceanBaseSource;
-import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-
-public class OceanBaseSourceExample {
- public static void main(String[] args) throws Exception {
- SourceFunction<String> oceanBaseSource =
- OceanBaseSource.<String>builder()
- .startupOptions(StartupOptions.initial())
- .hostname("127.0.0.1")
- .port(2881)
- .username("user@test_tenant")
- .password("pswd")
- .compatibleMode("mysql")
- .jdbcDriver("com.mysql.cj.jdbc.Driver")
- .tenantName("test_tenant")
- .databaseName("^test_db$")
- .tableName("^test_table$")
- .logProxyHost("127.0.0.1")
- .logProxyPort(2983)
- .rsList("127.0.0.1:2882:2881")
- .serverTimeZone("+08:00")
- .deserializer(new JsonDebeziumDeserializationSchema())
- .build();
-
- StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
-
- // enable checkpoint
- env.enableCheckpointing(3000);
-
- env.addSource(oceanBaseSource).print().setParallelism(1);
- env.execute("Print OceanBase Snapshot + Change Events");
- }
-}
+```sql
+CREATE TABLE products
+(
+ db_name STRING METADATA FROM 'database_name' VIRTUAL,
+ table_name STRING METADATA FROM 'table_name' VIRTUAL,
+ operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+ operation STRING METADATA FROM 'row_kind' VIRTUAL,
+ order_id INT,
+ order_date TIMESTAMP(0),
+ customer_name STRING,
+ price DECIMAL(10, 5),
+ product_id INT,
+ order_status BOOLEAN,
+ PRIMARY KEY (order_id) NOT ENFORCED
+) WITH (
+ 'connector' = 'oceanbase-cdc',
+ 'hostname' = 'localhost',
+ 'port' = '2881',
+ 'username' = 'root',
+ 'password' = '123456',
+ 'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})',
+ 'table-name' = '(t[5-8]|tt)'
+ );
```
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 15%">example</th>
+ <th class="text-left" style="width: 30%">expression</th>
+ <th class="text-left" style="width: 55%">description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>prefix match</td>
+ <td>^(test).*</td>
+ <td>This matches the database name or table name starts with prefix of
test, e.g test1、test2. </td>
+ </tr>
+ <tr>
+ <td>suffix match</td>
+ <td>.*[p$]</td>
+ <td>This matches the database name or table name ends with suffix of p,
e.g cdcp、edcp. </td>
+ </tr>
+ <tr>
+ <td>specific match</td>
+ <td>txc</td>
+ <td>This matches the database name or table name according to a specific
name, e.g txc. </td>
+ </tr>
+ </tbody>
+</table>
-### Available Source metrics
-
-Metrics can help understand the progress of assignments, and the following are
the supported [Flink
metrics](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/):
-
-| Group | Name | Type | Description
|
-|------------------------|----------------------------|-------|-----------------------------------------------------|
-| namespace.schema.table | isSnapshotting | Gauge | Weather the
table is snapshotting or not |
-| namespace.schema.table | isStreamReading | Gauge | Weather the
table is stream reading or not |
-| namespace.schema.table | numTablesSnapshotted | Gauge | The number of
tables that have been snapshotted |
-| namespace.schema.table | numTablesRemaining | Gauge | The number of
tables that have not been snapshotted |
-| namespace.schema.table | numSnapshotSplitsProcessed | Gauge | The number of
splits that is being processed |
-| namespace.schema.table | numSnapshotSplitsRemaining | Gauge | The number of
splits that have not been processed |
-| namespace.schema.table | numSnapshotSplitsFinished | Gauge | The number of
splits that have been processed |
-| namespace.schema.table | snapshotStartTime | Gauge | The time when
the snapshot started |
-| namespace.schema.table | snapshotEndTime | Gauge | The time when
the snapshot ended |
+It will use `database-name\\.table-name` as a pattern to match tables, as
above examples using pattern
`(^(test).*|^(tpc).*|txc|.*[p$]|t{2})\\.(t[5-8]|tt)` matches txc.tt、test2.test5.
Review Comment:
This is copied from the mysql-cdc document. It is best to keep it consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
