[Fix][Connector-V2] Fix maxcompute read with partition spec (#8896)

Author: Hisoka-X

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/catalog/MaxComputeCatalog.java

@@ -158,14 +158,12 @@ public CatalogTable getTable(TablePath tablePath, List<String> fieldNames)
         }
         Table odpsTable;
         com.aliyun.odps.TableSchema odpsSchema;
-        boolean isPartitioned;
         try {
             Odps odps = getOdps(tablePath.getDatabaseName());
             odpsTable =
                     MaxcomputeUtil.parseTable(
                             odps, tablePath.getDatabaseName(), tablePath.getTableName());
             odpsSchema = odpsTable.getSchema();
-            isPartitioned = odpsTable.isPartitioned();
         } catch (Exception ex) {
             throw new CatalogException(catalogName, ex);
         }
@@ -193,31 +191,6 @@ public CatalogTable getTable(TablePath tablePath, List<String> fieldNames)
                                     .build();
                     return MaxComputeTypeConverter.INSTANCE.convert(typeDefine);
                 });
-        if (isPartitioned) {
-            buildColumnsWithErrorCheck(
-                    tablePath,
-                    builder,
-                    odpsSchema.getPartitionColumns().stream()
-                            .filter(
-                                    column ->
-                                            fieldNames == null
-                                                    || fieldNames.isEmpty()
-                                                    || fieldNames.contains(column.getName()))
-                            .iterator(),
-                    (column) -> {
-                        BasicTypeDefine<TypeInfo> typeDefine =
-                                BasicTypeDefine.<TypeInfo>builder()
-                                        .name(column.getName())
-                                        .nativeType(column.getTypeInfo())
-                                        .columnType(column.getTypeInfo().getTypeName())
-                                        .dataType(column.getTypeInfo().getTypeName())
-                                        .nullable(column.isNullable())
-                                        .comment(column.getComment())
-                                        .build();
-                        partitionKeys.add(column.getName());
-                        return MaxComputeTypeConverter.INSTANCE.convert(typeDefine);
-                    });
-        }
         TableSchema tableSchema = builder.build();
         TableIdentifier tableIdentifier = getTableIdentifier(tablePath);
         return CatalogTable.of(

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSource.java

@@ -95,6 +95,11 @@ private Map<TablePath, SourceTableInfo> getSourceTableInfos(ReadonlyConfig reado
                                         .orElse(readonlyConfig.get(PROJECT));
                         TablePath tablePath =
                                 TablePath.of(project, subReadonlyConfig.get(TABLE_NAME));
+                        String partitionSpec =
+                                subReadonlyConfig
+                                        .getOptional(PARTITION_SPEC)
+                                        .orElse(readonlyConfig.get(PARTITION_SPEC));
+
                         if (subReadonlyConfig
                                 .getOptional(ConnectorCommonOptions.SCHEMA)
                                 .isPresent()) {
@@ -108,7 +113,7 @@ private Map<TablePath, SourceTableInfo> getSourceTableInfos(ReadonlyConfig reado
                                     catalogTable.getTablePath(),
                                     new SourceTableInfo(
                                             catalogTable,
-                                            subReadonlyConfig.get(PARTITION_SPEC),
+                                            partitionSpec,
                                             subReadonlyConfig.get(SPLIT_ROW)));
                         } else {
                             Integer splitRow =
@@ -120,7 +125,7 @@ private Map<TablePath, SourceTableInfo> getSourceTableInfos(ReadonlyConfig reado
                                     new SourceTableInfo(
                                             catalog.getTable(
                                                     tablePath, subReadonlyConfig.get(READ_COLUMNS)),
-                                            subReadonlyConfig.get(PARTITION_SPEC),
+                                            partitionSpec,
                                             splitRow));
                         }
                     }