prepare 0.50.2-public

Author: dingxin-tech

CHANGELOG.md

@@ -1,4 +1,27 @@
 # Changelog
+## [0.50.2-public] - 2024-10-23
+### Features
+- **SQLExecutor** Enhanced MCQA 2.0 functionality:
+    - `isActive` will return false, indicating that there are no active Sessions in MCQA 2.0 mode.
+    - Added a `cancel` method to terminate ongoing jobs.
+    - `getExecutionLog` now returns a deep copy of the current log and clears the current log,
+      preventing duplicates.
+    - New `quota` method in `SQLExecutorBuilder` allows reusing already loaded `Quota`, reducing
+      load times.
+    - New `regionId` method in `SQLExecutorBuilder` allows specifying the region where the quota is
+      located.
+- **Quotas** Added `getWlmQuota` method with `regionId` parameter to fetch quota for a specified regionId.
+- **Quota** Introduced `setMcqaConnHeader` method to allow users to override quota using a custom
+    McqaConnHeader, supporting MCQA 2.0.
+- **Instances** Added `get` method applicable for MCQA 2.0 jobs, requiring additional parameters for QuotaName
+    and RegionId.
+- **Instance** Further adapted for MCQA 2.0 jobs.
+- **TableSchema** `basicallyEquals` method will no longer strictly check for identical Class types.
+### Optimization
+- **SQLExecutor** The `run` method's hints will now be deep-copied, preserving the user-provided Map and
+    supporting immutable types (e.g., `ImmutableMap`).
+### Fixes
+- **Stream** Fixed potential SQL syntax errors in the `create` method.
 
 ## [0.50.1-public] - 2024-10-11
 

CHANGELOG_CN.md

@@ -1,5 +1,23 @@
 # 更新日志
 
+## [0.50.2-public] - 2024-10-23
+### 功能
+- **SQLExecutor** 增强 MCQA 2.0 功能：
+  - `isActive` 将返回 false，指示在 MCQA 2.0 模式下没有活跃的 Session。
+  - 新增 `cancel` 方法，用于中止正在执行的作业。
+  - `getExecutionLog` 现在返回当前日志的深拷贝并清空当前日志，避免重复获取。
+  - 在 `SQLExecutorBuilder` 新增 `quota` 方法，支持复用已加载的 `Quota`，减少加载时间。
+  - 在 `SQLExecutorBuilder` 新增 `regionId` 方法，允许指定 quota 所在的 region。
+- **Quotas** 新增带 `regionId` 参数的 `getWlmQuota` 方法，用于获取指定 regionId 的 quota。
+- **Quota** 新增 `setMcqaConnHeader` 方法，支持用户通过自定义的 McqaConnHeader 重载 quota，以适配 MCQA 2.0。
+- **Instances** 新增适用于 MCQA 2.0 的 `get` 方法，需额外传入 MCQA 2.0 的 QuotaName 和 RegionId。
+- **Instance** 进一步适配 MCQA 2.0 作业。
+- **TableSchema** `basicallyEquals` 方法将不再严格检查两个类的 Class 类型一致性。
+### 优化
+- **SQLExecutor** `run` 方法中的 hints 现在会进行深拷贝，保护用户传入的 Map，支持不可变类型（如 `ImmutableMap`）。
+### 修复
+- **Stream** 修复 `create` 方法中的潜在 SQL 语法错误。
+
 ## [0.50.1-public] - 2024-10-11
 
 ### 修复

odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/TableSchema.java

@@ -253,7 +253,7 @@ public boolean basicallyEquals(Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof TableSchema)) {
       return false;
     }
     TableSchema that = (TableSchema) o;

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Instance.java

@@ -130,6 +130,7 @@ public class SetInformationResult {
   private String project;
   private Map<String, TaskResult> results;
   private boolean isSync = false;
+  private boolean isMcqaV2 = false;
 
   private TaskStatusModel model;
 
@@ -157,6 +158,10 @@ public class SetInformationResult {
 
     this.odps = odps;
     this.client = odps.getRestClient();
+
+    if (getId().endsWith("_mcqa")) {
+      setMcqaV2(true);
+    }
   }
 
   @Root(name = "Instance", strict = false)
@@ -297,7 +302,8 @@ private void reload(boolean isBlock) throws OdpsException {
       params.put("instancestatus", null);
     }
 
-    Response resp = client.request(getResource(), "GET", params, userDefinedHeaders, null);
+    String resource = isMcqaV2 ? "/mcqa" + getResource() : getResource();
+    Response resp = client.request(resource, "GET", params, userDefinedHeaders, null);
     model.owner = resp.getHeaders().get(Headers.ODPS_OWNER);
     String startTimeStr = resp.getHeaders().get(Headers.ODPS_START_TIME);
     String endTimeStr = resp.getHeaders().get(Headers.ODPS_END_TIME);
@@ -670,7 +676,7 @@ public TaskSummary getTaskSummary(String taskName) throws OdpsException {
     params.put("instancesummary", null);
     params.put("taskname", taskName);
     Response result = client.request(getResource(), "GET", params, null, null);
-
+    System.out.println(new String(result.getBody()));
     TaskSummary summary = null;
     try {
       Gson gson = GsonObjectBuilder.get();
@@ -855,6 +861,9 @@ public Map<String, TaskStatus> getTaskStatus() throws OdpsException {
   }
 
   private boolean hasTaskStatus(TaskStatusModel model) {
+    if (model.tasks == null || model.tasks.isEmpty()) {
+      return false;
+    }
     for (InstanceTaskModel taskModel : model.tasks) {
       TaskStatus status = new TaskStatus(taskModel);
       if (status.model.status == null) {
@@ -1391,6 +1400,9 @@ public String getLog(String workerId, LogType logType, int size) throws OdpsExce
 
   /* Un-document */
   public String getTaskDetailJson(String taskName) throws OdpsException {
+    if (isMcqaV2) {
+      return getTaskDetailJson2(taskName);
+    }
     Map<String, String> params = new HashMap<String, String>();
     params.put("instancedetail", null);
     params.put("taskname", taskName);
@@ -1422,7 +1434,8 @@ public String getTaskDetailJson2(String taskName) throws OdpsException {
     Map<String, String> params = new HashMap<String, String>();
     params.put("detail", null);
     params.put("taskname", taskName);
-    Response result = client.request(getResource(), "GET", params, null, null);
+    String resource = isMcqaV2 ? "/mcqa" + getResource() : getResource();
+    Response result = client.request(resource, "GET", params, null, null);
     return new String(result.getBody());
   }
 
@@ -1723,7 +1736,7 @@ public Status getStatus() {
    * @throws OdpsException
    */
   public InstanceQueueingInfo getQueueingInfo() throws OdpsException {
-    Map<String, String> params = new HashMap<String, String>();
+    Map<String, String> params = new HashMap<>();
     params.put("cached", null);
 
     Response resp = client.request(getResource(), "GET", params, null, null);
@@ -1744,6 +1757,12 @@ public void addUserDefinedHeaders(Map<String, String> headers) {
     userDefinedHeaders.putAll(headers);
   }
 
+  public void setMcqaV2(boolean mcqaV2) {
+    isMcqaV2 = mcqaV2;
+    client = odps.clone().getRestClient();
+    client.setPrefix("");
+  }
+
   private Map<String, String> getCommonHeaders() {
     Map<String, String> headers = new HashMap<>();
     headers.put(Headers.CONTENT_TYPE, "application/xml");

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Instances.java

@@ -197,6 +197,35 @@ public Instance get(String projectName, String id) {
     return new Instance(projectName, model, null, odps);
   }
 
+  /**
+   * 获取指定Instance（适用于 MCQA2.0）
+   *
+   * @param projectName
+   *     {@link Instance}所在的{@link Project}名称
+   * @param id
+   *     Instance ID
+   * @param quotaName
+   *     指定的交互式quota
+   * @param regionId
+   *     指定的region，如果为null，则使用project的region
+   * @return {@link Instance}对象
+   * @throws OdpsException
+   */
+  public Instance get(String projectName, String id, String quotaName, String regionId) throws OdpsException {
+    Instance instance = get(projectName, id);
+    if (id.endsWith("_mcqa") && StringUtils.isNullOrEmpty(quotaName)) {
+      throw new IllegalArgumentException("quotaName cannot be null when get MCQA 2.0 instance");
+    }
+    if (!StringUtils.isNullOrEmpty(quotaName)) {
+      Quota quota = odps.quotas().getWlmQuota(projectName, quotaName, regionId);
+      if (quota.isInteractiveQuota()) {
+        String mcqaConnHeader = quota.getMcqaConnHeader();
+        instance.addUserDefinedHeaders(ImmutableMap.of(Headers.ODPS_MCQA_CONN, mcqaConnHeader));
+      }
+    }
+    return instance;
+  }
+
 
   /**
    * 判断指定 Instance 是否存在

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Quota.java

@@ -612,6 +612,18 @@ public String getMcqaConnHeader() {
     return mcqaConnHeader;
   }
 
+  public void setMcqaConnHeader(String mcqaConnHeader) {
+    if (StringUtils.isNullOrEmpty(mcqaConnHeader)) {
+      throw new IllegalArgumentException("McqaConnHeader cannot be null or empty.");
+    }
+    setLoaded(true);
+    if (model == null) {
+      model = new QuotaModel();
+    }
+    this.model.resourceSystemType = ResourceSystemType.FUXI_VW.name();
+    this.mcqaConnHeader = mcqaConnHeader;
+  }
+
   public Map<String, String> getProperties() {
     lazyLoad();
     if (model.properties != null) {

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Quotas.java

@@ -92,14 +92,18 @@ public Quota get(String regionId, String name) {
   }
 
   public Quota getWlmQuota(String project, String name) throws OdpsException {
+    return getWlmQuota(project, name, null);
+  }
+
+  public Quota getWlmQuota(String project, String name, String regionId) throws OdpsException {
     if (StringUtils.isNullOrEmpty(name)) {
       throw new IllegalArgumentException("Argument 'name' cannot be null or empty");
     }
     if (StringUtils.isNullOrEmpty(project)) {
       throw new IllegalArgumentException("Argument 'project' cannot be null or empty");
     }
     String tenantId = odps.projects().get(project).getTenantId();
-    return new Quota(odps, null, name, tenantId);
+    return new Quota(odps, regionId, name, tenantId);
   }
 
   public Quota get(String name) {

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Streams.java

@@ -130,7 +130,7 @@ public void create(StreamIdentifier identifier, TableIdentifier refTable, boolea
     StringBuilder sql = new StringBuilder();
     sql.append("CREATE STREAM ");
     if (ifNotExists) {
-      sql.append("IF NOTEXISTS ");
+      sql.append("IF NOT EXISTS ");
     }
     sql.append(identifier).append(" ON TABLE ");
     sql.append(refTable);
@@ -140,7 +140,7 @@ public void create(StreamIdentifier identifier, TableIdentifier refTable, boolea
     if (version != null) {
       sql.append(" VERSION AS OF ").append(version);
     }
-    sql.append(" STMPROPERTIES(\"read_mode\"=\"").append(readMode.name().toLowerCase())
+    sql.append(" STRMPROPERTIES(\"read_mode\"=\"").append(readMode.name().toLowerCase())
         .append("\")");
     if (StringUtils.isNotBlank(comment)) {
       sql.append(" COMMENT ").append(comment);

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/SQLExecutorBuilder.java

@@ -6,6 +6,7 @@
 import com.aliyun.odps.Instance;
 import com.aliyun.odps.Odps;
 import com.aliyun.odps.OdpsException;
+import com.aliyun.odps.Quota;
 import com.aliyun.odps.table.utils.Preconditions;
 
 /**
@@ -41,6 +42,8 @@ public class SQLExecutorBuilder {
   private boolean sessionSupportNonSelect = false;
   private boolean useMcqaV2 = false;
   private Integer offlineJobPriority = null;
+  private String regionId = null;
+  private Quota quota = null;
 
   public static SQLExecutorBuilder builder() {
     return new SQLExecutorBuilder();
@@ -90,6 +93,12 @@ public SQLExecutorBuilder tunnelEndpoint(String tunnelEndpoint) {
     return this;
   }
 
+  public SQLExecutorBuilder quota(Quota quota) {
+    this.quota = quota;
+    this.quotaName = quota.getNickname();
+    return this;
+  }
+
   public SQLExecutorBuilder quotaName(String quotaName) {
     this.quotaName = quotaName;
     return this;
@@ -175,6 +184,11 @@ public SQLExecutorBuilder offlineJobPriority(Integer offlineJobPriority) {
     return this;
   }
 
+  public SQLExecutorBuilder regionId(String regionId) {
+    this.regionId = regionId;
+    return this;
+  }
+
   public ExecuteMode getExecuteMode() {
     return executeMode;
   }
@@ -211,6 +225,10 @@ public String getQuotaName() {
     return quotaName;
   }
 
+  public Quota getQuota() {
+    return quota;
+  }
+
   public SQLExecutorPool getPool() {
     return pool;
   }
@@ -262,4 +280,8 @@ public boolean isUseMcqaV2() {
   public Integer getOfflineJobPriority() {
     return offlineJobPriority;
   }
+
+  public String getRegionId() {
+    return regionId;
+  }
 }

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/SQLExecutorImpl.java

@@ -779,6 +779,8 @@ public ResultSet getResultSet(Long offset, Long countLimit, Long sizeLimit, bool
   public void run(String sql, Map<String, String> hint) throws OdpsException {
     if (hint == null) {
       hint = new HashMap<>();
+    } else {
+      hint = new HashMap<>(hint);
     }
     queryInfo = new QueryInfo(sql, hint, executeMode);
     queryInfo.setCommandInfo(new CommandInfo(sql, hint));