[fix](fe) Improve MaxCompute catalog validation

hubgeter · hubgeter · commit f89dad9bfaf6 · 2026-06-04T18:14:47.000+08:00
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
@@ -369,6 +369,10 @@ under the License.
             <artifactId>odps-sdk-table-api</artifactId>
             <version>${maxcompute.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.aliyun</groupId>
+            <artifactId>maxcompute20220104</artifactId>
+        </dependency>
         <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java
@@ -71,6 +71,10 @@ public void updateMCCommitData(List<TMCCommitData> commitDataList) {
 
     public void beginInsert(ExternalTable dorisTable, Optional<InsertCommandContext> ctx) throws UserException {
         this.table = (MaxComputeExternalTable) dorisTable;
+        if (table.isUnsupportedOdpsTable()) {
+            throw new UserException("Writing MaxCompute external table or logical view is not supported: "
+                    + table.getDbName() + "." + table.getName());
+        }
 
         try {
             TableIdentifier tableId = catalog.getOdpsTableIdentifier(table.getDbName(), table.getName());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java
@@ -27,7 +27,11 @@
 import org.apache.doris.datasource.SessionContext;
 import org.apache.doris.transaction.TransactionManagerFactory;
 
+import com.aliyun.maxcompute20220104.models.GetProjectRequest;
+import com.aliyun.maxcompute20220104.models.GetProjectResponse;
+import com.aliyun.maxcompute20220104.models.GetProjectResponseBody.GetProjectResponseBodyData;
 import com.aliyun.odps.Odps;
+import com.aliyun.odps.OdpsException;
 import com.aliyun.odps.Partition;
 import com.aliyun.odps.account.AccountFormat;
 import com.aliyun.odps.table.TableIdentifier;
@@ -52,6 +56,13 @@ public class MaxComputeExternalCatalog extends ExternalCatalog {
 
     // you can ref : https://help.aliyun.com/zh/maxcompute/user-guide/endpoints
     private static final String endpointTemplate = "http://service.{}.maxcompute.aliyun-inc.com/api";
+    private static final int VALIDATION_CONNECT_TIMEOUT_SECONDS = 10;
+    private static final int VALIDATION_READ_TIMEOUT_SECONDS = 10;
+    private static final int VALIDATION_RETRY_TIMES = 1;
+    private static final String OPEN_API_AUTH_TYPE_ACCESS_KEY = "access_key";
+    private static final String OPEN_API_AUTH_TYPE_RAM_ROLE_ARN = "ram_role_arn";
+    private static final String OPEN_API_AUTH_TYPE_ECS_RAM_ROLE = "ecs_ram_role";
+    private static final String OPEN_API_ROLE_SESSION_NAME = "doris-maxcompute-catalog";
 
     private Map<String, String> props;
     private Odps odps;
@@ -206,6 +217,7 @@ protected void initLocalObjectsImpl() {
         odps = MCUtils.createMcClient(props);
         odps.setDefaultProject(defaultProject);
         odps.setEndpoint(endpoint);
+        configureOdpsRestClient();
 
         String accountFormatProp = props.getOrDefault(MCProperties.ACCOUNT_FORMAT, MCProperties.DEFAULT_ACCOUNT_FORMAT);
         if (accountFormatProp.equals(MCProperties.ACCOUNT_FORMAT_NAME)) {
@@ -227,12 +239,136 @@ protected void initLocalObjectsImpl() {
         boolean enableNamespaceSchema = Boolean.parseBoolean(
                 props.getOrDefault(MCProperties.ENABLE_NAMESPACE_SCHEMA, MCProperties.DEFAULT_ENABLE_NAMESPACE_SCHEMA));
         mcStructureHelper = McStructureHelper.getHelper(enableNamespaceSchema, defaultProject);
+        validateMaxComputeConnection(enableNamespaceSchema);
 
         initPreExecutionAuthenticator();
         metadataOps = new MaxComputeMetadataOps(this, odps);
         transactionManager = TransactionManagerFactory.createMCTransactionManager(this);
     }
 
+    private void configureOdpsRestClient() {
+        odps.getRestClient().setConnectTimeout(connectTimeout);
+        odps.getRestClient().setReadTimeout(readTimeout);
+        odps.getRestClient().setRetryTimes(retryTimes);
+    }
+
+    protected void validateMaxComputeConnection(boolean enableNamespaceSchema) {
+        int originConnectTimeout = odps.getRestClient().getConnectTimeout();
+        int originReadTimeout = odps.getRestClient().getReadTimeout();
+        int originRetryTimes = odps.getRestClient().getRetryTimes();
+        configureOdpsRestClientForValidation();
+        try {
+            if (enableNamespaceSchema) {
+                validateMaxComputeProjectAndNamespaceSchema();
+            } else {
+                validateMaxComputeProject();
+            }
+        } finally {
+            restoreOdpsRestClient(originConnectTimeout, originReadTimeout, originRetryTimes);
+        }
+    }
+
+    private void configureOdpsRestClientForValidation() {
+        odps.getRestClient().setConnectTimeout(Math.min(connectTimeout, VALIDATION_CONNECT_TIMEOUT_SECONDS));
+        odps.getRestClient().setReadTimeout(Math.min(readTimeout, VALIDATION_READ_TIMEOUT_SECONDS));
+        odps.getRestClient().setRetryTimes(Math.min(retryTimes, VALIDATION_RETRY_TIMES));
+    }
+
+    private void restoreOdpsRestClient(int originConnectTimeout, int originReadTimeout, int originRetryTimes) {
+        odps.getRestClient().setConnectTimeout(originConnectTimeout);
+        odps.getRestClient().setReadTimeout(originReadTimeout);
+        odps.getRestClient().setRetryTimes(originRetryTimes);
+    }
+
+    private void validateMaxComputeProject() {
+        try {
+            if (!maxComputeProjectExists(defaultProject)) {
+                throw new RuntimeException("MaxCompute project '" + defaultProject
+                        + "' does not exist or is not accessible");
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to validate MaxCompute project '" + defaultProject
+                    + "'. Check " + MCProperties.PROJECT + ", " + MCProperties.ENDPOINT
+                    + " and credentials. Cause: " + e.getMessage());
+        }
+    }
+
+    private void validateMaxComputeProjectAndNamespaceSchema() {
+        try {
+            if (!isMaxComputeProjectThreeTierModel(defaultProject)) {
+                throw new RuntimeException("MaxCompute project '" + defaultProject
+                        + "' does not support namespace schema");
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to validate MaxCompute project '" + defaultProject
+                    + "' with GetProject. Check " + MCProperties.PROJECT + ", " + MCProperties.ENDPOINT
+                    + ", credentials and namespace schema support. Cause: " + e.getMessage());
+        }
+    }
+
+    protected boolean maxComputeProjectExists(String projectName) throws OdpsException {
+        return odps.projects().exists(projectName);
+    }
+
+    protected boolean isMaxComputeProjectThreeTierModel(String projectName) throws Exception {
+        GetProjectRequest request = new GetProjectRequest()
+                .setVerbose(false)
+                .setWithQuotaProductType(false)
+                .setWithStorageTierInfo(false);
+        GetProjectResponse response = createMaxComputeOpenApiClient().getProject(projectName, request);
+        GetProjectResponseBodyData data = response.getBody() == null ? null : response.getBody().getData();
+        if (data == null || data.getThreeTierModel() == null) {
+            throw new RuntimeException("MaxCompute GetProject did not return threeTierModel");
+        }
+        return Boolean.TRUE.equals(data.getThreeTierModel());
+    }
+
+    private com.aliyun.maxcompute20220104.Client createMaxComputeOpenApiClient() throws Exception {
+        com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
+                .setCredential(createOpenApiCredentialClient())
+                .setRegionId(getMaxComputeRegion())
+                .setConnectTimeout(Math.min(connectTimeout, VALIDATION_CONNECT_TIMEOUT_SECONDS) * 1000)
+                .setReadTimeout(Math.min(readTimeout, VALIDATION_READ_TIMEOUT_SECONDS) * 1000);
+        return new com.aliyun.maxcompute20220104.Client(config);
+    }
+
+    private com.aliyun.credentials.Client createOpenApiCredentialClient() {
+        String authType = props.getOrDefault(MCProperties.AUTH_TYPE, MCProperties.DEFAULT_AUTH_TYPE);
+        com.aliyun.credentials.models.Config credentialConfig = new com.aliyun.credentials.models.Config();
+        if (authType.equalsIgnoreCase(MCProperties.AUTH_TYPE_AK_SK)) {
+            credentialConfig.setType(OPEN_API_AUTH_TYPE_ACCESS_KEY)
+                    .setAccessKeyId(props.get(MCProperties.ACCESS_KEY))
+                    .setAccessKeySecret(props.get(MCProperties.SECRET_KEY));
+        } else if (authType.equalsIgnoreCase(MCProperties.AUTH_TYPE_RAM_ROLE_ARN)) {
+            credentialConfig.setType(OPEN_API_AUTH_TYPE_RAM_ROLE_ARN)
+                    .setAccessKeyId(props.get(MCProperties.ACCESS_KEY))
+                    .setAccessKeySecret(props.get(MCProperties.SECRET_KEY))
+                    .setRoleArn(props.get(MCProperties.RAM_ROLE_ARN))
+                    .setRoleSessionName(OPEN_API_ROLE_SESSION_NAME);
+        } else if (authType.equalsIgnoreCase(MCProperties.AUTH_TYPE_ECS_RAM_ROLE)) {
+            credentialConfig.setType(OPEN_API_AUTH_TYPE_ECS_RAM_ROLE)
+                    .setRoleName(props.get(MCProperties.ECS_RAM_ROLE));
+        } else {
+            throw new RuntimeException("Unsupported MaxCompute auth type: " + authType);
+        }
+        return new com.aliyun.credentials.Client(credentialConfig);
+    }
+
+    protected String getMaxComputeRegion() {
+        if (props.containsKey(MCProperties.REGION)) {
+            return normalizeMaxComputeRegion(props.get(MCProperties.REGION));
+        }
+        String[] endpointSplit = endpoint.split("\\.");
+        if (endpointSplit.length >= 2) {
+            return normalizeMaxComputeRegion(endpointSplit[1]);
+        }
+        throw new RuntimeException("Can not infer MaxCompute region from endpoint: " + endpoint);
+    }
+
+    private String normalizeMaxComputeRegion(String region) {
+        return region.replace("oss-", "").replace("-vpc", "").replace("-intranet", "");
+    }
+
     public Odps getClient() {
         makeSureInitialized();
         return odps;
@@ -401,7 +537,7 @@ public void checkProperties() throws DdlException {
                         MCProperties.DEFAULT_SPLIT_BYTE_SIZE));
 
                 if (splitByteSize < 10485760L) {
-                    throw new DdlException(MCProperties.SPLIT_ROW_COUNT + " must be greater than or equal to 10485760");
+                    throw new DdlException(MCProperties.SPLIT_BYTE_SIZE + " must be greater than or equal to 10485760");
                 }
 
             } else if (splitStrategy.equals(MCProperties.SPLIT_BY_ROW_COUNT_STRATEGY)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalTable.java
@@ -328,6 +328,11 @@ public Table getOdpsTable() {
                 .orElse(null);
     }
 
+    public boolean isUnsupportedOdpsTable() {
+        Table odpsTable = getOdpsTable();
+        return odpsTable.isExternalTable() || odpsTable.isVirtualView();
+    }
+
     @Override
     public boolean isPartitionedTable() {
         makeSureInitialized();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/source/MaxComputeScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/source/MaxComputeScanNode.java
@@ -707,6 +707,11 @@ public List<Split> getSplits(int numBackends) throws UserException {
         long getOdpsTableTime = System.currentTimeMillis();
         LOG.info("MaxComputeScanNode getSplits: getOdpsTable cost {} ms", getOdpsTableTime - startTime);
 
+        if (odpsTable.isExternalTable() || odpsTable.isVirtualView()) {
+            throw new UserException("Reading MaxCompute external table or logical view is not supported: "
+                    + table.getDbName() + "." + table.getName());
+        }
+
         if (desc.getSlots().isEmpty() || odpsTable.getFileNum() <= 0) {
             return result;
         }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/MCTransactionTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/MCTransactionTest.java
@@ -0,0 +1,54 @@
+// 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.doris.datasource.maxcompute;
+
+import org.apache.doris.common.UserException;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Optional;
+
+public class MCTransactionTest {
+    @Test
+    public void testBeginInsertRejectsOdpsExternalTable() {
+        assertBeginInsertRejectsUnsupportedOdpsTable("mc_external_table");
+    }
+
+    @Test
+    public void testBeginInsertRejectsOdpsLogicalView() {
+        assertBeginInsertRejectsUnsupportedOdpsTable("mc_logical_view");
+    }
+
+    private void assertBeginInsertRejectsUnsupportedOdpsTable(String tableName) {
+        MaxComputeExternalCatalog catalog = Mockito.mock(MaxComputeExternalCatalog.class);
+        MaxComputeExternalTable table = Mockito.mock(MaxComputeExternalTable.class);
+        Mockito.when(table.isUnsupportedOdpsTable()).thenReturn(true);
+        Mockito.when(table.getDbName()).thenReturn("default");
+        Mockito.when(table.getName()).thenReturn(tableName);
+
+        MCTransaction transaction = new MCTransaction(catalog);
+
+        UserException exception = Assert.assertThrows(UserException.class,
+                () -> transaction.beginInsert(table, Optional.empty()));
+        Assert.assertTrue(exception.getMessage().contains(
+                "Writing MaxCompute external table or logical view is not supported: default." + tableName));
+        Mockito.verify(catalog, Mockito.never()).getOdpsTableIdentifier(Mockito.anyString(), Mockito.anyString());
+    }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalogTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalogTest.java
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/source/MaxComputeScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/maxcompute/source/MaxComputeScanNodeTest.java
diff --git a/fe/pom.xml b/fe/pom.xml
