This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 78df51cf0f [Feature][Connector-V2][MaxCompute] Support Schema Name and
Passwordless Auth (#10831)
78df51cf0f is described below
commit 78df51cf0fdcb10dda3c04697380d25d5a7026d0
Author: zhiliang-wu <[email protected]>
AuthorDate: Tue May 12 16:29:58 2026 +0200
[Feature][Connector-V2][MaxCompute] Support Schema Name and Passwordless
Auth (#10831)
Co-authored-by: WU Zhiliang (External) <[email protected]>
---
docs/en/connectors/sink/Maxcompute.md | 21 ++-
docs/en/connectors/source/Maxcompute.md | 23 ++-
docs/zh/connectors/sink/Maxcompute.md | 21 ++-
docs/zh/connectors/source/Maxcompute.md | 27 +++-
.../connector-maxcompute/pom.xml | 22 ++-
.../maxcompute/catalog/MaxComputeCatalog.java | 34 +++--
.../catalog/MaxComputeCatalogFactory.java | 6 +-
.../maxcompute/config/MaxcomputeBaseOptions.java | 14 ++
.../maxcompute/sink/MaxcomputeSinkFactory.java | 9 +-
.../maxcompute/source/MaxcomputeSource.java | 16 +++
.../maxcompute/source/MaxcomputeSourceFactory.java | 7 +-
.../maxcompute/util/MaxcomputeOutputFormat.java | 23 ++-
.../seatunnel/maxcompute/util/MaxcomputeUtil.java | 43 +++++-
.../maxcompute/util/MaxcomputeUtilTest.java | 157 +++++++++++++++++++++
14 files changed, 371 insertions(+), 52 deletions(-)
diff --git a/docs/en/connectors/sink/Maxcompute.md
b/docs/en/connectors/sink/Maxcompute.md
index 347bba8777..4d6ee7fb1b 100644
--- a/docs/en/connectors/sink/Maxcompute.md
+++ b/docs/en/connectors/sink/Maxcompute.md
@@ -16,11 +16,13 @@ Used to read data from Maxcompute.
| name | type | required | default value |
|----------------|---------|----------|---------------|
-| accessId | string | yes | - |
-| accesskey | string | yes | - |
+| accessId | string | no | - |
+| accesskey | string | no | - |
+| sts_token | string | no | - |
| endpoint | string | yes | - |
| project | string | yes | - |
| table_name | string | yes | - |
+| schema_name | string | no | - |
| partition_spec | string | no | - |
| overwrite | boolean | no | false |
| insert_strategy| string | no | upload |
@@ -34,6 +36,13 @@ Used to read data from Maxcompute.
`accesskey` Your Maxcompute accessKey which cloud be access from Alibaba Cloud.
+### sts_token [string]
+
+`sts_token` Your MaxCompute STS Token for temporary authentication. **Note:**
If `sts_token` is provided, `accessId` and `accesskey` are strictly required.
+
+> **Passwordless Authentication (ECS RAM Role, Environment Variables, etc.)**
+> To use passwordless authentication seamlessly, simply leave `accessId`,
`accesskey`, and `sts_token` all blank. The connector will automatically fall
back to the Aliyun DefaultCredentialsProvider chain (Environment Variables,
System Properties, CLI Profiles, OIDC, ECS RAM Roles).
+
### endpoint [string]
`endpoint` Your Maxcompute endpoint start with http.
@@ -50,6 +59,14 @@ Used to read data from Maxcompute.
`partition_spec` This spec of Maxcompute partition table eg:ds='20220101'.
+### schema_name [string]
+
+`schema_name` The MaxCompute Schema name (the namespace between Project and
Table).
+Only required when the table resides in a **non-default schema** within your
MaxCompute project.
+See [Schema-related
operations](https://www.alibabacloud.com/help/en/maxcompute/user-guide/schema-related-operations).
+
+Default: not set (uses the project default schema).
+
### overwrite [boolean]
`overwrite` Whether to overwrite the table or partition, default: false.
diff --git a/docs/en/connectors/source/Maxcompute.md
b/docs/en/connectors/source/Maxcompute.md
index b1203f4fca..9395a857a7 100644
--- a/docs/en/connectors/source/Maxcompute.md
+++ b/docs/en/connectors/source/Maxcompute.md
@@ -20,11 +20,13 @@ Used to read data from Maxcompute.
| name | type | required | default value |
|----------------|--------|----------|---------------|
-| accessId | string | yes | - |
-| accesskey | string | yes | - |
+| accessId | string | no | - |
+| accesskey | string | no | - |
+| sts_token | string | no | - |
| endpoint | string | yes | - |
| project | string | yes | - |
| table_name | string | yes | - |
+| schema_name | string | no | - |
| partition_spec | string | no | - |
| split_row | int | no | 10000 |
| read_columns | Array | no | - |
@@ -40,6 +42,13 @@ Used to read data from Maxcompute.
`accesskey` Your Maxcompute accessKey which cloud be access from Alibaba Cloud.
+### sts_token [string]
+
+`sts_token` Your MaxCompute STS Token for temporary authentication. **Note:**
If `sts_token` is provided, `accessId` and `accesskey` are strictly required.
+
+> **Passwordless Authentication (ECS RAM Role, Environment Variables, etc.)**
+> To use passwordless authentication seamlessly, simply leave `accessId`,
`accesskey`, and `sts_token` all blank. The connector will automatically fall
back to the Aliyun DefaultCredentialsProvider chain (Environment Variables,
System Properties, CLI Profiles, OIDC, ECS RAM Roles).
+
### endpoint [string]
`endpoint` Your Maxcompute endpoint start with http.
@@ -56,6 +65,16 @@ Used to read data from Maxcompute.
`partition_spec` This spec of Maxcompute partition table eg:ds='20220101'.
+### schema_name [string]
+
+`schema_name` The MaxCompute Schema name (the namespace between Project and
Table).
+Only required when the table resides in a **non-default schema** within your
MaxCompute project.
+See [Schema-related
operations](https://www.alibabacloud.com/help/en/maxcompute/user-guide/schema-related-operations).
+
+When using `table_list`, each entry can specify its own `schema_name`, which
overrides the top-level value.
+
+Default: not set (uses the project default schema).
+
### split_row [int]
`split_row` Number of rows per split, default: 10000.
diff --git a/docs/zh/connectors/sink/Maxcompute.md
b/docs/zh/connectors/sink/Maxcompute.md
index eeb3f9710f..87f3984d9c 100644
--- a/docs/zh/connectors/sink/Maxcompute.md
+++ b/docs/zh/connectors/sink/Maxcompute.md
@@ -16,11 +16,13 @@ import ChangeLog from
'../changelog/connector-maxcompute.md';
| 参数名 | 类型 | 必须 | 默认值 |
|----------------|---------|------|--------|
-| accessId | string | 是 | - |
-| accesskey | string | 是 | - |
+| accessId | string | 否 | - |
+| accesskey | string | 否 | - |
+| sts_token | string | 否 | - |
| endpoint | string | 是 | - |
| project | string | 是 | - |
| table_name | string | 是 | - |
+| schema_name | string | 否 | - |
| partition_spec | string | 否 | - |
| overwrite | boolean | 否 | false |
| insert_strategy| string | no | upload |
@@ -34,6 +36,13 @@ import ChangeLog from '../changelog/connector-maxcompute.md';
`accesskey` 您的 Maxcompute accessKey，可从阿里云访问。
+### sts_token [string]
+
+`sts_token` 您的 MaxCompute STS Token，用于临时认证。 **注意：** 如果提供了 `sts_token`，则必须同时提供
`accessId` 和 `accesskey`。
+
+> **免密认证 (ECS RAM Role, 环境变量等)**
+> 要使用免密认证，只需将 `accessId`、`accesskey` 和 `sts_token` 全部留空不填。连接器将自动回退到阿里云默认凭据链
(DefaultCredentialsProvider) 读取凭证（包括环境变量、系统属性、CLI 配置文件、OIDC 以及 ECS RAM 角色）。
+
### endpoint [string]
`endpoint` 您的 Maxcompute endpoint，以 http 开头。
@@ -50,6 +59,14 @@ import ChangeLog from '../changelog/connector-maxcompute.md';
`partition_spec` Maxcompute 分区表的规范，例如：ds='20220101'。
+### schema_name [string]
+
+`schema_name` MaxCompute Schema 名称（Project 与 Table 之间的命名空间）。
+仅当表位于 MaxCompute 项目的**非默认 Schema** 时才需要设置。
+参见 [Schema
相关操作](https://help.aliyun.com/zh/maxcompute/user-guide/schema-related-operations)。
+
+默认值：不设置（使用项目默认 Schema）。
+
### overwrite [boolean]
`overwrite` 是否覆盖表或分区，默认值：false。
diff --git a/docs/zh/connectors/source/Maxcompute.md
b/docs/zh/connectors/source/Maxcompute.md
index 4048fb00af..78b7f86591 100644
--- a/docs/zh/connectors/source/Maxcompute.md
+++ b/docs/zh/connectors/source/Maxcompute.md
@@ -20,11 +20,13 @@ import ChangeLog from
'../changelog/connector-maxcompute.md';
| 名称 | 类型 | 必需 | 默认值 |
|----------------|--------|----|---------------|
-| accessId | string | 是 | - |
-| accesskey | string | 是 | - |
+| accessId | string | 否 | - |
+| accesskey | string | 否 | - |
+| sts_token | string | 否 | - |
| endpoint | string | 是 | - |
| project | string | 是 | - |
| table_name | string | 是 | - |
+| schema_name | string | 否 | - |
| partition_spec | string | 否 | - |
| split_row | int | 否 | 10000 |
| read_columns | Array | 否 | - |
@@ -34,11 +36,18 @@ import ChangeLog from
'../changelog/connector-maxcompute.md';
### accessId [string]
-`accessId` 您的 Maxcompute 密钥 Id, 可以从阿里云访问哪个云.
+`accessId` 您的 Maxcompute 密钥 Id.
### accesskey [string]
-`accesskey` Your Maxcompute 密钥, 可以从阿里云访问哪个云.
+`accesskey` 您的 Maxcompute 密钥.
+
+### sts_token [string]
+
+`sts_token` 您的 MaxCompute STS Token，用于临时认证。 **注意：** 如果提供了 `sts_token`，则必须同时提供
`accessId` 和 `accesskey`。
+
+> **免密认证 (ECS RAM Role, 环境变量等)**
+> 要使用免密认证，只需将 `accessId`、`accesskey` 和 `sts_token` 全部留空不填。连接器将自动回退到阿里云默认凭据链
(DefaultCredentialsProvider) 读取凭证（包括环境变量、系统属性、CLI 配置文件、OIDC 以及 ECS RAM 角色）。
### endpoint [string]
@@ -56,6 +65,16 @@ import ChangeLog from '../changelog/connector-maxcompute.md';
`partition_spec` Maxcompute分区表的此规范，例如:ds='20220101'.
+### schema_name [string]
+
+`schema_name` MaxCompute Schema 名称（Project 与 Table 之间的命名空间）。
+仅当表位于 MaxCompute 项目的**非默认 Schema** 时才需要设置。
+参见 [Schema
相关操作](https://help.aliyun.com/zh/maxcompute/user-guide/schema-related-operations)。
+
+使用 `table_list` 时，每个条目可以单独指定 `schema_name`，会覆盖顶层的值。
+
+默认值：不设置（使用项目默认 Schema）。
+
### split_row [int]
`split_row` 每次拆分的行数，默认值: 10000.
diff --git a/seatunnel-connectors-v2/connector-maxcompute/pom.xml
b/seatunnel-connectors-v2/connector-maxcompute/pom.xml
index 00ce56b703..e095231668 100644
--- a/seatunnel-connectors-v2/connector-maxcompute/pom.xml
+++ b/seatunnel-connectors-v2/connector-maxcompute/pom.xml
@@ -30,7 +30,7 @@
<name>SeaTunnel : Connectors V2 : Maxcompute</name>
<properties>
- <maxcompute.version>0.51.0</maxcompute.version>
+ <maxcompute.version>0.51.2</maxcompute.version>
<commons.lang3.version>3.18.0</commons.lang3.version>
</properties>
@@ -47,13 +47,13 @@
</dependency>
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- <version>${commons.lang3.version}</version>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
- <artifactId>connector-common</artifactId>
+ <artifactId>seatunnel-commons-lang3</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
@@ -63,6 +63,18 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
+ <dependencies>
+ <dependency>
+ <groupId>org.ow2.asm</groupId>
+ <artifactId>asm</artifactId>
+ <version>9.6</version>
+ </dependency>
+ <dependency>
+ <groupId>org.ow2.asm</groupId>
+ <artifactId>asm-commons</artifactId>
+ <version>9.6</version>
+ </dependency>
+ </dependencies>
<executions>
<execution>
<goals>
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalog.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalog.java
index 36ed0deb20..9c7fc5c7d9 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalog.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalog.java
@@ -46,7 +46,6 @@ import com.aliyun.odps.Projects;
import com.aliyun.odps.Table;
import com.aliyun.odps.Tables;
import com.aliyun.odps.account.Account;
-import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.task.SQLTask;
import com.aliyun.odps.type.TypeInfo;
import lombok.extern.slf4j.Slf4j;
@@ -73,10 +72,7 @@ public class MaxComputeCatalog implements Catalog {
@Override
public void open() throws CatalogException {
- account =
- new AliyunAccount(
- readonlyConfig.get(MaxcomputeBaseOptions.ACCESS_ID),
- readonlyConfig.get(MaxcomputeBaseOptions.ACCESS_KEY));
+ account = MaxcomputeUtil.getAccount(readonlyConfig);
}
@Override
@@ -134,7 +130,7 @@ public class MaxComputeCatalog implements Catalog {
@Override
public boolean tableExists(TablePath tablePath) throws CatalogException {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
com.aliyun.odps.Tables tables = odps.tables();
return tables.exists(tablePath.getTableName());
} catch (OdpsException e) {
@@ -157,7 +153,7 @@ public class MaxComputeCatalog implements Catalog {
Table odpsTable;
com.aliyun.odps.TableSchema odpsSchema;
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
odpsTable =
MaxcomputeUtil.parseTable(
odps, tablePath.getDatabaseName(),
tablePath.getTableName());
@@ -204,7 +200,7 @@ public class MaxComputeCatalog implements Catalog {
public void createTable(TablePath tablePath, CatalogTable table, boolean
ignoreIfExists)
throws TableAlreadyExistException, DatabaseNotExistException,
CatalogException {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
SQLTask.run(
odps,
MaxComputeCatalogUtil.getCreateTableStatement(
@@ -222,7 +218,7 @@ public class MaxComputeCatalog implements Catalog {
public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
SQLTask.run(odps,
MaxComputeCatalogUtil.getDropTableQuery(tablePath, ignoreIfNotExists))
.waitForSuccess();
} catch (OdpsException e) {
@@ -246,7 +242,7 @@ public class MaxComputeCatalog implements Catalog {
public void truncateTable(TablePath tablePath, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
Table odpsTable = odps.tables().get(tablePath.getTableName());
if (odpsTable.isPartitioned()
&& StringUtils.isNotEmpty(
@@ -265,7 +261,7 @@ public class MaxComputeCatalog implements Catalog {
public void createPartition(TablePath tablePath, PartitionSpec
partitionSpec) {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
Table odpsTable = odps.tables().get(tablePath.getTableName());
odpsTable.createPartition(partitionSpec, true);
} catch (Exception e) {
@@ -275,7 +271,7 @@ public class MaxComputeCatalog implements Catalog {
public void truncatePartition(TablePath tablePath, PartitionSpec
partitionSpec) {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
Table odpsTable = odps.tables().get(tablePath.getTableName());
odpsTable.deletePartition(partitionSpec, true);
odpsTable.createPartition(partitionSpec, true);
@@ -292,7 +288,7 @@ public class MaxComputeCatalog implements Catalog {
@Override
public void executeSql(TablePath tablePath, String sql) {
try {
- Odps odps = getOdps(tablePath.getDatabaseName());
+ Odps odps = getOdps(tablePath.getDatabaseName(),
tablePath.getSchemaName());
String[] sqls = sql.split(";");
for (String s : sqls) {
if (!s.trim().isEmpty()) {
@@ -324,10 +320,22 @@ public class MaxComputeCatalog implements Catalog {
}
}
+ /**
+ * Creates an ODPS client for the given project. When {@code schemaName}
is non-blank, sets it
+ * as the current schema so that subsequent ODPS API calls resolve tables
within that MaxCompute
+ * Schema namespace.
+ */
private Odps getOdps(String project) {
+ return getOdps(project, null);
+ }
+
+ private Odps getOdps(String project, String schemaName) {
Odps odps = new Odps(account);
odps.setEndpoint(readonlyConfig.get(MaxcomputeBaseOptions.ENDPOINT));
odps.setDefaultProject(project);
+ if (StringUtils.isNotEmpty(schemaName)) {
+ odps.setCurrentSchema(schemaName);
+ }
return odps;
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalogFactory.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalogFactory.java
index b8f6187617..b40c1d4ae0 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalogFactory.java
@@ -44,14 +44,16 @@ public class MaxComputeCatalogFactory implements
CatalogFactory {
public OptionRule optionRule() {
return OptionRule.builder()
.required(
- MaxcomputeBaseOptions.ACCESS_ID,
- MaxcomputeBaseOptions.ACCESS_KEY,
MaxcomputeBaseOptions.ENDPOINT,
MaxcomputeBaseOptions.PROJECT,
MaxcomputeBaseOptions.TABLE_NAME)
.optional(
+ MaxcomputeBaseOptions.ACCESS_ID,
+ MaxcomputeBaseOptions.ACCESS_KEY,
+ MaxcomputeBaseOptions.STS_TOKEN,
MaxcomputeBaseOptions.PARTITION_SPEC,
MaxcomputeBaseOptions.SPLIT_ROW,
+ MaxcomputeBaseOptions.SCHEMA_NAME,
ConnectorCommonOptions.SCHEMA)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/config/MaxcomputeBaseOptions.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/config/MaxcomputeBaseOptions.java
index c581430e89..c445f58503 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/config/MaxcomputeBaseOptions.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/config/MaxcomputeBaseOptions.java
@@ -38,6 +38,11 @@ public class MaxcomputeBaseOptions implements Serializable {
.noDefaultValue()
.withDescription(
"Your Maxcompute accessKey which cloud be access
from Alibaba Cloud");
+ public static final Option<String> STS_TOKEN =
+ Options.key("sts_token")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Your Maxcompute stsToken for temporary
access");
public static final Option<String> ENDPOINT =
Options.key("endpoint")
.stringType()
@@ -56,6 +61,15 @@ public class MaxcomputeBaseOptions implements Serializable {
.noDefaultValue()
.withDescription("Target Maxcompute table name eg: fake");
+ public static final Option<String> SCHEMA_NAME =
+ Options.key("schema_name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The MaxCompute Schema name (namespace between
Project and Table). "
+ + "Only required when the table resides in
a non-default schema. "
+ + "See
https://www.alibabacloud.com/help/en/maxcompute/user-guide/schema-related-operations";);
+
public static final Option<String> PARTITION_SPEC =
Options.key("partition_spec")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/sink/MaxcomputeSinkFactory.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/sink/MaxcomputeSinkFactory.java
index d58f9a4eba..d4467c788b 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/sink/MaxcomputeSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/sink/MaxcomputeSinkFactory.java
@@ -41,13 +41,15 @@ public class MaxcomputeSinkFactory implements
TableSinkFactory {
public OptionRule optionRule() {
return OptionRule.builder()
.required(
- MaxcomputeSinkOptions.ACCESS_ID,
- MaxcomputeSinkOptions.ACCESS_KEY,
MaxcomputeSinkOptions.ENDPOINT,
MaxcomputeSinkOptions.PROJECT,
MaxcomputeSinkOptions.TABLE_NAME)
.optional(
+ MaxcomputeSinkOptions.ACCESS_ID,
+ MaxcomputeSinkOptions.ACCESS_KEY,
+ MaxcomputeSinkOptions.STS_TOKEN,
MaxcomputeSinkOptions.PARTITION_SPEC,
+ MaxcomputeSinkOptions.SCHEMA_NAME,
MaxcomputeSinkOptions.OVERWRITE,
MaxcomputeSinkOptions.SCHEMA_SAVE_MODE,
MaxcomputeSinkOptions.DATA_SAVE_MODE,
@@ -68,6 +70,9 @@ public class MaxcomputeSinkFactory implements
TableSinkFactory {
TableIdentifier.of(
context.getCatalogTable().getCatalogName(),
context.getOptions().get(MaxcomputeSinkOptions.PROJECT),
+ context.getOptions()
+
.getOptional(MaxcomputeSinkOptions.SCHEMA_NAME)
+ .orElse(null),
context.getOptions().get(MaxcomputeSinkOptions.TABLE_NAME)),
context.getCatalogTable()));
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSource.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSource.java
index 84f3370fe6..1a03a23b48 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSource.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSource.java
@@ -69,6 +69,9 @@ public class MaxcomputeSource
TableIdentifier.of(
"maxcompute",
readonlyConfig.get(MaxcomputeSourceOptions.PROJECT),
+ readonlyConfig
+
.getOptional(MaxcomputeSourceOptions.SCHEMA_NAME)
+ .orElse(null),
readonlyConfig.get(MaxcomputeSourceOptions.TABLE_NAME)),
catalogTable);
tables.put(
@@ -90,9 +93,19 @@ public class MaxcomputeSource
.orElse(
readonlyConfig.get(
MaxcomputeSourceOptions.PROJECT));
+ // schema_name: per-table override, falls back to
top-level value
+ String schemaName =
+ subReadonlyConfig
+
.getOptional(MaxcomputeSourceOptions.SCHEMA_NAME)
+ .orElse(
+ readonlyConfig
+ .getOptional(
+
MaxcomputeSourceOptions.SCHEMA_NAME)
+ .orElse(null));
TablePath tablePath =
TablePath.of(
project,
+ schemaName,
subReadonlyConfig.get(MaxcomputeSourceOptions.TABLE_NAME));
String partitionSpec =
subReadonlyConfig
@@ -139,6 +152,9 @@ public class MaxcomputeSource
TablePath tablePath =
TablePath.of(
readonlyConfig.get(MaxcomputeSourceOptions.PROJECT),
+ readonlyConfig
+
.getOptional(MaxcomputeSourceOptions.SCHEMA_NAME)
+ .orElse(null),
readonlyConfig.get(MaxcomputeSourceOptions.TABLE_NAME));
tables.put(
tablePath,
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceFactory.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceFactory.java
index c720fef204..65b1be3839 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceFactory.java
@@ -42,13 +42,14 @@ public class MaxcomputeSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(
+ .required(MaxcomputeSourceOptions.ENDPOINT)
+ .optional(
MaxcomputeSourceOptions.ACCESS_ID,
MaxcomputeSourceOptions.ACCESS_KEY,
- MaxcomputeSourceOptions.ENDPOINT)
- .optional(
+ MaxcomputeSourceOptions.STS_TOKEN,
MaxcomputeSourceOptions.PARTITION_SPEC,
MaxcomputeSourceOptions.SPLIT_ROW,
+ MaxcomputeSourceOptions.SCHEMA_NAME,
ConnectorCommonOptions.SCHEMA,
MaxcomputeSourceOptions.PROJECT,
MaxcomputeSourceOptions.READ_COLUMNS,
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeOutputFormat.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeOutputFormat.java
index a4a3dfd847..f3fee4f965 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeOutputFormat.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeOutputFormat.java
@@ -185,13 +185,14 @@ public class MaxcomputeOutputFormat {
readonlyConfig.get(MaxcomputeSinkOptions.PROJECT),
readonlyConfig.get(MaxcomputeSinkOptions.TABLE_NAME))
.setPartitionSpec(partitionSpec)
+
.setSchemaName(tunnel.getConfig().getOdps().getCurrentSchema())
.build();
-
} else {
upsertSession =
tunnel.buildUpsertSession(
readonlyConfig.get(MaxcomputeSinkOptions.PROJECT),
readonlyConfig.get(MaxcomputeSinkOptions.TABLE_NAME))
+
.setSchemaName(tunnel.getConfig().getOdps().getCurrentSchema())
.build();
}
}
@@ -208,20 +209,16 @@ public class MaxcomputeOutputFormat {
private void initializeInsertSession() throws TunnelException {
TableTunnel tunnel = MaxcomputeUtil.getTableTunnel(readonlyConfig);
+ PartitionSpec partitionSpec = null;
if
(readonlyConfig.getOptional(MaxcomputeSinkOptions.PARTITION_SPEC).isPresent()) {
- PartitionSpec partitionSpec =
+ partitionSpec =
new
PartitionSpec(readonlyConfig.get(MaxcomputeSinkOptions.PARTITION_SPEC));
- uploadSession =
- tunnel.createUploadSession(
- readonlyConfig.get(MaxcomputeSinkOptions.PROJECT),
-
readonlyConfig.get(MaxcomputeSinkOptions.TABLE_NAME),
- partitionSpec);
-
- } else {
- uploadSession =
- tunnel.createUploadSession(
- readonlyConfig.get(MaxcomputeSinkOptions.PROJECT),
-
readonlyConfig.get(MaxcomputeSinkOptions.TABLE_NAME));
}
+ uploadSession =
+ MaxcomputeUtil.buildUploadSession(
+ tunnel,
+ readonlyConfig.get(MaxcomputeSinkOptions.PROJECT),
+ readonlyConfig.get(MaxcomputeSinkOptions.TABLE_NAME),
+ partitionSpec);
}
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtil.java
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtil.java
index 910b2db0bf..94e7b53aa7 100644
---
a/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtil.java
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtil.java
@@ -25,11 +25,14 @@ import
org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import
org.apache.seatunnel.connectors.seatunnel.maxcompute.config.MaxcomputeBaseOptions;
import
org.apache.seatunnel.connectors.seatunnel.maxcompute.exception.MaxcomputeConnectorException;
+import com.aliyun.credentials.provider.DefaultCredentialsProvider;
import com.aliyun.odps.Odps;
import com.aliyun.odps.PartitionSpec;
import com.aliyun.odps.Table;
import com.aliyun.odps.account.Account;
+import com.aliyun.odps.account.AklessAccount;
import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.account.StsAccount;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TunnelException;
import lombok.extern.slf4j.Slf4j;
@@ -50,14 +53,32 @@ public class MaxcomputeUtil {
return tableTunnel;
}
+ public static Account getAccount(ReadonlyConfig readonlyConfig) {
+ String stsToken =
readonlyConfig.getOptional(MaxcomputeBaseOptions.STS_TOKEN).orElse(null);
+ String accessId =
readonlyConfig.getOptional(MaxcomputeBaseOptions.ACCESS_ID).orElse(null);
+ String accessKey =
+
readonlyConfig.getOptional(MaxcomputeBaseOptions.ACCESS_KEY).orElse(null);
+
+ if (StringUtils.isNotEmpty(stsToken)) {
+ if (StringUtils.isEmpty(accessId) ||
StringUtils.isEmpty(accessKey)) {
+ throw new IllegalArgumentException(
+ "accessId and accesskey must be provided when
sts_token is used.");
+ }
+ return new StsAccount(accessId, accessKey, stsToken);
+ } else if (StringUtils.isNotEmpty(accessId) &&
StringUtils.isNotEmpty(accessKey)) {
+ return new AliyunAccount(accessId, accessKey);
+ } else {
+ return new AklessAccount(new DefaultCredentialsProvider());
+ }
+ }
+
public static Odps getOdps(ReadonlyConfig readonlyConfig) {
- Account account =
- new AliyunAccount(
- readonlyConfig.get(MaxcomputeBaseOptions.ACCESS_ID),
- readonlyConfig.get(MaxcomputeBaseOptions.ACCESS_KEY));
+ Account account = getAccount(readonlyConfig);
Odps odps = new Odps(account);
odps.setEndpoint(readonlyConfig.get(MaxcomputeBaseOptions.ENDPOINT));
odps.setDefaultProject(readonlyConfig.get(MaxcomputeBaseOptions.PROJECT));
+ odps.setCurrentSchema(
+
readonlyConfig.getOptional(MaxcomputeBaseOptions.SCHEMA_NAME).orElse(null));
return odps;
}
@@ -140,4 +161,18 @@ public class MaxcomputeUtil {
.setPartitionSpec(partitionSpec)
.build();
}
+
+ /**
+ * Note: Unlike DownloadSession and UpsertSession, we do not need to
explicitly pass the schema
+ * name to the builder here. The underlying
`TableTunnel.createUploadSession` automatically
+ * inherits the schema_name directly from
`tunnel.getConfig().getOdps().getCurrentSchema()`.
+ */
+ public static TableTunnel.UploadSession buildUploadSession(
+ TableTunnel tunnel, String projectName, String tableName,
PartitionSpec partitionSpec)
+ throws TunnelException {
+ if (partitionSpec != null) {
+ return tunnel.createUploadSession(projectName, tableName,
partitionSpec);
+ }
+ return tunnel.createUploadSession(projectName, tableName);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtilTest.java
b/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtilTest.java
new file mode 100644
index 0000000000..e3aa017bdd
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-maxcompute/src/test/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/util/MaxcomputeUtilTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.maxcompute.util;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.aliyun.odps.account.Account;
+import com.aliyun.odps.account.AklessAccount;
+import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.account.StsAccount;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MaxcomputeUtilTest {
+
+ // --- getAccount() tests ---
+
+ /**
+ * When only accessId + accessKey are supplied, the connector must use a
standard AliyunAccount
+ * (long-lived static AK/SK).
+ */
+ @Test
+ void testGetAccountReturnsAliyunAccountWhenAkSkProvided() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("accessId", "my-id");
+ config.put("accesskey", "my-key");
+
+ Account account =
MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertInstanceOf(AliyunAccount.class, account);
+ }
+
+ /**
+ * When accessId + accessKey + sts_token are all supplied, the connector
must use an StsAccount
+ * (temporary credential).
+ */
+ @Test
+ void testGetAccountReturnsStsAccountWhenAllThreeProvided() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("accessId", "my-id");
+ config.put("accesskey", "my-key");
+ config.put("sts_token", "my-sts-token");
+
+ Account account =
MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertInstanceOf(StsAccount.class, account);
+ }
+
+ /**
+ * When none of the access credentials are supplied, the connector must
fall back to the
+ * DefaultCredentialsProvider chain (AklessAccount), enabling passwordless
auth via ECS RAM
+ * roles, environment variables, etc.
+ */
+ @Test
+ void testGetAccountReturnsAklessAccountWhenNothingProvided() {
+ Account account = MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(new
HashMap<>()));
+
+ Assertions.assertInstanceOf(AklessAccount.class, account);
+ }
+
+ /**
+ * Providing sts_token WITHOUT accessId/accessKey must fail fast with an
explicit
+ * IllegalArgumentException instead of letting a NullPointerException
surface at runtime inside
+ * the ODPS SDK.
+ */
+ @Test
+ void testGetAccountThrowsWhenStsTokenProvidedWithoutAkSk() {
+ Map<String, Object> configMissingBoth = new HashMap<>();
+ configMissingBoth.put("sts_token", "my-sts-token");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(configMissingBoth)),
+ "Expected IllegalArgumentException when sts_token is set but
accessId/accesskey are missing");
+ }
+
+ /** Providing sts_token with only accessId (no accessKey) must also fail
fast. */
+ @Test
+ void testGetAccountThrowsWhenStsTokenProvidedWithoutAccessKey() {
+ Map<String, Object> configMissingKey = new HashMap<>();
+ configMissingKey.put("accessId", "my-id");
+ configMissingKey.put("sts_token", "my-sts-token");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(configMissingKey)));
+ }
+
+ /** Providing sts_token with only accessKey (no accessId) must also fail
fast. */
+ @Test
+ void testGetAccountThrowsWhenStsTokenProvidedWithoutAccessId() {
+ Map<String, Object> configMissingId = new HashMap<>();
+ configMissingId.put("accesskey", "my-key");
+ configMissingId.put("sts_token", "my-sts-token");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
MaxcomputeUtil.getAccount(ReadonlyConfig.fromMap(configMissingId)));
+ }
+
+ // --- getOdps() / schema_name propagation test ---
+
+ /**
+ * When schema_name is provided, it must be propagated to
Odps.getCurrentSchema(). This verifies
+ * that non-default MaxCompute schemas are correctly set on the Odps
client, enabling
+ * schema-aware table routing for source and sink operations.
+ */
+ @Test
+ void testGetOdpsSetsCurrentSchemaWhenSchemaNameProvided() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("accessId", "my-id");
+ config.put("accesskey", "my-key");
+ config.put("endpoint", "http://service.odps.aliyun.com/api";);
+ config.put("project", "my_project");
+ config.put("schema_name", "my_schema");
+
+ com.aliyun.odps.Odps odps =
MaxcomputeUtil.getOdps(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertEquals("my_schema", odps.getCurrentSchema());
+ }
+
+ /**
+ * When schema_name is absent, Odps.getCurrentSchema() must remain null,
letting the SDK fall
+ * back to the project's default schema transparently.
+ */
+ @Test
+ void testGetOdpsLeavesCurrentSchemaEmptyWhenSchemaNameAbsent() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("accessId", "my-id");
+ config.put("accesskey", "my-key");
+ config.put("endpoint", "http://service.odps.aliyun.com/api";);
+ config.put("project", "my_project");
+
+ com.aliyun.odps.Odps odps =
MaxcomputeUtil.getOdps(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertNull(odps.getCurrentSchema());
+ }
+}
