prepare for 0.50.4

dingxin-tech · dingxin-tech · commit 0542ad3dc2dc · 2024-10-29T17:09:16.000+08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,19 @@
 # Changelog
+## [0.50.4-public] - 2024-10-29
+
+### Features
+- **PartitionSpec** Added a new constructor `(String, boolean)` that uses a boolean parameter to specify whether to trim partition values. This caters to scenarios (such as using char type as a partition field) where users may not want to trim partition values.
+
+### Changes
+- **Instance** The OdpsException thrown when calling the stop method will no longer be wrapped a second time.
+
+### Fixes
+- **SQLExecutor**
+  - Fixed an issue in MCQA 1.0 mode where the user-specified `fallbackPolicy.isFallback4AttachError` did not take effect correctly.
+  - Fixed an issue in MCQA 2.0 mode where the `cancel` method threw an exception when the job failed.
+  - Fixed an issue in MCQA 2.0 mode where using instanceTunnel to fetch results resulted in an error when the isSelect check was incorrect.
+- **Table** Fixed an issue with the `getPartitionSpecs` method that trimmed partition values, causing the retrieval of non-existing partitions.
+
 ## [0.50.3-public] - 2024-10-23
 ### Features
 - **SQLExecutor** In MCQA 1.0 mode, it is allowed to add custom fallback policies, add subclass `FallbackPolicy.UserDefinedFallbackPolicy`.
diff --git a/CHANGELOG_CN.md b/CHANGELOG_CN.md
@@ -1,4 +1,20 @@
 # 更新日志
+## [0.50.4-public] - 2024-10-29
+
+### 功能
+- **PartitionSpec** 新增`(String, boolean)`构造方法，通过布尔参数指定是否对分区值进行trim操作，以满足某些场景（如使用char类型作为分区字段）用户不希望trim的需求。
+
+### 变更
+- **Instance** 在调用stop方法时，抛出的OdpsException将不再被二次包装。
+
+### 修复
+- **SQLExecutor**
+  - 修复了在MCQA 1.0模式下，用户指定`fallbackPolicy.isFallback4AttachError`时未正确生效的问题。
+  - 修复了在MCQA 2.0模式下，作业失败时`cancel`方法抛出异常的问题。
+  - 修复了在MCQA 2.0模式下，当isSelect判断错误时，通过instanceTunnel取结果报错的问题。
+- **Table** 修复了`getPartitionSpecs`方法会trim分区值，导致无法获取存在的分区的问题。
+
+
 ## [0.50.3-public] - 2024-10-23
 ### 功能
 - **SQLExecutor** 在 MCQA 1.0 模式下，允许增加自定义回退策略，新增类`FallbackPolicy.UserDefinedFallbackPolicy`。
diff --git a/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/PartitionSpec.java b/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/PartitionSpec.java
@@ -49,6 +49,10 @@ public PartitionSpec() {
    * 分区定义字符串，分区列之间可以用逗号 (",") 或斜线 ("/") 分隔. 例如: "pt1=foo/pt2=1" 或 "pt1=foo,pt2=1"
    */
   public PartitionSpec(String spec) {
+    this(spec, true);
+  }
+
+  public PartitionSpec(String spec, boolean trim) {
     if (spec == null) {
       throw new IllegalArgumentException("Argument \'spec\' cannot be null");
     }
@@ -59,8 +63,10 @@ public PartitionSpec(String spec) {
         throw new IllegalArgumentException("Invalid partition spec.");
       }
 
-      String k = kv[0].trim();
-      String v = kv[1].trim().replaceAll("'", "").replaceAll("\"", "");
+      String k = trim ? kv[0].trim() : kv[0];
+      String v = (trim ? kv[1].trim() : kv[1])
+          .replaceAll("'", "")
+          .replaceAll("\"", "");
       if (k.length() == 0 || v.length() == 0) {
         throw new IllegalArgumentException("Invalid partition spec.");
       }
diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Instance.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Instance.java
@@ -350,6 +350,8 @@ public void stop() throws OdpsException {
       String ret = SimpleXmlUtils.marshal(sm);
       Map<String, String> headers = getCommonHeaders();
       client.stringRequest(getResource(), "PUT", null, headers, ret);
+    } catch (OdpsException e) {
+      throw e;
     } catch (Exception e) {
       throw new OdpsException(e.getMessage(), e);
     }
diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Table.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Table.java
@@ -1690,7 +1690,7 @@ public List<PartitionSpec> getPartitionSpecs() throws OdpsException {
                                                      "GET",
                                                      params);
     for (PartitionSpecModel partitionSpecModel : resp.partitionSpecs) {
-      partitionSpecs.add(new PartitionSpec(partitionSpecModel.partitionSpec));
+      partitionSpecs.add(new PartitionSpec(partitionSpecModel.partitionSpec, false));
     }
 
     return partitionSpecs;
diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/SQLExecutorConstants.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/SQLExecutorConstants.java
@@ -3,7 +3,7 @@
 /**
  * Created by dongxiao on 2020/3/16.
  */
-class SQLExecutorConstants {
+public class SQLExecutorConstants {
   public static final String DEFAULT_TASK_NAME = "console_sqlrt_task";
   public static final String DEFAULT_SERVICE = "public.default";
   public static final String DEFAULT_OFFLINE_TASKNAME = "sqlrt_fallback_task";
diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/v2/SQLExecutorImpl.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/sqa/v2/SQLExecutorImpl.java
@@ -26,6 +26,7 @@
 import com.aliyun.odps.sqa.QueryInfo;
 import com.aliyun.odps.sqa.SQLExecutor;
 import com.aliyun.odps.sqa.SQLExecutorBuilder;
+import com.aliyun.odps.sqa.SQLExecutorConstants;
 import com.aliyun.odps.sqa.SQLExecutorPool;
 import com.aliyun.odps.sqa.commandapi.Command;
 import com.aliyun.odps.sqa.commandapi.CommandInfo;
@@ -35,6 +36,7 @@
 import com.aliyun.odps.table.utils.Preconditions;
 import com.aliyun.odps.task.SQLTask;
 import com.aliyun.odps.tunnel.InstanceTunnel;
+import com.aliyun.odps.tunnel.TunnelException;
 import com.aliyun.odps.tunnel.io.TunnelRecordReader;
 import com.aliyun.odps.utils.CSVRecordParser;
 import com.aliyun.odps.utils.StringUtils;
@@ -191,8 +193,13 @@ public boolean isActive() {
   public void cancel() throws OdpsException {
     if (queryInfo != null) {
       Instance instance = queryInfo.getInstance();
-      if (instance.getStatus() == Instance.Status.RUNNING) {
+      try {
         instance.stop();
+      } catch (OdpsException e) {
+        // stop will throw exception when instance is not running, so we check here.
+        if (!"InvalidStateSetting".equals(e.getErrorCode())) {
+          throw e;
+        }
       }
     }
   }
@@ -580,7 +587,12 @@ private ResultSet getResultSetByInstanceTunnel(Long offset, Long countLimit, Lon
         downloadSession =
             instanceTunnel.createDownloadSession(odps.getDefaultProject(), queryInfo.getInstance().getId(),
                                          limitEnabled);
-      } catch (OdpsException e) {
+      } catch (TunnelException e) {
+        if (e.getErrorCode().equals(SQLExecutorConstants.sessionNotSelectException)
+            || e.getErrorMsg().contains(SQLExecutorConstants.sessionNotSelectMessage)) {
+          queryInfo.setSelect(false);
+          return getResultSetByInstanceTunnel(offset, countLimit, sizeLimit, limitEnabled);
+        }
         if (e.getErrorCode().equals("TaskFailed")) {
           // wait for success will check task status and throw exception
           queryInfo.getInstance().waitForSuccess();
diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/impl/StreamUploadSessionImpl.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/impl/StreamUploadSessionImpl.java
@@ -272,11 +272,6 @@ private Connection getConnection(CompressOption compress, Slot slot, long size,
 
         headers.put(HttpHeaders.HEADER_ODPS_SLOT_NUM, String.valueOf(slots.getSlotNum()));
 
-        List<String> tags = config.getTags();
-        if (tags != null) {
-            headers.put(HttpHeaders.HEADER_ODPS_TUNNEL_TAGS, String.join(",", tags));
-        }
-
         if (!StringUtils.isNullOrEmpty(config.getQuotaName())) {
             params.put(TunnelConstants.PARAM_QUOTA_NAME, config.getQuotaName());
         }
@@ -460,50 +455,4 @@ public TableTunnel.StreamRecordPack newRecordPack(CompressOption option) throws
     public Record newRecord() {
         return new ArrayRecord(schema.getColumns().toArray(new Column[0]));
     }
-
-    // abort does not support retry
-    public void abort() throws TunnelException {
-        HashMap<String, String> params = new HashMap<String, String>();
-
-        params.put(TunnelConstants.UPLOADID, id);
-
-        if (this.partitionSpec != null && this.partitionSpec.length() > 0) {
-            params.put(TunnelConstants.RES_PARTITION, partitionSpec);
-        }
-
-        HashMap<String, String> headers = Util.getCommonHeader();
-        List<String> tags = config.getTags();
-        if (tags != null) {
-            headers.put(HttpHeaders.HEADER_ODPS_TUNNEL_TAGS, String.join(",", tags));
-        }
-        Slot slot = slots.iterator().next();
-        headers.put(HttpHeaders.HEADER_ODPS_ROUTED_SERVER, slot.getServer());
-
-        Connection conn = null;
-        String requestId = null;
-        try {
-            conn = httpClient.connect(getResource(), "POST", params, headers);
-            Response resp = conn.getResponse();
-            requestId = resp.getHeader(HEADER_ODPS_REQUEST_ID);
-
-            if (!resp.isOK()) {
-                throw new TunnelException(requestId, conn.getInputStream(), resp.getStatus());
-            }
-        } catch (IOException e) {
-            throw new TunnelException(requestId, "Failed abort upload session with tunnel endpoint "
-                    + httpClient.getEndpoint(), e);
-        } catch (TunnelException e) {
-            // Do not delete here! TunnelException extends from OdpsException.
-            throw e;
-        } catch (OdpsException e) {
-            throw new TunnelException("not available", e.getMessage(), e);
-        } finally {
-            if (conn != null) {
-                try {
-                    conn.disconnect();
-                } catch (IOException e) {
-                }
-            }
-        }
-    }
 }
diff --git a/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/TableTest.java b/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/TableTest.java
@@ -63,6 +63,7 @@
 import com.aliyun.odps.type.MapTypeInfo;
 import com.aliyun.odps.type.TypeInfo;
 import com.aliyun.odps.type.TypeInfoFactory;
+import com.google.common.collect.ImmutableMap;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import com.sun.org.apache.xml.internal.utils.URI;
@@ -263,6 +264,30 @@ public void testReadTable() throws OdpsException, IOException {
     Assert.assertEquals(count, limit);
   }
 
+  @Test
+  public void testListPartitionSpecsWithoutTrim() throws OdpsException {
+    String tableName = "testListPartitionSpecsWithoutTrim";
+    odps.tables().newTableCreator(tableName,
+                                  TableSchema.builder().withBigintColumn("id")
+                                      .withPartitionColumn(Column.newBuilder("char1", TypeInfoFactory.getCharTypeInfo(10)).build())
+                                      .withPartitionColumn(Column.newBuilder("char2", TypeInfoFactory.getCharTypeInfo(10)).build())
+        .build()).ifNotExists().withHints(ImmutableMap.of("odps.sql.type.system.odps2", "true")).debug().create();
+
+    Table table = odps.tables().get(tableName);
+
+    table.createPartition(new PartitionSpec("char1=' 1 ',char2=' bar '", false), true);
+
+    table.getPartitionSpecs().forEach(p -> {
+      System.out.println(p.toString());
+      try {
+        Assert.assertTrue(table.hasPartition(p));
+      } catch (OdpsException e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
+
+  /* BEGIN: opensource removal */
   @Test
   public void testColumnLabel() throws OdpsException {
     Table table = odps.tables().get(TABLE_NAME);
diff --git a/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/sqa/SQLExecutorV2Test.java b/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/sqa/SQLExecutorV2Test.java
@@ -43,6 +43,24 @@ private static Odps getTestOdps() throws OdpsException {
     odps.setDefaultProject("vw2");
     return odps;
   }
+
+
+  @Test
+  public void testCancelJobTest() throws OdpsException, IOException {
+    executor.run("select 1;", hints);
+    executor.getResultSet().forEach(System.out::println);
+    executor.cancel();
+    try {
+      executor.getInstance().stop();
+    } catch (OdpsException e) {
+      System.out.println(e.getErrorCode());
+      if (e.getErrorCode().equals("InvalidStateSetting")) {
+        System.out.println("good");
+      }
+    }
+  }
+
+
   @Test
   public void testGetTenantId() throws OdpsException {
     System.out.println(odps.projects().get("three_schema_project").getTenantId());

Original file line number	Diff line number	Diff line change
`@@ -350,6 +350,8 @@ public void stop() throws OdpsException {`
`350`	`350`	`String ret = SimpleXmlUtils.marshal(sm);`
`351`	`351`	`Map<String, String> headers = getCommonHeaders();`
`352`	`352`	`client.stringRequest(getResource(), "PUT", null, headers, ret);`
	`353`	`+ } catch (OdpsException e) {`
	`354`	`+ throw e;`
`353`	`355`	`} catch (Exception e) {`
`354`	`356`	`throw new OdpsException(e.getMessage(), e);`
`355`	`357`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1690,7 +1690,7 @@ public List<PartitionSpec> getPartitionSpecs() throws OdpsException {`
`1690`	`1690`	`"GET",`
`1691`	`1691`	`params);`
`1692`	`1692`	`for (PartitionSpecModel partitionSpecModel : resp.partitionSpecs) {`
`1693`		`- partitionSpecs.add(new PartitionSpec(partitionSpecModel.partitionSpec));`
	`1693`	`+ partitionSpecs.add(new PartitionSpec(partitionSpecModel.partitionSpec, false));`
`1694`	`1694`	`}`
`1695`	`1695`
`1696`	`1696`	`return partitionSpecs;`