prepare for 0.51.7-public

Author: dingxin-tech

odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/type/NestedTypeInfo.java

@@ -1,5 +1,6 @@
 package com.aliyun.odps.type;
 
+
 /**
  * @author dingxin ([email protected])
  */

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

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.Callable;
 
 import com.aliyun.odps.commons.transport.Headers;
 import com.aliyun.odps.commons.transport.Response;
@@ -91,7 +92,12 @@ public static enum ProjectType {
     /**
      * 映射到 Odps 的外部项目，例如 hive
      */
-    external
+    external,
+
+    /**
+     * 映射到 Odps 的外部项目 V2，后续将和 external 合并
+     */
+    external_v2
   }
 
   /**
@@ -428,12 +434,78 @@ public void reload() throws OdpsException {
 
       properties = model.properties;
       clusters = model.clusters;
+
+      if (isExternalV2(model.name)) {
+        model.type = ProjectType.external_v2.name();
+      }
     } catch (Exception e) {
       throw new OdpsException("Can't bind xml to " + ProjectModel.class, e);
     }
     setLoaded(true);
   }
 
+  /**
+   * judge whether external epv2
+   *
+   * @param projectName
+   * @throws OdpsException
+   */
+  private boolean isExternalV2(String projectName) throws OdpsException {
+    if (projectName.equalsIgnoreCase("system_catalog")) {
+      return false;
+    }
+    // get isEx from api-worker and put in cache
+    String propertyStr = properties.get("external_project_properties");
+    if (propertyStr != null) {
+      //judge isExternalCatalogBound=true whether in external_project_properties of properties
+      return propertyStr.contains("\"isExternalCatalogBound\":true");
+    } else {
+      return false;
+    }
+  }
+
+  /**
+   * 首先检查当前项目是否为 epv2（但检查时可能会报错）。
+   * 如果是 epv2，执行 ifEpv2。
+   * 如果不是 epv2，或者检查时报错，则执行 ifNotEpv2。
+   * 当执行 ifNotEpv2 时，如果报错且检查 project 时也报错，则将两个报错合成一个，抛出。
+   */
+  public <T> T executeIfEpv2(Callable<T> ifEpv2, Callable<T> ifNotEpv2) throws OdpsException {
+    Exception readProjectException = null;
+    boolean isEpv2;
+    try {
+      isEpv2 = getType() == ProjectType.external_v2;
+    } catch (Exception e) {
+      readProjectException = e;
+      isEpv2 = false;
+    }
+    try {
+      if (isEpv2) {
+        return ifEpv2.call();
+      } else {
+        return ifNotEpv2.call();
+      }
+    } catch (Exception e) {
+      // 处理 ifNotEpv2 执行中的异常
+      if (readProjectException != null) {
+        // 如果项目检查时有异常，合成两个异常的消息抛出
+        String errMsg = String.format(
+                "%s, If any entities in the statement you execute are from external projects, the error may be caused by [%s]",
+                e.getMessage(), readProjectException.getMessage());
+        if (e instanceof NoSuchObjectException) {
+          throw new NoSuchObjectException(errMsg, e);
+        }
+        throw new OdpsException(errMsg, e);
+      } else {
+        if (e instanceof OdpsException) {
+          throw (OdpsException) e;
+        } else {
+          throw new OdpsException(e.getMessage(), e);
+        }
+      }
+    }
+  }
+
   /**
    * 获取Project名称
    *
@@ -452,7 +524,6 @@ public ProjectType getType() throws OdpsException {
     if (model.type == null) {
       lazyLoad();
     }
-
     if (model.type != null) {
       try {
         return ProjectType.valueOf(model.type.toLowerCase());

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

@@ -92,7 +92,7 @@ public Project get() throws OdpsException {
    *
    * @param projectName {@link Project}名称
    * @return {@link Project}对象
-   * @throws NoSuchObjectException Project不存在
+   * @throws OdpsException No exception will throw, however for compatibility
    */
   public Project get(String projectName) throws OdpsException {
     ProjectModel model = new ProjectModel();

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

@@ -19,21 +19,28 @@
 
 package com.aliyun.odps;
 
+import static com.aliyun.odps.task.SQLTask.parseCsvRecord;
+
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
 
 import com.aliyun.odps.Schema.SchemaModel;
 import com.aliyun.odps.commons.transport.Headers;
+import com.aliyun.odps.data.Record;
 import com.aliyun.odps.rest.ResourceBuilder;
 import com.aliyun.odps.rest.RestClient;
 import com.aliyun.odps.rest.SimpleXmlUtils;
 import com.aliyun.odps.simpleframework.xml.Element;
 import com.aliyun.odps.simpleframework.xml.ElementList;
 import com.aliyun.odps.simpleframework.xml.Root;
 import com.aliyun.odps.simpleframework.xml.convert.Convert;
+import com.aliyun.odps.task.SQLTask;
 import com.aliyun.odps.utils.ExceptionUtils;
 import com.aliyun.odps.utils.StringUtils;
 
@@ -289,9 +296,62 @@ public String getMarker() {
       return params.get("marker");
     }
 
+
+    /**
+     * getExternalProjectSchemaList , special for EPV2
+     * @param projectName
+     * @return
+     * @throws OdpsException
+     */
+    public List<Schema> getExternalProjectSchemaList(String projectName) throws OdpsException {
+      //Temporary code for openlake Demo(YunQi big conference):
+      // Long-term this code block should be converted to SQL for both internal and external projects,
+      // and should not have any setting flag, except for opening the 3 layer model based on project attributes(user can choose use 2 or 3tier for 3 layer Project ).
+      //warning filter not support in EPV2
+      Map<String, String> queryHint = new HashMap<>();
+      InputStream is = null;
+      try {
+        is = Schemas.class.getResourceAsStream("/com/aliyun/odps/core/base.conf");
+        Properties properties = new Properties();
+        properties.load(is);
+        String majorVersion = properties.getProperty("epv2flighting");
+        if (majorVersion != null && !majorVersion.isEmpty() && !"default".equals(majorVersion)) {
+          queryHint.put("odps.task.major.version", majorVersion);
+        }
+
+      } catch (Exception e) {
+      } finally {
+        org.apache.commons.io.IOUtils.closeQuietly(is);
+      }
+      queryHint.put("odps.sql.select.output.format", "csv");
+      Instance i = SQLTask.run(odps, projectName, "show schemas;", queryHint, null);
+      i.waitForSuccess();
+      Instance.InstanceResultModel.TaskResult taskResult = i.getRawTaskResults().get(0);
+      Instance.TaskStatus.Status taskStatus =
+          Instance.TaskStatus.Status.valueOf(taskResult.status.toUpperCase());
+      if (taskStatus != Instance.TaskStatus.Status.SUCCESS) {
+        throw new RuntimeException("show schemas failed. instanceId:" + i.getId());
+      }
+
+      String result = taskResult.result.getString();
+      List<Record> schemalist = parseCsvRecord(result);
+      if (schemalist == null || schemalist.isEmpty()) {
+        return null;
+      }
+      ArrayList<Schema> schemas = new ArrayList<>();
+      for (Record s : schemalist) {
+        SchemaModel model = new SchemaModel();
+        model.name = s.get(0).toString();
+        Schema schema = new Schema(model, projectName, odps);
+        schema.setLoaded(true);
+        schemas.add(schema);
+      }
+      return schemas;
+    }
+
     @Override
     protected List<Schema> list() {
-      ArrayList<Schema> schemas = new ArrayList<>();
+      AtomicReference<List<Schema>> schemas = new AtomicReference<>(new ArrayList<>());
       params.put("expectmarker", "true");
       String lastMarker = params.get("marker");
       if (params.containsKey("marker") && StringUtils.isNullOrEmpty(lastMarker)) {
@@ -307,26 +367,28 @@ protected List<Schema> list() {
           params.put("owner", filter.getOwner());
         }
       }
-
       String resource = ResourceBuilder.buildSchemaResource(projectName);
-
       try {
-
-        ListSchemasResponse resp = client.request(
-                ListSchemasResponse.class, resource, "GET", params);
-
-        for (SchemaModel model: resp.schemas) {
-          Schema schema = new Schema(model, projectName, odps);
-          schemas.add(schema);
-        }
-
-        params.put("marker", resp.marker);
-
+        return odps.projects().get(projectName).executeIfEpv2(() -> {
+          schemas.set(getExternalProjectSchemaList(projectName));
+          if (schemas.get() == null || schemas.get().isEmpty()) {
+            return null;
+          }
+          params.put("marker", "");
+          return schemas.get();
+        }, () -> {
+          ListSchemasResponse resp = client.request(
+              ListSchemasResponse.class, resource, "GET", params);
+          for (SchemaModel model : resp.schemas) {
+            Schema schema = new Schema(model, projectName, odps);
+            schemas.get().add(schema);
+          }
+          params.put("marker", resp.marker);
+          return schemas.get();
+        });
       } catch (OdpsException e) {
-        throw new RuntimeException(e.getMessage(), e);
+        throw new UncheckedOdpsException(e);
       }
-
-      return schemas;
     }
   }
 }