prepare version 0.48.7-public

Author: dingxin-tech

CHANGELOG.md

@@ -1,5 +1,24 @@
 # Changelog
 
+## [0.48.7-public] - 2024-08-07
+### Enhancements
+- **TableTunnel Configuration Optimization**: Introduced the `tags` attribute to `TableTunnel Configuration`, enabling users to attach custom tags to tunnel operations for enhanced logging and management. These tags are recorded in the tenant-level `information schema`.
+```java
+Odps odps;
+Configuration configuration =
+    Configuration.builder(odps)
+                 .withTags(Arrays.asList("tag1", "tag2")) // Utilize Arrays.asList for code standardization
+                 .build();
+TableTunnel tableTunnel = odps.tableTunnel(configuration);
+// Proceed with tunnel operations
+```
+- **Instance Enhancement**: Added the `waitForTerminatedAndGetResult` method to the `Instance` class, integrating optimization strategies from versions 0.48.6 and 0.48.7 for the `SQLExecutor` interface, enhancing operational efficiency. Refer to `com.aliyun.odps.sqa.SQLExecutorImpl.getOfflineResultSet` for usage.
+
+### Improve
+- **SQLExecutor Offline Job Processing Optimization**: Significantly reduced end-to-end latency by enabling immediate result retrieval after critical processing stages of offline jobs executed by `SQLExecutor`, without waiting for the job to fully complete, thus boosting response speed and resource utilization.
+### Fixes
+- **TunnelRetryHandler NPE Fix**: Rectified a potential null pointer exception issue in the `getRetryPolicy` method when the error code (`error code`) was `null`.
+
 ## [0.48.6-public] - 2024-07-17
 
 ### Added

CHANGELOG_CN.md

@@ -1,25 +1,63 @@
 # 更新日志
 
+## [0.48.7-public] - 2024-08-07
+
+### 增强
+
+- **TableTunnel 配置优化**：引入 `tags` 属性至 `TableTunnel Configuration`
+  ，旨在允许用户为隧道相关操作附上自定义标签。这些标签会被记录在租户层级的 `information schema`
+  中，便于日志追踪与管理。
+
+```java
+Odps odps;
+    Configuration configuration=
+    Configuration.builder(odps)
+    .withTags(Arrays.asList("tag1","tag2")) // 使用 Arrays.asList 以提升代码规范性
+    .build();
+    TableTunnel tableTunnel=odps.tableTunnel(configuration);
+// 继续执行隧道相关操作
+```
+
+- **Instance 增强**：在 `Instance` 类中新增 `waitForTerminatedAndGetResult` 方法，此方法整合了 0.48.6
+  及 0.48.7 版本中对 `SQLExecutor`
+  接口的优化策略，提升了操作效率。使用方式可参考 `com.aliyun.odps.sqa.SQLExecutorImpl.getOfflineResultSet`
+  方法。
+
+### 优化
+
+- **SQLExecutor 离线作业处理优化**：显著减少了端到端延迟，通过改进使得由 `SQLExecutor`
+  执行的离线作业能在关键处理阶段完成后即刻获取结果，无需等待作业全部完成，提高了响应速度和资源利用率。
+
+### 修复
+
+- **TunnelRetryHandler NPE修复**：修正了 `getRetryPolicy` 方法中在错误码 (`error code`) 为 `null`
+  的情况下潜在空指针异常问题。
+
 ## [0.48.6-public] - 2024-07-17
 
 ### 新增
+
 - **支持序列化**：
-  - 主要数据类型如 `ArrayRecord`、`Column`、`TableSchema` 和 `TypeInfo` 现在支持序列化和反序列化，能够进行缓存和进程间通信。
+    - 主要数据类型如 `ArrayRecord`、`Column`、`TableSchema` 和 `TypeInfo` 现在支持序列化和反序列化，能够进行缓存和进程间通信。
 - **谓词下推**：
-  - 新增 `Attribute` 类型的谓词，用于指定列名。
+    - 新增 `Attribute` 类型的谓词，用于指定列名。
 
 ### 变更
+
 - **Tunnel 接口重构**：
-  - 重构了 Tunnel 相关接口，加入了无感知的重试逻辑，大大增强了稳定性和鲁棒性。
-  - 删除了 `TunnelRetryStrategy` 和 `ConfigurationImpl` 类，分别被 `TunnelRetryHandler` 和 `Configuration` 所取代。
+    - 重构了 Tunnel 相关接口，加入了无感知的重试逻辑，大大增强了稳定性和鲁棒性。
+    - 删除了 `TunnelRetryStrategy` 和 `ConfigurationImpl` 类，分别被 `TunnelRetryHandler`
+      和 `Configuration` 所取代。
 
 ### 优化
+
 - **SQLExecutor 优化**：
-  - 在使用 `SQLExecutor` 接口执行离线 SQL 作业时进行优化，减少每个作业获取结果时的一次网络请求，从而减少端到端延时。
+    - 在使用 `SQLExecutor` 接口执行离线 SQL 作业时进行优化，减少每个作业获取结果时的一次网络请求，从而减少端到端延时。
 
 ### 修复
+
 - **Table.read Decimal 读取**：
-  - 修复了 `Table.read` 接口在读取 `decimal` 类型时，后面补零不符合预期的问题。
+    - 修复了 `Table.read` 接口在读取 `decimal` 类型时，后面补零不符合预期的问题。
 
 ## [0.48.5-public] - 2024-06-17
 

odps-sdk-impl/odps-mapred-bridge/src/main/java/com/aliyun/odps/mapred/BridgeJobRunner.java

@@ -296,6 +296,7 @@ protected void setUp() throws OdpsException {
       OutputUtils.setTables(infos, job);
     }
 
+    getProjectModeConf();
     processTempResources();
 
     // Adding jobconf jar.
@@ -386,6 +387,16 @@ protected void tearDown() throws OdpsException {
     isClean = true;
   }
 
+  private void getProjectModeConf(){
+    try {
+      String mode = metaExplorer.getProjectProperty(SessionState.MR_EXECUTION_MODE);
+      if (mode != null) {
+        job.set("odps.mr.project.conf", mode);
+      }
+    } catch (OdpsException ignore) {
+    }
+  }
+
   abstract protected Instance submitInternal() throws OdpsException;
 
   @Override

odps-sdk-impl/odps-mapred-bridge/src/main/java/com/aliyun/odps/mapred/bridge/UDTFTaskContextImpl.java

@@ -371,6 +371,7 @@ protected void configure(ExecutionContext ctx) {
           this.pipeIndex = this.pipeIndex < 0 ? 0 :this.pipeIndex;
         }
       }
+      resolveSqlModeNodeIndex(exeMode, tid);
       this.pipeNode = pipeline.getNode(pipeIndex);
     }
 
@@ -388,6 +389,42 @@ protected void configure(ExecutionContext ctx) {
     }
   }
 
+  private void resolveSqlModeNodeIndex(String exeMode, String tid) {
+    if (exeMode != null && exeMode.equalsIgnoreCase("lot")) {
+      return;
+    }
+    TableInfo[] infos = InputUtils.getTables(conf);
+    if (infos == null || infos.length < 2) {
+      return;
+    }
+    String projectConfMode = "";
+    if (conf.get("odps.mr.project.conf") != null) {
+      projectConfMode = conf.get("odps.mr.project.conf").toLowerCase();
+    }
+    if (projectConfMode.equals("lot")) {
+      return;
+    }
+    if (tid.startsWith("M")) {
+      this.pipeIndex = 0;
+    } else {
+      int pipePrefix = Integer.parseInt(tid.split("_")[0].substring(1));
+      String taskIdPrefix = "R" + pipePrefix;
+      int prefixLength = pipePrefix > 9 ? 9 : pipePrefix;
+      for (int i = 1; i < prefixLength; i++) {
+        taskIdPrefix += "_" + i;
+      }
+      if (tid.startsWith(taskIdPrefix)) {
+        this.pipeIndex = 1;
+      } else if (tid.startsWith("R3_2_")) {
+        this.pipeIndex = 2;
+      } else if (this.pipeIndex > 2 &&
+              (projectConfMode.equals("sql") || projectConfMode.equals("hybrid"))) {
+        this.pipeIndex = pipePrefix - infos.length;
+        this.pipeIndex = this.pipeIndex < 0 ? 0 : this.pipeIndex;
+      }
+    }
+  }
+
   protected boolean hasLabel(String label) {
     if (label2offset.get(label) == null) {
       return false;

odps-sdk-impl/odps-mapred-bridge/src/main/java/com/aliyun/odps/mapred/bridge/sqlgen/SqlGenContext.java

@@ -201,27 +201,32 @@ public String getId() {
   public String getMapStreamProcessor() {
     String cmd = job.getStreamProcessor("map");
     if (cmd != null) {
-      try {
-        return URLDecoder.decode(cmd, "UTF-8");
-      } catch (UnsupportedEncodingException e) {
-        e.printStackTrace();
-      }
+      return decodeStreamProcessor(cmd);
     }
     return null;
   }
 
   public String getReduceStreamProcessor() {
     String cmd = job.getStreamProcessor("reduce");
     if (cmd != null) {
-      try {
-        return URLDecoder.decode(cmd, "UTF-8");
-      } catch (UnsupportedEncodingException e) {
-        e.printStackTrace();
-      }
+      return decodeStreamProcessor(cmd);
     }
     return null;
   }
 
+  String decodeStreamProcessor(String cmd) {
+    String result = null;
+    try {
+      result = URLDecoder.decode(cmd, "UTF-8");
+      if (result.contains("'")) {
+        result = result.replaceAll("'", "\\\\'");
+      }
+    } catch (UnsupportedEncodingException e) {
+      e.printStackTrace();
+    }
+    return result;
+  }
+
   public String getFunctionCreateText() {
     return job.getFunctionCreateText();
   }

odps-sdk-impl/odps-mapred-bridge/src/main/java/com/aliyun/odps/mapred/bridge/sqlgen/SqlGenerator.java

@@ -45,9 +45,6 @@
 public class SqlGenerator {
 
   public static String generate(JobConf job, String id, MetaExplorer metaExplorer, Map<String, String> aliasToTempResource) {
-    if(!isSqlMode(job, metaExplorer)) {
-      return null;
-    }
     createFunction((BridgeJobConf) job, id, aliasToTempResource);
     Properties p = new Properties();
     p.setProperty("resource.loader", "class");
@@ -103,6 +100,13 @@ static void createFunction(BridgeJobConf job, String jobId, Map<String, String>
       resouceText.append(s).append(",");
     }
     if (job.isStreamJob()) {
+      if (job.getResources() != null && aliasToTempResource != null) {
+        for (String s : job.getResources()) {
+          if (!aliasToTempResource.containsKey(s)) {
+            resouceText.append(s).append(",");
+          }
+        }
+      }
       resouceText.deleteCharAt(resouceText.lastIndexOf(","));
       createText.append(String.format("set %s=", SessionState.MR_EXECUTION_SESSION_RESOURCES));
       if (job.get(SessionState.MR_EXECUTION_SESSION_RESOURCES) != null) {

odps-sdk-impl/odps-mapred-bridge/src/main/resources/pl.vm

@@ -71,9 +71,9 @@ $!{ctx.FunctionCreateText}
         #end
         #if(${ctx.isNoInputTableInfos()})
         #if(!${ctx.mapOnly()})
-        SELECT mr2sql_mapper_${ctx.Id}_${nodeId}() as (${node.IntermediateColsJoinedMapOut}) FROM values('') _t (_c)
+        SELECT mr2sql_mapper_${ctx.Id}_${nodeId}() as (${node.IntermediateColsJoinedMapOut}) FROM values('') _t (_c) where false
         #else
-        SELECT mr2sql_mapper_${ctx.Id}_${nodeId}() as (${ctx.PackagedColsJoined}) FROM values('') _t (_c)
+        SELECT mr2sql_mapper_${ctx.Id}_${nodeId}() as (${ctx.PackagedColsJoined}) FROM values('') _t (_c) where false
         #end##
         #end
         ) open_mr_alias_${nodeId}_1

odps-sdk-impl/odps-mapred-bridge/src/main/resources/sql.vm

@@ -68,9 +68,9 @@ $!{ctx.functionCreateText}
         #end##
         #if(${ctx.isNoInputTableInfos()})
 		#if(!${ctx.mapOnly()})
-        SELECT mr2sql_mapper_${ctx.Id}() as (${ctx.IntermediateColsJoinedMapOut}) FROM values('') _t (_c)
+        SELECT mr2sql_mapper_${ctx.Id}() as (${ctx.IntermediateColsJoinedMapOut}) FROM values('') _t (_c) where false
         #else
-        SELECT mr2sql_mapper_${ctx.Id}() as (${ctx.PackagedColsJoined}) FROM values('') _t (_c)
+        SELECT mr2sql_mapper_${ctx.Id}() as (${ctx.PackagedColsJoined}) FROM values('') _t (_c) where false
         #end##
         #end
     ) open_mr_alias1
@@ -104,4 +104,4 @@ SELECT #collist($output_table)
 #else
 SELECT *
 #end
-;
+;

odps-sdk-impl/odps-mapred-bridge/src/main/resources/stream.vm

@@ -48,31 +48,43 @@ ${entry.Key} = "${entry.Value}" #if(${foreach.hasNext}) AND #end
 
 $!{ctx.functionCreateText}
 @sub_query_mapper :=
-    FROM
-    (
+    select * FROM (
     #foreach ( $input_table in ${ctx.InputTableInfos} )
-    SELECT #selist($input_table ${ctx.InputTableInfos[0].Cols}) FROM #tblname($input_table)
-        #if (${input_table.PartSpec} && ${input_table.PartSpec.size()} > 0)
-        WHERE #partfilter($input_table)
+        FROM (SELECT * FROM #tblname($input_table)
+                #if (${input_table.PartSpec} && ${input_table.PartSpec.size()} > 0)
+                WHERE #partfilter($input_table)
+                #end##
+        )
+        MAP #collist($input_table) $!{ctx.mapInputSeparator}
+        USING '$!{ctx.MapStreamProcessor}'
+        #if(!${ctx.mapOnly()})
+        AS ${ctx.IntermediateColsJoinedMapOut} $!{ctx.mapOutputSeparator}
+        #else
+        AS #outputColsJoined() $!{ctx.mapOutputSeparator}
+        #end##
+        #if(${foreach.hasNext})UNION ALL
         #end##
-        #if(${foreach.hasNext})UNION ALL#end##
     #end##
     #if(${ctx.isNoInputTableInfos()})
-    select _c from values('') _t (_c)
+        FROM (select _c from values('') _t (_c) where false)
+        MAP '' $!{ctx.mapInputSeparator}
+        USING '$!{ctx.MapStreamProcessor}'
+        #if(!${ctx.mapOnly()})
+        AS ${ctx.IntermediateColsJoinedMapOut} $!{ctx.mapOutputSeparator}
+        #else
+        AS #outputColsJoined() $!{ctx.mapOutputSeparator}
+        #end##
     #end
     ) open_mr_alias1
-    MAP #inputColsJoined()
-    USING '$!{ctx.MapStreamProcessor}' $!{ctx.mapInputSeparator}
 #if(!${ctx.mapOnly()})
-    AS ${ctx.IntermediateColsJoined} $!{ctx.mapOutputSeparator};
-#else
-    AS #outputColsJoined() $!{ctx.mapOutputSeparator};
-#end##
+DISTRIBUTE BY ${ctx.PartitionColsJoined} SORT BY ${ctx.SortColsJoined}
+#end
+;
 #if(!${ctx.mapOnly()})
 
 @sub_query_reducer :=
-    REDUCE ${ctx.IntermediateColsJoined}
-    USING '$!{ctx.ReduceStreamProcessor}' $!{ctx.reduceInputSeparator}
+    REDUCE ${ctx.IntermediateColsJoined} $!{ctx.reduceInputSeparator}
+    USING '$!{ctx.ReduceStreamProcessor}'
     AS #outputColsJoined() $!{ctx.reduceOutputSeparator}
     FROM @sub_query_mapper;
 #end##

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

@@ -903,7 +903,7 @@ public void waitForSuccess() throws OdpsException {
    *     Instance失败
    */
   public void waitForSuccess(long interval) throws OdpsException {
-    waitForTerminated(interval);
+    waitForTerminated(interval, false);
 
     if (!isSuccessful()) {
       for (Entry<String, TaskStatus> e : getTaskStatus().entrySet()) {
@@ -921,9 +921,14 @@ public void waitForSuccess(long interval) throws OdpsException {
    *
    * @param interval
    *     内部轮询间隔
+   * @param isBlock
+   *     是否阻塞,
+   *     不开启 block，将在客户端长轮询，直到作业结束
+   *     开启 block 请求将会在服务端等待一段时间（每次请求等待5s，直到作业结束。long-polling），
+   *     block 模式能够跳过 instance post running 阶段（key-path-end optimize）
    */
-  public void waitForTerminated(long interval) {
-    while (!isTerminated()) {
+  public void waitForTerminated(long interval, boolean isBlock) {
+    while (getStatus(isBlock) != Instance.Status.TERMINATED) {
       try {
         Thread.sleep(interval);
       } catch (InterruptedException e) {
@@ -937,7 +942,7 @@ public void waitForTerminated(long interval) {
    * 该方法仅适用于离线作业（1个Instance对应1个Task）
    */
   public String waitForTerminatedAndGetResult() throws OdpsException {
-    waitForTerminated(1000);
+    waitForTerminated(1000, true);
 
     TaskResult taskResult = getRawTaskResults().get(0);
     String resultStr = taskResult.getResult().getString();