This is an automated email from the ASF dual-hosted git repository.
wanghailin pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 0d08b20061 [Improve][Connector-v2] Optimize the count table rows for
jdbc-oracle and oracle-cdc (#7248)
0d08b20061 is described below
commit 0d08b200619bdc2017236c46d4134e56f095e738
Author: dailai <dai...@chinatelecom.cn>
AuthorDate: Thu Jul 25 10:00:13 2024 +0800
[Improve][Connector-v2] Optimize the count table rows for jdbc-oracle and
oracle-cdc (#7248)
---
docs/en/connector-v2/source/Jdbc.md | 198 ++++++---------------
docs/en/connector-v2/source/Oracle-CDC.md | 40 +++++
.../cdc/oracle/config/OracleSourceConfig.java | 9 +
.../oracle/config/OracleSourceConfigFactory.java | 16 ++
.../cdc/oracle/source/OracleIncrementalSource.java | 2 +
.../source/OracleIncrementalSourceFactory.java | 2 +
.../cdc/oracle/source/OracleSourceOptions.java | 12 ++
.../source/eumerator/OracleChunkSplitter.java | 6 +-
.../seatunnel/cdc/oracle/utils/OracleUtils.java | 55 +++---
.../seatunnel/jdbc/config/JdbcSourceOptions.java | 12 ++
.../jdbc/config/JdbcSourceTableConfig.java | 8 +
.../internal/dialect/oracle/OracleDialect.java | 39 ++--
.../seatunnel/jdbc/source/JdbcSourceFactory.java | 4 +
.../seatunnel/jdbc/source/JdbcSourceTable.java | 2 +
.../seatunnel/jdbc/utils/JdbcCatalogUtils.java | 2 +
.../seatunnel/cdc/oracle/OracleCDCIT.java | 31 +++-
...dc_to_console.conf => oraclecdc_to_oracle.conf} | 0
...conf => oraclecdc_to_oracle_skip_analysis.conf} | 3 +-
...f => oraclecdc_to_oracle_use_select_count.conf} | 3 +-
.../connectors/seatunnel/jdbc/JdbcOracleIT.java | 28 ++-
.../jdbc_oracle_source_to_sink_use_select1.conf | 58 ++++++
.../jdbc_oracle_source_to_sink_use_select2.conf | 58 ++++++
.../jdbc_oracle_source_to_sink_use_select3.conf | 59 ++++++
23 files changed, 466 insertions(+), 181 deletions(-)
diff --git a/docs/en/connector-v2/source/Jdbc.md
b/docs/en/connector-v2/source/Jdbc.md
index 7655083172..7fab8d50b2 100644
--- a/docs/en/connector-v2/source/Jdbc.md
+++ b/docs/en/connector-v2/source/Jdbc.md
@@ -39,104 +39,32 @@ supports query SQL and can achieve projection effect.
## Options
-| name | type | required | default
value |
-|--------------------------------------------|--------|----------|-----------------|
-| url | String | Yes | -
|
-| driver | String | Yes | -
|
-| user | String | No | -
|
-| password | String | No | -
|
-| query | String | No | -
|
-| compatible_mode | String | No | -
|
-| connection_check_timeout_sec | Int | No | 30
|
-| partition_column | String | No | -
|
-| partition_upper_bound | Long | No | -
|
-| partition_lower_bound | Long | No | -
|
-| partition_num | Int | No | job
parallelism |
-| fetch_size | Int | No | 0
|
-| properties | Map | No | -
|
-| table_path | String | No | -
|
-| table_list | Array | No | -
|
-| where_condition | String | No | -
|
-| split.size | Int | No | 8096
|
-| split.even-distribution.factor.lower-bound | Double | No | 0.05
|
-| split.even-distribution.factor.upper-bound | Double | No | 100
|
-| split.sample-sharding.threshold | Int | No | 1000
|
-| split.inverse-sampling.rate | Int | No | 1000
|
-| common-options | | No | -
|
-
-### driver [string]
-
-The jdbc class name used to connect to the remote data source, if you use
MySQL the value is `com.mysql.cj.jdbc.Driver`.
-
-### user [string]
-
-userName
-
-### password [string]
-
-password
-
-### url [string]
-
-The URL of the JDBC connection. Refer to a case:
jdbc:postgresql://localhost/test
-
-### query [string]
-
-Query statement
-
-### compatible_mode [string]
-
-The compatible mode of database, required when the database supports multiple
compatible modes. For example, when using OceanBase database, you need to set
it to 'mysql' or 'oracle'.
-
-### connection_check_timeout_sec [int]
-
-The time in seconds to wait for the database operation used to validate the
connection to complete.
-
-### fetch_size [int]
-
-For queries that return a large number of objects, you can configure the row
fetch size used in the query to
-improve performance by reducing the number database hits required to satisfy
the selection criteria. Zero means use jdbc default value.
-
-### properties
-
-Additional connection configuration parameters,when properties and URL have
the same parameters, the priority is determined by the <br/>specific
implementation of the driver. For example, in MySQL, properties take precedence
over the URL.
-
-### table_path
-
-The path to the full path of table, you can use this configuration instead of
`query`.
-
-examples:
-- mysql: "testdb.table1"
-- oracle: "test_schema.table1"
-- sqlserver: "testdb.test_schema.table1"
-- postgresql: "testdb.test_schema.table1"
-- iris: "test_schema.table1"
-
-### table_list
-
-The list of tables to be read, you can use this configuration instead of
`table_path`
-
-example
-
-```hocon
-table_list = [
- {
- table_path = "testdb.table1"
- }
- {
- table_path = "testdb.table2"
- query = "select * from testdb.table2 where id > 100"
- }
-]
-```
-
-### where_condition
-
-Common row filter conditions for all tables/queries, must start with `where`.
for example `where id > 100`
-
-### common options
-
-Source plugin common parameters, please refer to [Source Common
Options](common-options.md) for details.
+| name | type | required | default
value |
description
[...]
+|--------------------------------------------|---------|----------|-----------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| url | String | Yes | -
| The URL of the JDBC connection. Refer to a case:
jdbc:postgresql://localhost/test
[...]
+| driver | String | Yes | -
| The jdbc class name used to connect to the remote data source, if you
use MySQL the value is `com.mysql.cj.jdbc.Driver`.
[...]
+| user | String | No | -
| userName
[...]
+| password | String | No | -
| password
[...]
+| query | String | No | -
| Query statement
[...]
+| compatible_mode | String | No | -
| The compatible mode of database, required when the database supports
multiple compatible modes. For example, when using OceanBase database, you need
to set it to 'mysql' or 'oracle'.
[...]
+| connection_check_timeout_sec | Int | No | 30
| The time in seconds to wait for the database operation used to validate
the connection to complete.
[...]
+| partition_column | String | No | -
| The column name for split data.
[...]
+| partition_upper_bound | Long | No | -
| The partition_column max value for scan, if not set SeaTunnel will query
database get max value.
[...]
+| partition_lower_bound | Long | No | -
| The partition_column min value for scan, if not set SeaTunnel will query
database get min value.
[...]
+| partition_num | Int | No | job
parallelism | Not recommended for use, The correct approach is to control the
number of split through `split.size`<br/> How many splits do we need to split
into, only support positive integer. default value is job parallelism.
[...]
+| use_select_count | Boolean | No | false
| Use select count for table count rather then other methods in dynamic
chunk split stage. This is currently only available for jdbc-oracle.In this
scenario, select count directly is used when it is faster to update statistics
using sql from analysis table
[...]
+| skip_analyze | Boolean | No | false
| Skip the analysis of table count in dynamic chunk split stage. This is
currently only available for jdbc-oracle.In this scenario, you schedule
analysis table sql to update related table statistics periodically or your
table data does not change frequently
[...]
+| fetch_size | Int | No | 0
| For queries that return a large number of objects, you can configure the
row fetch size used in the query to improve performance by reducing the number
database hits required to satisfy the selection criteria. Zero means use jdbc
default value.
[...]
+| properties | Map | No | -
| Additional connection configuration parameters,when properties and URL
have the same parameters, the priority is determined by the <br/>specific
implementation of the driver. For example, in MySQL, properties take precedence
over the URL.
[...]
+| table_path | String | No | -
| The path to the full path of table, you can use this configuration
instead of `query`. <br/>examples: <br/>`- mysql: "testdb.table1" `<br/>`-
oracle: "test_schema.table1" `<br/>`- sqlserver: "testdb.test_schema.table1"`
<br/>`- postgresql: "testdb.test_schema.table1"` <br/>`- iris:
"test_schema.table1"`
[...]
+| table_list | Array | No | -
| The list of tables to be read, you can use this configuration instead of
`table_path`
[...]
+| where_condition | String | No | -
| Common row filter conditions for all tables/queries, must start with
`where`. for example `where id > 100`
[...]
+| split.size | Int | No | 8096
| How many rows in one split, captured tables are split into multiple
splits when read of table.
[...]
+| split.even-distribution.factor.lower-bound | Double | No | 0.05
| Not recommended for use.<br/> The lower bound of the chunk key
distribution factor. This factor is used to determine whether the table data is
evenly distributed. If the distribution factor is calculated to be greater than
or equal to this lower bound (i.e., (MAX(id) - MIN(id) + 1) / row count), the
table chunks would be optimized for even distribution. Otherwise, if the
distribution factor is less, the [...]
+| split.even-distribution.factor.upper-bound | Double | No | 100
| Not recommended for use.<br/> The upper bound of the chunk key
distribution factor. This factor is used to determine whether the table data is
evenly distributed. If the distribution factor is calculated to be less than or
equal to this upper bound (i.e., (MAX(id) - MIN(id) + 1) / row count), the
table chunks would be optimized for even distribution. Otherwise, if the
distribution factor is greater, the [...]
+| split.sample-sharding.threshold | Int | No | 1000
| This configuration specifies the threshold of estimated shard count to
trigger the sample sharding strategy. When the distribution factor is outside
the bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strat [...]
+| split.inverse-sampling.rate | Int | No | 1000
| The inverse of the sampling rate used in the sample sharding strategy.
For example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is pref [...]
+| common-options | | No | -
| Source plugin common parameters, please refer to [Source Common
Options](common-options.md) for details.
[...]
## Parallel Reader
@@ -152,50 +80,6 @@ The JDBC Source connector supports parallel reading of data
from tables. SeaTunn
* Number(int, bigint, decimal, ...)
* Date
-### Options Related To Split
-
-#### split.size
-
-How many rows in one split, captured tables are split into multiple splits
when read of table.
-
-#### split.even-distribution.factor.lower-bound
-
-> Not recommended for use
-
-The lower bound of the chunk key distribution factor. This factor is used to
determine whether the table data is evenly distributed. If the distribution
factor is calculated to be greater than or equal to this lower bound (i.e.,
(MAX(id) - MIN(id) + 1) / row count), the table chunks would be optimized for
even distribution. Otherwise, if the distribution factor is less, the table
will be considered as unevenly distributed and the sampling-based sharding
strategy will be used if the estim [...]
-
-#### split.even-distribution.factor.upper-bound
-
-> Not recommended for use
-
-The upper bound of the chunk key distribution factor. This factor is used to
determine whether the table data is evenly distributed. If the distribution
factor is calculated to be less than or equal to this upper bound (i.e.,
(MAX(id) - MIN(id) + 1) / row count), the table chunks would be optimized for
even distribution. Otherwise, if the distribution factor is greater, the table
will be considered as unevenly distributed and the sampling-based sharding
strategy will be used if the estim [...]
-
-#### split.sample-sharding.threshold
-
-This configuration specifies the threshold of estimated shard count to trigger
the sample sharding strategy. When the distribution factor is outside the
bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy will be used. This can help to handle large datasets
more efficiently. The default [...]
-
-#### split.inverse-sampling.rate
-
-The inverse of the sampling rate used in the sample sharding strategy. For
example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is preferred. The default value is 1000.
-
-#### partition_column [string]
-
-The column name for split data.
-
-#### partition_upper_bound [BigDecimal]
-
-The partition_column max value for scan, if not set SeaTunnel will query
database get max value.
-
-#### partition_lower_bound [BigDecimal]
-
-The partition_column min value for scan, if not set SeaTunnel will query
database get min value.
-
-#### partition_num [int]
-
-> Not recommended for use, The correct approach is to control the number of
split through `split.size`
-
-How many splits do we need to split into, only support positive integer.
default value is job parallelism.
-
## tips
> If the table can not be split(for example, table have no Primary Key or
> Unique Index, and `partition_column` is not set), it will run in single
> concurrency.
@@ -235,6 +119,35 @@ there are some reference value for params above.
### simple
+#### Case 1
+
+```
+Jdbc {
+ url = "jdbc:mysql://localhost/test?serverTimezone=GMT%2b8"
+ driver = "com.mysql.cj.jdbc.Driver"
+ connection_check_timeout_sec = 100
+ user = "root"
+ password = "123456"
+ query = "select * from type_bin"
+}
+```
+
+#### Case 2 Use the select count(*) instead of analysis table for count table
rows in dynamic chunk split stage
+
+```
+Jdbc {
+ url = "jdbc:mysql://localhost/test?serverTimezone=GMT%2b8"
+ driver = "com.mysql.cj.jdbc.Driver"
+ connection_check_timeout_sec = 100
+ user = "root"
+ password = "123456"
+ use_select_count = true
+ query = "select * from type_bin"
+}
+```
+
+#### Case 3 Use the select NUM_ROWS from all_tables for the table rows but
skip the analyze table.
+
```
Jdbc {
url = "jdbc:mysql://localhost/test?serverTimezone=GMT%2b8"
@@ -242,6 +155,7 @@ Jdbc {
connection_check_timeout_sec = 100
user = "root"
password = "123456"
+ skip_analyze = true
query = "select * from type_bin"
}
```
diff --git a/docs/en/connector-v2/source/Oracle-CDC.md
b/docs/en/connector-v2/source/Oracle-CDC.md
index 2dfffedc66..cedbda141f 100644
--- a/docs/en/connector-v2/source/Oracle-CDC.md
+++ b/docs/en/connector-v2/source/Oracle-CDC.md
@@ -244,6 +244,8 @@ exit;
| sample-sharding.threshold | Integer | No | 1000
| This configuration specifies the threshold of estimated shard count to
trigger the sample sharding strategy. When the distribution factor is outside
the bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy [...]
| inverse-sampling.rate | Integer | No | 1000
| The inverse of the sampling rate used in the sample sharding strategy. For
example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is preferr [...]
| exactly_once | Boolean | No | false
| Enable exactly once semantic.
[...]
+| use_select_count | Boolean | No | false
| Use select count for table count rather then other methods in full stage.In
this scenario, select count directly is used when it is faster to update
statistics using sql from analysis table
[...]
+| skip_analyze | Boolean | No | false
| Skip the analysis of table count in full stage.In this scenario, you
schedule analysis table sql to update related table statistics periodically or
your table data does not change frequently
[...]
| format | Enum | No |
DEFAULT | Optional output format for Oracle CDC, valid enumerations are
`DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| debezium | Config | No | -
| Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/oracle.adoc#connector-properties)
to Debezium Embedded Engine which is used to capture data changes from Oracle
server.
[...]
| common-options | | no | -
| Source plugin common parameters, please refer to [Source Common
Options](common-options.md) for details
[...]
@@ -270,6 +272,44 @@ source {
}
```
+> Use the select count(*) instead of analysis table for count table rows in
full stage
+>
+> ```conf
+> source {
+> # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+> Oracle-CDC {
+> result_table_name = "customers"
+> use_select_count = true
+> username = "system"
+> password = "oracle"
+> database-names = ["XE"]
+> schema-names = ["DEBEZIUM"]
+> table-names = ["XE.DEBEZIUM.FULL_TYPES"]
+> base-url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+> source.reader.close.timeout = 120000
+> }
+> }
+> ```
+>
+> Use the select NUM_ROWS from all_tables for the table rows but skip the
analyze table.
+>
+> ```conf
+> source {
+> # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+> Oracle-CDC {
+> result_table_name = "customers"
+> skip_analyze = true
+> username = "system"
+> password = "oracle"
+> database-names = ["XE"]
+> schema-names = ["DEBEZIUM"]
+> table-names = ["XE.DEBEZIUM.FULL_TYPES"]
+> base-url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
+> source.reader.close.timeout = 120000
+> }
+> }
+> ```
+
### Support custom primary key for table
```
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfig.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfig.java
index 5cdf1e9eec..32bcb41f78 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfig.java
@@ -24,6 +24,7 @@ import
org.apache.seatunnel.connectors.cdc.base.config.StopConfig;
import io.debezium.config.Configuration;
import io.debezium.connector.oracle.OracleConnectorConfig;
import io.debezium.relational.RelationalTableFilters;
+import lombok.Getter;
import java.util.List;
import java.util.Properties;
@@ -32,11 +33,17 @@ import java.util.Properties;
* Describes the connection information of the Oracle database and the
configuration information for
* performing snapshotting and streaming reading, such as splitSize.
*/
+@Getter
public class OracleSourceConfig extends JdbcSourceConfig {
private static final long serialVersionUID = 1L;
+ private final Boolean useSelectCount;
+ private final Boolean skipAnalyze;
+
public OracleSourceConfig(
+ Boolean useSelectCount,
+ Boolean skipAnalyze,
StartupConfig startupConfig,
StopConfig stopConfig,
List<String> databaseList,
@@ -82,6 +89,8 @@ public class OracleSourceConfig extends JdbcSourceConfig {
connectMaxRetries,
connectionPoolSize,
exactlyOnce);
+ this.useSelectCount = useSelectCount;
+ this.skipAnalyze = skipAnalyze;
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java
index 8e175bd7fe..d6018083c2 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java
@@ -38,6 +38,10 @@ public class OracleSourceConfigFactory extends
JdbcSourceConfigFactory {
private static final String DRIVER_CLASS_NAME =
"oracle.jdbc.driver.OracleDriver";
private List<String> schemaList;
+
+ private Boolean useSelectCount;
+
+ private Boolean skipAnalyze;
/**
* An optional list of regular expressions that match schema names to be
monitored; any schema
* name not included in the whitelist will be excluded from monitoring. By
default all
@@ -48,6 +52,16 @@ public class OracleSourceConfigFactory extends
JdbcSourceConfigFactory {
return this;
}
+ public JdbcSourceConfigFactory useSelectCount(Boolean useSelectCount) {
+ this.useSelectCount = useSelectCount;
+ return this;
+ }
+
+ public JdbcSourceConfigFactory skipAnalyze(Boolean skipAnalyze) {
+ this.skipAnalyze = skipAnalyze;
+ return this;
+ }
+
/** Creates a new {@link OracleSourceConfig} for the given subtask {@code
subtaskId}. */
public OracleSourceConfig create(int subtask) {
@@ -123,6 +137,8 @@ public class OracleSourceConfigFactory extends
JdbcSourceConfigFactory {
}
return new OracleSourceConfig(
+ useSelectCount,
+ skipAnalyze,
startupConfig,
stopConfig,
databaseList,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
index f3357f46e0..a1bbd0cb25 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
@@ -88,6 +88,8 @@ public class OracleIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceC
configFactory.startupOptions(startupConfig);
configFactory.stopOptions(stopConfig);
configFactory.schemaList(config.get(OracleSourceOptions.SCHEMA_NAMES));
+
configFactory.useSelectCount(config.get(OracleSourceOptions.USE_SELECT_COUNT));
+
configFactory.skipAnalyze(config.get(OracleSourceOptions.SKIP_ANALYZE));
configFactory.originUrl(config.get(JdbcCatalogOptions.BASE_URL));
return configFactory;
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
index 2a0dc6b290..21e08c2af7 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
@@ -61,6 +61,8 @@ public class OracleIncrementalSourceFactory implements
TableSourceFactory {
JdbcCatalogOptions.BASE_URL,
JdbcSourceOptions.DATABASE_NAMES,
OracleSourceOptions.SCHEMA_NAMES,
+ OracleSourceOptions.USE_SELECT_COUNT,
+ OracleSourceOptions.SKIP_ANALYZE,
JdbcSourceOptions.SERVER_TIME_ZONE,
JdbcSourceOptions.CONNECT_TIMEOUT_MS,
JdbcSourceOptions.CONNECT_MAX_RETRIES,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSourceOptions.java
index e6bbd77a41..f87ea1ccf5 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSourceOptions.java
@@ -53,4 +53,16 @@ public class OracleSourceOptions {
.listType()
.noDefaultValue()
.withDescription("Schema name of the database to
monitor.");
+
+ public static final Option<Boolean> USE_SELECT_COUNT =
+ Options.key("use_select_count")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Use select count for table count in full
stage");
+
+ public static final Option<Boolean> SKIP_ANALYZE =
+ Options.key("skip_analyze")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Skip the analysis of table count in full
stage");
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/eumerator/OracleChunkSplitter.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/eumerator/OracleChunkSplitter.java
index 6525c3a2db..21cfebcd47 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/eumerator/OracleChunkSplitter.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/eumerator/OracleChunkSplitter.java
@@ -23,6 +23,7 @@ import
org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.dialect.JdbcDataSourceDialect;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.splitter.AbstractJdbcSourceChunkSplitter;
import org.apache.seatunnel.connectors.cdc.base.utils.ObjectUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.config.OracleSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.utils.OracleTypeUtils;
import org.apache.seatunnel.connectors.seatunnel.cdc.oracle.utils.OracleUtils;
@@ -41,8 +42,11 @@ import java.sql.SQLException;
@Slf4j
public class OracleChunkSplitter extends AbstractJdbcSourceChunkSplitter {
+ private final OracleSourceConfig oracleSourceConfig;
+
public OracleChunkSplitter(JdbcSourceConfig sourceConfig,
JdbcDataSourceDialect dialect) {
super(sourceConfig, dialect);
+ this.oracleSourceConfig = (OracleSourceConfig) sourceConfig;
}
@Override
@@ -80,7 +84,7 @@ public class OracleChunkSplitter extends
AbstractJdbcSourceChunkSplitter {
@Override
public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
throws SQLException {
- return OracleUtils.queryApproximateRowCnt(jdbc, tableId);
+ return OracleUtils.queryApproximateRowCnt(oracleSourceConfig, jdbc,
tableId);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java
index 1994bd6e03..fbb3664be0 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.utils.SeaTunnelException;
import org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.config.OracleSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.offset.RedoLogOffset;
import org.apache.kafka.connect.source.SourceRecord;
@@ -81,27 +82,41 @@ public class OracleUtils {
});
}
- public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId
tableId)
+ public static long queryApproximateRowCnt(
+ OracleSourceConfig oracleSourceConfig, JdbcConnection jdbc,
TableId tableId)
throws SQLException {
- final String analyzeTable =
- String.format(
- "analyze table %s compute statistics for table",
- quoteSchemaAndTable(tableId));
- final String rowCountQuery =
- String.format(
- "select NUM_ROWS from all_tables where TABLE_NAME =
'%s'", tableId.table());
- return jdbc.execute(analyzeTable)
- .queryAndMap(
- rowCountQuery,
- rs -> {
- if (!rs.next()) {
- throw new SQLException(
- String.format(
- "No result returned after
running query [%s]",
- rowCountQuery));
- }
- return rs.getLong(1);
- });
+ Boolean useSelectCount = oracleSourceConfig.getUseSelectCount();
+ String rowCountQuery;
+ if (useSelectCount) {
+ rowCountQuery = String.format("select count(*) from %s",
quoteSchemaAndTable(tableId));
+ } else {
+ rowCountQuery =
+ String.format(
+ "select NUM_ROWS from all_tables where TABLE_NAME
= '%s'",
+ tableId.table());
+ Boolean skipAnalyze = oracleSourceConfig.getSkipAnalyze();
+ if (!skipAnalyze) {
+ final String analyzeTable =
+ String.format(
+ "analyze table %s compute statistics for
table",
+ quoteSchemaAndTable(tableId));
+ // not skip analyze
+ log.info("analyze table sql: {}", analyzeTable);
+ jdbc.execute(analyzeTable);
+ }
+ }
+ log.info("row count query: {}", rowCountQuery);
+ return jdbc.queryAndMap(
+ rowCountQuery,
+ rs -> {
+ if (!rs.next()) {
+ throw new SQLException(
+ String.format(
+ "No result returned after running
query [%s]",
+ rowCountQuery));
+ }
+ return rs.getLong(1);
+ });
}
public static Object queryMin(
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
index 14ea587335..6647d9c8eb 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
@@ -93,4 +93,16 @@ public interface JdbcSourceOptions {
+ "The value represents the denominator of
the sampling rate fraction. "
+ "For example, a value of 1000 means a
sampling rate of 1/1000. "
+ "This parameter is used when the sample
sharding strategy is triggered.");
+
+ Option<Boolean> USE_SELECT_COUNT =
+ Options.key("use_select_count")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Use select count for table count");
+
+ Option<Boolean> SKIP_ANALYZE =
+ Options.key("skip_analyze")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Skip the analysis of table count");
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceTableConfig.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceTableConfig.java
index b4a6e890df..d217a0b745 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceTableConfig.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceTableConfig.java
@@ -58,6 +58,12 @@ public class JdbcSourceTableConfig implements Serializable {
@JsonProperty("partition_upper_bound")
private BigDecimal partitionEnd;
+ @JsonProperty("use_select_count")
+ private Boolean useSelectCount;
+
+ @JsonProperty("skip_analyze")
+ private Boolean skipAnalyze;
+
@Tolerate
public JdbcSourceTableConfig() {}
@@ -79,6 +85,8 @@ public class JdbcSourceTableConfig implements Serializable {
.partitionNumber(connectorConfig.get(JdbcOptions.PARTITION_NUM))
.partitionStart(connectorConfig.get(JdbcOptions.PARTITION_LOWER_BOUND))
.partitionEnd(connectorConfig.get(JdbcOptions.PARTITION_UPPER_BOUND))
+
.useSelectCount(connectorConfig.get(JdbcSourceOptions.USE_SELECT_COUNT))
+
.skipAnalyze(connectorConfig.get(JdbcSourceOptions.SKIP_ANALYZE))
.build();
tableList = Collections.singletonList(tableProperty);
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
index 1bf1466949..e1aee7f7d8 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
@@ -180,34 +180,47 @@ public class OracleDialect implements JdbcDialect {
public Long approximateRowCntStatement(Connection connection,
JdbcSourceTable table)
throws SQLException {
- // 1. If no query is configured, use TABLE STATUS.
- // 2. If a query is configured but does not contain a WHERE clause and
tablePath is
+ // 1. Use select count
+ // 2. If no query is configured, use TABLE STATUS.
+ // 3. If a query is configured but does not contain a WHERE clause and
tablePath is
// configured, use TABLE STATUS.
- // 3. If a query is configured with a WHERE clause, or a query
statement is configured but
+ // 4. If a query is configured with a WHERE clause, or a query
statement is configured but
// tablePath is TablePath.DEFAULT, use COUNT(*).
+ String query = table.getQuery();
+
boolean useTableStats =
- StringUtils.isBlank(table.getQuery())
- || (!table.getQuery().toLowerCase().contains("where")
+ StringUtils.isBlank(query)
+ || (!query.toLowerCase().contains("where")
&& table.getTablePath() != null
&& !TablePath.DEFAULT
.getFullName()
.equals(table.getTablePath().getFullName()));
+ if (table.getUseSelectCount()) {
+ useTableStats = false;
+ if (StringUtils.isBlank(query)) {
+ query = "SELECT * FROM " +
tableIdentifier(table.getTablePath());
+ }
+ }
+
if (useTableStats) {
TablePath tablePath = table.getTablePath();
- String analyzeTable =
- String.format(
- "analyze table %s compute statistics for table",
- tableIdentifier(tablePath));
String rowCountQuery =
String.format(
"select NUM_ROWS from all_tables where OWNER =
'%s' AND TABLE_NAME = '%s' ",
tablePath.getSchemaName(),
tablePath.getTableName());
-
try (Statement stmt = connection.createStatement()) {
- log.info("Split Chunk, approximateRowCntStatement: {}",
analyzeTable);
- stmt.execute(analyzeTable);
+ String analyzeTable =
+ String.format(
+ "analyze table %s compute statistics for
table",
+ tableIdentifier(tablePath));
+ if (!table.getSkipAnalyze()) {
+ log.info("Split Chunk, approximateRowCntStatement: {}",
analyzeTable);
+ stmt.execute(analyzeTable);
+ } else {
+ log.warn("Skip analyze, approximateRowCntStatement: {}",
analyzeTable);
+ }
log.info("Split Chunk, approximateRowCntStatement: {}",
rowCountQuery);
try (ResultSet rs = stmt.executeQuery(rowCountQuery)) {
if (!rs.next()) {
@@ -220,7 +233,7 @@ public class OracleDialect implements JdbcDialect {
}
}
}
- return SQLUtils.countForSubquery(connection, table.getQuery());
+ return SQLUtils.countForSubquery(connection, query);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
index 54e8d5173b..b9ca90ed53 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
@@ -46,6 +46,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcOptions.
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcOptions.QUERY;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcOptions.URL;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcOptions.USER;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.SKIP_ANALYZE;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.SPLIT_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.SPLIT_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.SPLIT_INVERSE_SAMPLING_RATE;
@@ -53,6 +54,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOp
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.SPLIT_SIZE;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.TABLE_LIST;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.TABLE_PATH;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.USE_SELECT_COUNT;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceOptions.WHERE_CONDITION;
@Slf4j
@@ -94,6 +96,8 @@ public class JdbcSourceFactory implements TableSourceFactory {
COMPATIBLE_MODE,
PROPERTIES,
QUERY,
+ USE_SELECT_COUNT,
+ SKIP_ANALYZE,
TABLE_PATH,
WHERE_CONDITION,
TABLE_LIST,
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceTable.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceTable.java
index fea7382472..8aad94c8b6 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceTable.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceTable.java
@@ -37,5 +37,7 @@ public class JdbcSourceTable implements Serializable {
private final Integer partitionNumber;
private final BigDecimal partitionStart;
private final BigDecimal partitionEnd;
+ private final Boolean useSelectCount;
+ private final Boolean skipAnalyze;
private final CatalogTable catalogTable;
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java
index 83d5bfa769..a689632206 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java
@@ -90,6 +90,8 @@ public class JdbcCatalogUtils {
.partitionNumber(tableConfig.getPartitionNumber())
.partitionStart(tableConfig.getPartitionStart())
.partitionEnd(tableConfig.getPartitionEnd())
+
.useSelectCount(tableConfig.getUseSelectCount())
+
.skipAnalyze(tableConfig.getSkipAnalyze())
.catalogTable(catalogTable)
.build();
tables.put(tablePath, jdbcSourceTable);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
index 125d57915c..0192fae3f7 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
@@ -137,7 +137,22 @@ public class OracleCDCIT extends TestSuiteBase implements
TestResource {
@TestTemplate
public void testOracleCdcCheckDataE2e(TestContainer container) throws
Exception {
+ checkDataForTheJob(container, "/oraclecdc_to_oracle.conf", false);
+ }
+
+ @TestTemplate
+ public void testOracleCdcCheckDataE2eForUseSelectCount(TestContainer
container)
+ throws Exception {
+ checkDataForTheJob(container,
"/oraclecdc_to_oracle_use_select_count.conf", false);
+ }
+ @TestTemplate
+ public void testOracleCdcCheckDataE2eForSkipAnalysis(TestContainer
container) throws Exception {
+ checkDataForTheJob(container,
"/oraclecdc_to_oracle_skip_analysis.conf", true);
+ }
+
+ private void checkDataForTheJob(
+ TestContainer container, String jobConfPath, Boolean skipAnalysis)
throws Exception {
clearTable(DATABASE, SOURCE_TABLE1);
clearTable(DATABASE, SOURCE_TABLE2);
clearTable(DATABASE, SINK_TABLE1);
@@ -145,10 +160,24 @@ public class OracleCDCIT extends TestSuiteBase implements
TestResource {
insertSourceTable(DATABASE, SOURCE_TABLE1);
+ if (skipAnalysis) {
+ // analyzeTable before execute job
+ String analyzeTable =
+ String.format(
+ "analyze table "
+ + "\"DEBEZIUM\".\"FULL_TYPES\" "
+ + "compute statistics for table");
+ log.info("analyze table {}", analyzeTable);
+ try (Connection connection = testConnection(ORACLE_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute(analyzeTable);
+ }
+ }
+
CompletableFuture.supplyAsync(
() -> {
try {
- container.executeJob("/oraclecdc_to_console.conf");
+ container.executeJob(jobConfPath);
} catch (Exception e) {
log.error("Commit task exception :" + e.getMessage());
throw new RuntimeException(e);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle.conf
similarity index 100%
copy from
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
copy to
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle.conf
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_skip_analysis.conf
similarity index 98%
copy from
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
copy to
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_skip_analysis.conf
index b915da6f90..233fc735ef 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_skip_analysis.conf
@@ -29,6 +29,7 @@ source {
# This is a example source plugin **only for test and demonstrate the
feature source plugin**
Oracle-CDC {
result_table_name = "customers"
+ skip_analyze = true
username = "system"
password = "top_secret"
database-names = ["ORCLCDB"]
@@ -62,4 +63,4 @@ Jdbc {
primary_keys = ["ID"]
connection.pool.size = 1
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_use_select_count.conf
similarity index 98%
rename from
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
rename to
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_use_select_count.conf
index b915da6f90..dd93cd4f9e 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_console.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_use_select_count.conf
@@ -29,6 +29,7 @@ source {
# This is a example source plugin **only for test and demonstrate the
feature source plugin**
Oracle-CDC {
result_table_name = "customers"
+ use_select_count = true
username = "system"
password = "top_secret"
database-names = ["ORCLCDB"]
@@ -62,4 +63,4 @@ Jdbc {
primary_keys = ["ID"]
connection.pool.size = 1
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleIT.java
index 70c9d39cf4..9d3597c435 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleIT.java
@@ -28,6 +28,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceTable;
import org.apache.commons.lang3.tuple.Pair;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.OracleContainer;
@@ -40,6 +41,7 @@ import com.google.common.collect.Lists;
import java.math.BigDecimal;
import java.sql.Date;
+import java.sql.Statement;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -63,7 +65,11 @@ public class JdbcOracleIT extends AbstractJdbcIT {
private static final String SINK_TABLE = "E2E_TABLE_SINK";
private static final String CATALOG_TABLE = "E2E_TABLE_CATALOG";
private static final List<String> CONFIG_FILE =
- Lists.newArrayList("/jdbc_oracle_source_to_sink.conf");
+ Lists.newArrayList(
+ "/jdbc_oracle_source_to_sink.conf",
+ "/jdbc_oracle_source_to_sink_use_select1.conf",
+ "/jdbc_oracle_source_to_sink_use_select2.conf",
+ "/jdbc_oracle_source_to_sink_use_select3.conf");
private static final String CREATE_SQL =
"create table %s\n"
@@ -163,7 +169,7 @@ public class JdbcOracleIT extends AbstractJdbcIT {
@Override
Pair<String[], List<SeaTunnelRow>> initTestData() {
List<SeaTunnelRow> rows = new ArrayList<>();
- for (int i = 0; i < 100; i++) {
+ for (int i = 0; i < 20000; i++) {
SeaTunnelRow row =
new SeaTunnelRow(
new Object[] {
@@ -237,4 +243,22 @@ public class JdbcOracleIT extends AbstractJdbcIT {
SCHEMA);
catalog.open();
}
+
+ @BeforeAll
+ @Override
+ public void startUp() {
+ super.startUp();
+ // analyzeTable before execute job
+ String analyzeTable =
+ String.format(
+ "analyze table "
+ + quoteIdentifier(SOURCE_TABLE)
+ + " compute statistics for table");
+ log.info("analyze table {}", analyzeTable);
+ try (Statement stmt = connection.createStatement()) {
+ stmt.execute(analyzeTable);
+ } catch (Exception e) {
+ log.error("Error when analyze table", e);
+ }
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select1.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select1.conf
new file mode 100644
index 0000000000..8a0c831044
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select1.conf
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in
seatunnel config
+######
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ use_select_count = true
+ query = "SELECT
VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL
FROM E2E_TABLE_SOURCE"
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of source plugins,
+ # please go to
https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
+}
+
+sink {
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ query = "INSERT INTO E2E_TABLE_SINK
(VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of sink plugins,
+ # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select2.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select2.conf
new file mode 100644
index 0000000000..ebebdb5505
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select2.conf
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in
seatunnel config
+######
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ use_select_count = true
+ table_path = TESTUSER.E2E_TABLE_SOURCE
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of source plugins,
+ # please go to
https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
+}
+
+sink {
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ query = "INSERT INTO E2E_TABLE_SINK
(VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of sink plugins,
+ # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select3.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select3.conf
new file mode 100644
index 0000000000..d00ce9b643
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink_use_select3.conf
@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in
seatunnel config
+######
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ use_select_count = false
+ skip_analyze = true
+ table_path = TESTUSER.E2E_TABLE_SOURCE
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of source plugins,
+ # please go to
https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
+}
+
+sink {
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
+ user = testUser
+ password = testPassword
+ query = "INSERT INTO E2E_TABLE_SINK
(VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
+ properties {
+ database.oracle.jdbc.timezoneAsRegion = "false"
+ }
+ }
+
+ # If you would like to get more information about how to configure seatunnel
and see full list of sink plugins,
+ # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
+}
