Merge pull request aliyun#62 from aliyun/release/0.48.x

update version to 0.48.0-public-SNAPSHOT
dingxin-tech · Mar 6, 2024 · e9f3ca0 · e9f3ca0
2 parents 830edc3 + 3ca4173
commit e9f3ca0
Show file tree

Hide file tree

Showing 51 changed files with 1,264 additions and 352 deletions.
diff --git a/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/TableSchema.java b/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/TableSchema.java
@@ -23,6 +23,8 @@
 import java.util.HashMap;
 import java.util.List;
 
+import com.aliyun.odps.type.TypeInfoFactory;
+
 /**
  * TableSchema表示ODPS中表的定义
  */
@@ -41,6 +43,10 @@ public TableSchema() {
 
   }
 
+  public static Builder builder() {
+    return new Builder();
+  }
+
   /**
    * 表增加一列
    * 本方法不用于直接修改 SQL 表结构，适用于执行 MR 作业定义输入列等场景。
@@ -184,6 +190,12 @@ public List<Column> getPartitionColumns() {
     return (List<Column>) partitionColumns.clone();
   }
 
+
+  public List<Column> getAllColumns() {
+    List<Column> allColumns = getColumns();
+    allColumns.addAll(partitionColumns);
+    return allColumns;
+  }
   /**
    * 获得分区列定义
    *
@@ -234,4 +246,95 @@ public Column getPartitionColumn(int idx) {
     }
     return partitionColumns.get(idx);
   }
+
+  public boolean basicallyEquals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    TableSchema that = (TableSchema) o;
+    List<Column> columnsA = getAllColumns();
+    List<Column> columnsB = that.getAllColumns();
+    if (columnsA.size() != columnsB.size()) {
+      return false;
+    }
+    for (int i = 0; i < columnsA.size(); i++) {
+      Column columnA = columnsA.get(i);
+      Column columnB = columnsB.get(i);
+      if (!columnA.getName().equals(columnB.getName()) || !columnA.getTypeInfo().getTypeName()
+          .equals(columnB.getTypeInfo().getTypeName())) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  public static class Builder {
+
+    private final TableSchema schema;
+
+    public Builder() {
+      schema = new TableSchema();
+    }
+
+    public Builder withColumns(List<Column> columns) {
+      for (Column column : columns) {
+        withColumn(column);
+      }
+      return this;
+    }
+
+    public Builder withColumn(Column column) {
+      schema.addColumn(column);
+      return this;
+    }
+
+    public Builder withPartitionColumn(Column column) {
+      schema.addPartitionColumn(column);
+      return this;
+    }
+
+    /**
+     * easy method to add specific type column.
+     * for more column type, please use withColumn
+     * @see Column
+     * @see com.aliyun.odps.Column.ColumnBuilder
+     */
+
+    public Builder withBigintColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.BIGINT));
+      return this;
+    }
+
+    public Builder withStringColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.STRING));
+      return this;
+    }
+
+    public Builder withDoubleColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.DOUBLE));
+      return this;
+    }
+
+    public Builder withDecimalColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.DECIMAL));
+      return this;
+    }
+
+    public Builder withDatetimeColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.DATETIME));
+      return this;
+    }
+
+    public Builder withBooleanColumn(String columnName) {
+      schema.addColumn(new Column(columnName, TypeInfoFactory.BOOLEAN));
+      return this;
+    }
+
+    public TableSchema build() {
+      return schema;
+    }
+  }
 }
diff --git a/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/data/ArrayRecord.java b/odps-sdk/odps-sdk-commons/src/main/java/com/aliyun/odps/data/ArrayRecord.java
@@ -1169,5 +1169,18 @@ protected static byte[] stringToBytes(String string) {
     }
   }
 
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    for (Object o : values) {
+      if (o instanceof byte[]) {
+        sb.append(bytesToString((byte[]) o)).append(",");
+      } else {
+        sb.append(o.toString()).append(",");
+      }
+    }
+    sb.setLength(sb.length() - 1);
+    return sb.toString();
+  }
 }
 
diff --git a/odps-sdk/odps-sdk-core/pom.xml b/odps-sdk/odps-sdk-core/pom.xml
@@ -61,14 +61,14 @@
         <dependency>
             <groupId>org.lz4</groupId>
             <artifactId>lz4-pure-java</artifactId>
-            <version>1.8.0</version>
         </dependency>
 
         <dependency>
             <groupId>com.aliyun</groupId>
             <artifactId>aliyun-java-auth</artifactId>
         </dependency>
 
+
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>

diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Odps.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Odps.java
@@ -19,18 +19,18 @@
 
 package com.aliyun.odps;
 
-import com.aliyun.odps.account.AppAccount;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.Map;
 
 import com.aliyun.odps.account.Account;
 import com.aliyun.odps.account.AccountFormat;
+import com.aliyun.odps.account.AppAccount;
 import com.aliyun.odps.account.AppStsAccount;
 import com.aliyun.odps.commons.transport.DefaultTransport;
 import com.aliyun.odps.ml.OfflineModels;
 import com.aliyun.odps.rest.RestClient;
-
-import java.util.Map;
+import com.aliyun.odps.tunnel.TableTunnel;
 
 /**
  * Odps类是ODPS SDK的入口
@@ -75,6 +75,7 @@ public class Odps {
   private String endpoint;
   private String defaultProject;
   private String currentSchema;
+  private String tunnelEndpoint;
 
   /* resource collection objects */
   private Tenant tenant;
@@ -238,6 +239,10 @@ public Classifications classifications() {
     return classifications;
   }
 
+  public TableTunnel tableTunnel() {
+    return new TableTunnel(this);
+  }
+
   public Quotas quotas() {
     return quotas;
   }
@@ -294,6 +299,14 @@ public void setEndpoint(String endpoint) {
     }
   }
 
+  public void setTunnelEndpoint(String tunnelEndpoint) {
+    this.tunnelEndpoint = tunnelEndpoint;
+  }
+
+  public String getTunnelEndpoint() {
+    return tunnelEndpoint;
+  }
+
   /**
    * 取得当前对象的默认{@link Project}名称
    *

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
@@ -49,6 +49,8 @@
 import com.aliyun.odps.simpleframework.xml.convert.Converter;
 import com.aliyun.odps.simpleframework.xml.stream.InputNode;
 import com.aliyun.odps.simpleframework.xml.stream.OutputNode;
+import com.aliyun.odps.table.StreamIdentifier;
+import com.aliyun.odps.table.TableIdentifier;
 import com.aliyun.odps.task.SQLTask;
 import com.aliyun.odps.tunnel.TableTunnel;
 import com.aliyun.odps.utils.ColumnUtils;
@@ -1836,4 +1838,15 @@ public int getAcidDataRetainHours() {
     return model.acidDataRetainHours;
   }
 
+  public Stream newStream(String streamName) throws OdpsException {
+    if (!isTransactional()) {
+      throw new IllegalArgumentException("only transactional table can attach stream");
+    }
+    StreamIdentifier
+        identifier =
+        StreamIdentifier.of(model.projectName, streamName);
+    odps.streams().create(identifier,
+                          TableIdentifier.of(model.projectName, model.schemaName, model.name));
+    return odps.streams().get(identifier);
+  }
 }