diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c40c9fd..2ab04901 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,7 +17,9 @@ - Renamed the `withBucketNum` method to `withDeltaTableBucketNum` to indicate this method is for Delta Tables only. - Modified the logic of `withHints`, `withAlias`, `withTblProperties`, and `withSerdeProperties` methods, now overwriting previous values instead of merging. - Removed the `createExternal` method; you can now use the `create` method instead. -- **Table** Introduced the `getSchemaVersion` method, allowing users to retrieve the current schema version of the table. The version number is updated each time a Schema Evolution occurs, and this field is used primarily for specifying when creating a StreamTunnel. +- **Table** + - Introduced the `getSchemaVersion` method, allowing users to retrieve the current schema version of the table. The version number is updated each time a Schema Evolution occurs, and this field is used primarily for specifying when creating a StreamTunnel. + - Added `setLifeCycle`, `changeOwner`, `changeComment`, `touch`, `changeClusterInfo`, `rename`, `addColumns`, `dropColumns` methods to support modification of table structure. - **StreamTunnel** Modified the initialization logic; if `allowSchemaMismatch` is set to `false`, it will automatically retry until the latest version of the table structure is used (with a timeout of 5 minutes). ### Fixes diff --git a/CHANGELOG_CN.md b/CHANGELOG_CN.md index 9be04628..dff9629f 100644 --- a/CHANGELOG_CN.md +++ b/CHANGELOG_CN.md @@ -17,7 +17,9 @@ - 重命名`withBucketNum`方法为`withDeltaTableBucketNum`，以表示该方法仅用于 Delta Table - 修改了 `withHints`，`withAlias`，`withTblProperties`，`withSerdeProperties` 方法的逻辑，现在会覆盖之前设置的值，而不是合并 - 移除了`createExternal`方法，现在使用`create`方法即可 -- **Table** 新增 `getSchemaVersion` 方法，用户获取当前表结构的版本，用户每次进行 SchemaEvolution 都会更新版本号，目前该字段仅用于在创建 StreamTunnel 时指定 +- **Table** + - 新增 `getSchemaVersion` 方法，用户获取当前表结构的版本，用户每次进行 SchemaEvolution 都会更新版本号，目前该字段仅用于在创建 StreamTunnel 时指定 + - 新增 `setLifeCycle`，`changeOwner`，`changeComment`，`touch`，`changeClusterInfo`，`rename`，`addColumns`，`dropColumns`方法，以支持对表结构进行修改 - **StreamTunnel** 修改初始化逻辑，当指定 `allowSchemaMismatch` 为 `false` 时，会自动重试直到使用最新版本的表结构(超时时间为5min) ### 修复 diff --git a/docs/docs/api-reference/Table.md b/docs/docs/api-reference/Table.md index ef0baf18..63caaa17 100644 --- a/docs/docs/api-reference/Table.md +++ b/docs/docs/api-reference/Table.md @@ -12,6 +12,7 @@ sidebar_position: 2 - [创建表实例对象](#创建表实例对象) - [表基本信息](#表基本信息) - [表数据操作](#表数据操作) +- [表更新操作](#表更新操作) - [分区操作](#分区操作) - [标签操作](#标签操作) - [表扩展信息](#表扩展信息) @@ -23,7 +24,7 @@ sidebar_position: 2 要操作表，首先需要创建一个表的实例对象。 -注意，获取表实例是一个lazy操作，即只有当调用`Table` 类的其他方法时，才会真正获取表的元数据信息。同时，只有表真实存在，才能获取到表实例。 +**注意**，获取表实例是一个lazy操作，即只有当调用`Table` 类的其他方法时，才会真正获取表的元数据信息。同时，只有表真实存在，才能获取到表实例。 ```java Table table = odps.tables().get("table_project", "table_name"); ``` @@ -132,13 +133,13 @@ Date lastMetaModifiedTime = table.getLastMetaModifiedTime(); ``` ### 获取表存储大小 -注意：此方法通常不保证和实际占用存储大小相同。单位为`bytes` +**注意**：此方法通常不保证和实际占用存储大小相同。单位为`bytes` ```java long size = table.getSize(); ``` ### 获取表数据数量 -注意：此方法当无准确数据时，返回-1 +**注意**：此方法当无准确数据时，返回-1 ```java long recordNum = table.getRecordNum(); ``` @@ -169,17 +170,17 @@ public RecordReader read(PartitionSpec partition, List columns, int limi public RecordReader read(PartitionSpec partition, List columns, int limit, String timezone) throws OdpsException ``` -参数说明 +**参数说明** - `limit`: 最多读取的记录行数。如果小于0，将抛出异常。 - `partition`: 表的分区（PartitionSpec 对象）。如果不指定分区，则传入 null。 - `columns`: 所要读取的列名列表。如果读取全表，则传入 null。 - `timezone`: 设置 datetime 类型数据的时区字符串，如 "Asia/Shanghai"。如果不设置，则使用默认时区。 -注意 +**注意** - 读取数据时，最多返回 1W 条记录，若超过，数据将被截断。 - 读取的数据大小不能超过 10MB，否则将抛出异常。 -使用示例 +**示例** ```java Table table = ...; // 获取Table对象的代码 try { @@ -200,6 +201,183 @@ try { table.truncate(); ``` + +## 表更新操作 + +### 修改表生命周期 + +修改已存在的分区表或非分区表的生命周期。 + +```java +public void setLifeCycle(int days) throws OdpsException +``` + +**参数说明** +- `days`: 新的表生命周期，单位为天，必须为正整数。 + +**示例** + +```java +table.setLifeCycle(90); +``` + +### 更改表所有者 + +更改表的所有者。只有项目所有者或具备超级管理角色的用户可以执行此命令。 + +```java +public void changeOwner(String newOwner) throws OdpsException +``` + +**参数说明** +- `newOwner`: 新的所有者ID，必须是项目内有效的用户ID。 + +**示例** + +```java +table.changeOwner("new_owner_id"); +``` + +### 修改表注释 + +修改表的注释内容。 + +```java +public void changeComment(String newComment) throws OdpsException +``` + +**参数说明** +- `newComment`: 新的注释文本，可以是空字符串。 + +**示例** + +```java +table.changeComment("This table contains user data."); +``` + +### 更新时间戳 + +更新时间戳，将表的最后修改时间更新为当前时间。 + +```java +public void touch() throws OdpsException +``` + +**示例** + +```java +table.touch(); +``` + +### 更改集群信息 + +修改表的集群信息。 + +```java +public void changeClusterInfo(ClusterInfo clusterInfo) throws OdpsException +``` + +**参数说明** +- `clusterInfo`: 要应用于表的新集群信息对象。 + +**示例** + +```java +ClusterInfo newClusterInfo = ...; // 创建或获取新的 ClusterInfo 对象 +table.changeClusterInfo(newClusterInfo); +``` + +### 重命名表 + +重命名表。 + +```java +public void rename(String newName) throws Exception +``` + +**参数说明** +- `newName`: 表的新名称，必须符合命名规则。 + +**示例** + +```java +table.rename("new_table_name"); +``` + +### 添加列 + +向表中添加新列。 + +```java +public void addColumns(List columns, boolean ifNotExists) throws Exception +``` + +**参数说明** +- `columns`: 要添加的列的列表，每个列应具有名称、类型和可选的注释。 +- `ifNotExists`: 如果为真，操作不会在列已存在时抛出错误。 + +**示例** + +```java +List newColumns = ...; // 创建或获取新的 Column 对象列表 +table.addColumns(newColumns, true); +``` + +### 删除列 + +从表中删除列。 + +```java +public void dropColumns(List columnNames) throws Exception +``` + +**参数说明** +- `columnNames`: 要删除的列名列表，每个名称必须有效。 + +**示例** + +```java +List columnsToDrop = List.of("column1", "column2"); +table.dropColumns(columnsToDrop); +``` + +### 更改列类型 + +更改表中现有列的类型。 + +```java +public void alterColumnType(String columnName, TypeInfo columnType) throws Exception +``` + +**参数说明** +- `columnName`: 要更改的列名称。 +- `columnType`: 新的列类型信息。 + +**示例** + +```java +TypeInfo newType = ...; // 创建或获取新的 TypeInfo 对象 +table.alterColumnType("existing_column", newType); +``` + +### 更改列名 + +更改表中特定列的名称。 + +```java +public void changeColumnName(String oldColumnName, String newColumnName) throws Exception +``` + +**参数说明** +- `oldColumnName`: 当前列名称。 +- `newColumnName`: 列的新名称。 + +**示例** + +```java +table.changeColumnName("old_name", "new_name"); +``` + ## 分区操作 ### 获取分区 @@ -214,6 +392,13 @@ Partition partition = table.getPartition(partitionSpec); List partitions = table.getPartitions(); ``` +### 获取所有分区值 +与 `getPartitions()` 不同，此方法仅会返回分区值，而不会返回分区的详细信息，因此这个接口效率会更高。 + +```java +List partitions = table.getPartitionSpecs(); +``` + ### 判断分区是否存在 ```java @@ -247,7 +432,7 @@ List tags = table.getTags("columnName"); ``` ### 添加标签 -注意：表和标签应当属于同一个`project` +**注意**：表和标签应当属于同一个`project` ```java // 表级别 table.addTag(tag); @@ -311,13 +496,13 @@ table.isTransactional(); ``` ### 查看表所占磁盘的物理大小 -注意：此类方法通常为估计值，不保证准确性 +**注意**：此类方法通常为估计值，不保证准确性 ```java long physicalSize = table.getPhysicalSize(); ``` ### 查看表占用文件数 -注意：此类方法通常为估计值，不保证准确性 +**注意**：此类方法通常为估计值，不保证准确性 ```java long fileNum = table.getFileNum(); ``` diff --git a/docs/docs/core-concept/create-table.md b/docs/docs/core-concept/create-table.md index ea66f528..4ae7870e 100644 --- a/docs/docs/core-concept/create-table.md +++ b/docs/docs/core-concept/create-table.md @@ -3,4 +3,265 @@ title: 执行建表操作 sidebar_label: 执行建表操作 sidebar_position: 3 --- -本文档待完善 \ No newline at end of file + +# 执行建表操作 + +`Tables.TableCreator` 是一个便捷的建表工具，旨在简化和标准化表创建过程中所需的各项步骤。 +在使用 `odps-sdk-java` 的过程中，用户可以通过 `TableCreator` 类来更高效地执行建表操作。 + +下面我们将依次详细介绍如何使用 `TableCreator` 类，包括如何获取对象、构建表结构、指定参数等。 + +## 获取 TableCreator 对象 + +使用 `TableCreator` 前，您需要先获取到 `Odps` 实例，获取`Odps`示例的方法参考[构建 ODPS 客户端](./init-odps-client.md)。 +同时你必须提供希望创建表所在的表名，以及表的模式（即 `TableSchema`），如何构建`TableSchema`可以参考本文[构建表结构](#构建表结构)。 + +以下是获取 `TableCreator` 对象的示例： + +```java +Odps odps = new Odps(...); // 创建 Odps 实例 +String projectName = "your_project"; // 项目名称 +String tableName = "your_table"; // 表名称 +TableSchema tableSchema = new TableSchema(...); // 表结构定义 + +Tables.TableCreator tableCreator = + odps.tables().newTableCreator(projectName, tableName, tableSchema); +``` +也可以不指定项目名，将在`Odps.defaultProject`中创建表。 +```java +Tables.TableCreator tableCreator = odps.tables().newTableCreator(tableName, tableSchema); +``` + +## 构建表结构 + +`TableSchema` 是描述表结构的核心对象，包括列名、数据类型及其其他特性。创建 `TableSchema` 对象通常需要定义多个 `Column` 对象，如何构建`Column`可以参考本文[构造列对象](#构造列对象)。 + +`TableSchema` 主要包含表的数据列和分区列信息，以下是构建 `TableSchema` 示例： +```java +Column dataColumn1 = new Column(...); +Column dataColumn2 = new Column(...); +List moreDataColumns = List.of(...); + +Column partitionColumn1 = new Column(...); +Column partitionColumn2 = new Column(...); + +TableSchema tableSchema = TableSchema.builder() + .withColumn(dataColumn1) + .withColumn(dataColumn2) + .withColumns(moreDataColumns) + .withPartitionColumn(partitionColumn1) + .withPartitionColumn(partitionColumn2) + .build(); +``` + +同时`TableSchema`还拥有几个简单的方法，来加速简单表结构的构建，如： +```java +TableSchema tableSchema = TableSchema.builder() + .withBigintColumn("bigint") + .withStringColumn("string") + .withDoubleColumn("double") + .withDecimalColumn("decimal") + .withDatetimeColumn("datetime") + .withBooleanColumn("boolean") + .build(); +``` + +## 构造列对象 + +`Column` 对象用于定义每个列的属性，包括列名和数据类型和其他特性。可以使用 `Column.newBuilder` 方法来构建列对象，示例如下： + +```java +Column idColumn = Column.newBuilder("id", TypeInfoFactory.BIGINT).build(); +Column nameColumn = Column.newBuilder("name", TypeInfoFactory.STRING).build(); +``` +`TypeInfoFactory` 是一个工厂类，用于创建不同类型的 `TypeInfo` 对象，如 `BIGINT`、`STRING`、`DOUBLE` 等。 +支持 MaxCompute 全部支持的列类型，详情参考[数据类型](https://help.aliyun.com/zh/maxcompute/user-guide/maxcompute-v2-0-data-type-edition)。 + +`ColumnBuilder` 额外支持配置列的属性，如： +- **设置列是否为非空**：`notNull` 方法 +- **设置列的注释**：`withComment` 方法 +- **设置列的默认值**：`withDefaultValue` 方法 +- **设置列的生成表达式**：`withGenerateExpression` 方法（仅分区列生效） + +## 指定建表参数 +`TableCreator` 提供了多种方法来灵活配置建表参数。以下是 `TableCreator` 类中可用的主要配置选项： + +### 表的基本信息配置 +- **withSchemaName(String schemaName)**：设置表所在 Schema 的名称（三层模型）。 +- **withComment(String comment)**：为表添加描述信息。 +- **ifNotExists()**：设置在表已存在的情况下不抛出异常，而是忽略创建请求。 +- **withHints(Map hints)**：所执行的 SQL 任务的 hints。 +- **withAliases(Map aliases)**：指定 SQL 语句中的别名。 + +### 表类型配置 +- **virtualView()**：将表设置为虚拟视图。仅支持 `create view as` 语句。 +- **transactionTable()**：将表设置为事务表。 +- **deltaTable()**：将表设置为 Delta 表。Delta 表必须配置主键信息，主键列必须`NotNull`。 +- **externalTable()**：将表设置为外部表。 + +### 表属性和选项配置 +- **withLifeCycle(Long lifeCycle)**：设置表的生命周期。 +- **withTblProperties(Map tblProperties)**：设置用户自定义的表属性。 +- **autoPartitionBy(GenerateExpression expression, String aliasName)**：创建 auto-partition 表，需要指定表的生成表达式和分区列名。该方法将使用指定分区信息，覆盖当前 TableSchema 中的分区信息。 + +### 聚簇表相关配置 +- **withClusterInfo(Table.ClusterInfo clusterInfo)**：为表配置聚簇信息，用来创建 Hash/Range Cluster 表。 + +### Delta 表相关配置 +- **withPrimaryKeys(List primaryKeys)**：指定表的主键列，仅`Delta Table`具有限定 distinct 作用。 +- **withDeltaTableBucketNum(Integer bucketNum)**：为 Delta 表设置桶的数量。 + +### 外表相关配置 +- **withStorageHandler(String storageHandler)**：按照外部表数据格式指定StorageHandler。 +- **withResources(List usingResources)**：指定使用的外部资源。 +- **withLocation(String location)**：外部表数据存储位置。 +- **withSerdeProperties(Map serdeProperties)**：外部表的授权、压缩、字符解析等相关参数。 + +### 视图相关配置 +- **withSelectStatement(String selectStatement)**：`create view as`/`create table as` 语句的 select cause。 + +### 其他配置 +- **withDataHubInfo(DataHubInfo dataHubInfo)**：创建 DataHub 表的相关信息。 +- **debug()**：启用调试模式，输出执行日志到控制台。 + +### SQL 生成和执行 +- **getSQL()**：生成用于创建表的 SQL 语句。 +- **create()**：执行创建表操作，实际在 ODPS 上创建表。 + +这些方法可以组合使用，以便根据具体的用例构建复杂的表定义。 通过灵活使用这些方法，您可以根据业务需求定制表的创建。 + +## 示例 + +完成所有配置后，可以通过 `getSQL` 方法获取创建表的 SQL 语句，或者直接调用 `create` 方法来执行建表操作： + +### 创建分区表 +分区表是用于优化大数据的存储和查询性能的一种表结构，适合存储海量数据。通过为表设置分区列，可以有效地提高查询效率和管理数据。 +以下示例演示如何创建一个分区表，定义 p_date 列作为分区列，并在表中添加其他数据列。 +```java +public void createPartitionTable() throws OdpsException { + TableSchema schema = TableSchema.builder() + .withColumn(Column.newBuilder("c1", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c2", TypeInfoFactory.STRING).build()) + .withPartitionColumn(Column.newBuilder("p_date", TypeInfoFactory.STRING).build()) + .build(); + odps.tables().newTableCreator("testProject", "testCreatePartitionTable", schema) + .ifNotExists() + .create(); +} +``` + +### 创建 Delta 表 +Delta 表是一种用于处理事务性数据的表类型，具有高性能和一致性保证。 +下列示例将创建一个 Delta 表，以 BIGINT 类型 'pk' 作为主键，其他列类型为 STRING，分桶数量设置为 16。 +```java +public void createDeltaTable() throws OdpsException { + TableSchema schema = TableSchema.builder() + .withColumn(Column.newBuilder("pk", TypeInfoFactory.BIGINT).notNull().build()) + .withColumn(Column.newBuilder("c1", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c2", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c3", TypeInfoFactory.STRING).build()) + .build(); + + odps.tables().newTableCreator("testProject", "testCreateDeltaTable", schema) + .deltaTable() + .withDeltaTableBucketNum(16) + .withPrimaryKeys(ImmutableList.of("pk")) + .ifNotExists() + .create(); +} +``` + +### 创建 Transaction 表 +Transaction 表是一种用于处理事务性数据的表类型。 +下列示例将创建一个 Transaction 表，使用 struct 类型作为列类型 +```java +public void createStructTransactionalTable() throws OdpsException { + StructTypeInfo structTypeInfo = TypeInfoFactory.getStructTypeInfo( + ImmutableList.of("end"), ImmutableList.of(TypeInfoFactory.STRING)); + + TableSchema schema = TableSchema.builder() + .withColumn(Column.newBuilder("c1", structTypeInfo).build()).build(); + + odps.tables().newTableCreator("testProject", "testCreateStructTransactionalTable", schema) + .transactionTable() + .ifNotExists() + .create(); +} +``` + +### 创建 auto-partition 表 +auto-partition 表是一种用于自动生成分区的表类型，当用户不指定分区信息时，系统会根据指定的生成表达式自动生成分区。 +下列示例将创建一个 auto-partition 表，其中 `p1` 列使用 `TruncTime` 生成表达式，根据 `c1` 列生成，`c1` 列存储日期时间数据。 +```java +public void createAutoPartitionTable() throws OdpsException { + Column autoParitionCol = Column.newBuilder("p1", TypeInfoFactory.STRING) + .withGenerateExpression(new TruncTime("c1", TruncTime.DatePart.DAY)) + .build(); + TableSchema schema = TableSchema.builder() + .withColumn(Column.newBuilder("c1", TypeInfoFactory.DATETIME).build()) + .withPartitionColumn(autoPar) + .build(); + odps.tables().newTableCreator("testProject", "testCreateAutoPartitionTable", schema) + .ifNotExists() + .create(); +} +``` +### 创建 Cluster 表 +Cluster 表是用于分布式处理的表，这种表可以通过指定的列进行哈希分区/Row-Range分区，从而实现负载均衡。 +通过在创建表时设置 `CLUSTERED BY` 语句，可以增强数据的存取效率。下面的示例代码演示如何创建一个 Hash Cluster 表，选择 `c1` 和 `c2` 两个列作为分区依据，并定义排序方式。 + +```java +public void createClusterTable() throws OdpsException { + Table.ClusterInfo clusterInfo = + new Table.ClusterInfo(Table.ClusterInfo.ClusterType.HASH, + ImmutableList.of("c1", "c2"), // clustered by 列 + ImmutableList.of( // 桶内排序列 + new Table.SortColumn("c1", Table.SortColumn.Order.DESC), + new Table.SortColumn("c2", Table.SortColumn.Order.ASC)), + 10); // 桶数量 + + odps.tables().newTableCreator("testProject", "testCreateClusterTable", + TableSchema.builder() + .withColumn(Column.newBuilder("c1", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c2", TypeInfoFactory.STRING).build()) + .build()) + .withClusterInfo(clusterInfo).create(); +} +``` + +### 创建外表 +外表用于连接数据仓库外部的数据源，比如OSS存储或其他数据源。 +以下示例展示了如何创建一个外表，指定它的数据存储处理器以及其它相关参数，如位置和序列化属性。 + +```java +public void testCreateExternalTable() throws OdpsException { + odps.tables().newTableCreator("testProject", "testCreateExternalTable", + TableSchema.builder() + .withColumn(Column.newBuilder("c1", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c2", TypeInfoFactory.STRING).build()) + .build()) + .externalTable() + .withStorageHandler("com.aliyun.odps.udf.example.text.TextStorageHandler") + .withResources(ImmutableList.of("odps-udf-example.jar", "another.jar")) + .withLocation("oss://path/to/your/external/data/location/") + .withSerdeProperties( + ImmutableMap.of("odps.text.option.delimiter", "|", "my.own.option", "value")) + .create(); +} +``` + +### 创建虚拟视图 +虚拟视图是一种用于查询数据源的视图，可以简化数据查询。 +以下示例展示了如何创建一个虚拟视图，指定它的查询语句。 +```java +public void testCreateView() throws OdpsException { + odps.tables().newTableCreator("testProject", "testCreateView", + TableSchema.builder() + .withColumn(Column.newBuilder("c1", TypeInfoFactory.STRING).build()) + .withColumn(Column.newBuilder("c2", TypeInfoFactory.STRING).build()) + .build()) + .view() + .withSelectStatement("select c1, c2 from testProject.testCreateView") + .create(); +} +``` \ No newline at end of file diff --git a/odps-examples/basic-examples/pom.xml b/odps-examples/basic-examples/pom.xml index 05169ed0..62ea4bc0 100644 --- a/odps-examples/basic-examples/pom.xml +++ b/odps-examples/basic-examples/pom.xml @@ -11,12 +11,12 @@ com.aliyun.odps odps-sdk-commons - 0.51.0-public.rc0 + 0.51.0-public.rc1 com.aliyun.odps odps-sdk-core - 0.51.0-public.rc0 + 0.51.0-public.rc1 diff --git a/odps-examples/basic-examples/src/main/java/GenerateExpressionSample.java b/odps-examples/basic-examples/src/main/java/GenerateExpressionSample.java index 686cf24a..5a405000 100644 --- a/odps-examples/basic-examples/src/main/java/GenerateExpressionSample.java +++ b/odps-examples/basic-examples/src/main/java/GenerateExpressionSample.java @@ -10,9 +10,6 @@ import com.aliyun.odps.task.SQLTask; import com.google.common.collect.ImmutableMap; -/** - * @author dingxin (zhangdingxin.zdx@alibaba-inc.com) - */ public class GenerateExpressionSample { private static String accessId = ""; @@ -41,12 +38,13 @@ public static void main(String[] args) throws Exception { record.set("a", 123L); record.set("d", Instant.now()); - PartitionSpec partitionSpec = getPartitionSpec(record, schema); + // Auto generate partitionSpec from record and schema + PartitionSpec partitionSpec = schema.generatePartitionSpec(record); System.out.println(partitionSpec); } /** - * Auto generate partitionSpec from record and schema + * How to manually generate partitionSpec from record and schema */ private static PartitionSpec getPartitionSpec(Record record, TableSchema schema) { // And partition spec means where the record will be written to diff --git a/odps-examples/tunnel-examples/pom.xml b/odps-examples/tunnel-examples/pom.xml index bf33c5ac..084d0bee 100644 --- a/odps-examples/tunnel-examples/pom.xml +++ b/odps-examples/tunnel-examples/pom.xml @@ -5,18 +5,30 @@ com.aliyun.odps tunnel-examples 1.0 + + + + org.apache.maven.plugins + maven-compiler-plugin + + 8 + 8 + + + + 4.0.0 com.aliyun.odps odps-sdk-commons - 0.50.1-public + 0.51.0-public.rc1 com.aliyun.odps odps-sdk-core - 0.50.1-public + 0.51.0-public.rc1 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 index 1b7929b7..0ffef716 100644 --- 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 @@ -30,6 +30,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.StringJoiner; import java.util.stream.Collectors; import org.apache.arrow.vector.ipc.ArrowStreamReader; @@ -56,6 +57,7 @@ import com.aliyun.odps.task.SQLTask; import com.aliyun.odps.tunnel.Configuration; import com.aliyun.odps.tunnel.TableTunnel; +import com.aliyun.odps.type.TypeInfo; import com.aliyun.odps.utils.ColumnUtils; import com.aliyun.odps.utils.NameSpaceSchemaUtils; import com.aliyun.odps.utils.OdpsCommonUtils; @@ -2002,6 +2004,9 @@ public boolean isPartitioned() { return !getSchema().getPartitionColumns().isEmpty(); } + private void runSQL(String query) throws OdpsException { + runSQL("AnonymousSQLTask", query); + } private void runSQL(String taskName, String query) throws OdpsException { Map hints = NameSpaceSchemaUtils.setSchemaFlagInHints(null, model.schemaName); Instance i = SQLTask.run(odps, odps.getDefaultProject(), query, taskName, hints, null); @@ -2012,7 +2017,7 @@ private HashMap initParamsWithSchema() throws OdpsException { return NameSpaceSchemaUtils.initParamsWithSchema(model.schemaName); } - private String getCoordinate() throws OdpsException { + private String getCoordinate() { return NameSpaceSchemaUtils.getFullName(model.projectName, model.schemaName, model.name); } @@ -2095,4 +2100,114 @@ public Stream newStream(String streamName) throws OdpsException { TableIdentifier.of(model.projectName, model.schemaName, model.name)); return odps.streams().get(identifier); } + + // update table methods + + /** + * Modify the life cycle of an existing partitioned table or non-partitioned table. + */ + public void setLifeCycle(int days) throws OdpsException { + String sql = String.format("ALTER TABLE %s SET LIFECYCLE %d;", getCoordinate(), days); + runSQL(sql); + } + + /** + * Only the Project Owner or users with the Super_Administrator role can execute commands that modify the table Owner. + */ + public void changeOwner(String newOwner) throws OdpsException { + String target = "table"; + if (isVirtualView() || isMaterializedView()) { + target = "view"; + } + runSQL(String.format("ALTER %s %s CHANGE OWNER TO %s;", target, getCoordinate(), OdpsCommonUtils.quoteStr(newOwner))); + } + + /** + * ChangeComment Modify the comment content of the table. + */ + public void changeComment(String newComment) throws OdpsException { + runSQL(String.format("ALTER TABLE %s SET COMMENT %s;", getCoordinate(), OdpsCommonUtils.quoteStr(newComment))); + } + + /** + * Touch can modify the LastModifiedTime of the table, making LastModifiedTime change to the current time + */ + public void touch() throws OdpsException { + runSQL(String.format("ALTER TABLE %s TOUCH;", getCoordinate())); + } + + /** + * ChangeClusterInfo Modify the cluster information of the table. + */ + public void changeClusterInfo(ClusterInfo clusterInfo) throws OdpsException { + runSQL(String.format("ALTER TABLE %s %s;", getCoordinate(), clusterInfo.toString())); + } + + /** + * Rename the table. + */ + public void rename(String newName) throws Exception { + String target = "table"; + if (isVirtualView()) { + target = "view"; + } + runSQL(String.format("ALTER %s %s RENAME TO %s;", target, getCoordinate(), OdpsCommonUtils.quoteRef(newName))); + model.name = newName; + } + + /** + * Add new columns to the table. + */ + public void addColumns(List columns, boolean ifNotExists) throws Exception { + runSQL(generateAddColumnsSQL(columns, ifNotExists)); + } + + private String generateAddColumnsSQL(List columns, boolean ifNotExists) { + StringBuilder sb = new StringBuilder(); + sb.append("ALTER TABLE ").append(getCoordinate()).append(" ADD COLUMNS "); + if (ifNotExists) { + sb.append("IF NOT EXISTS "); + } + sb.append("("); + StringJoiner joiner = new StringJoiner(", "); + for (Column column : columns) { + StringBuilder columnDef = new StringBuilder(); + columnDef.append(String.format("%s %s", OdpsCommonUtils.quoteRef(column.getName()), column.getTypeInfo().getTypeName())); + if (column.getComment() != null && !column.getComment().isEmpty()) { + columnDef.append(String.format(" COMMENT %s", OdpsCommonUtils.quoteStr(column.getComment()))); + } + joiner.add(columnDef.toString()); + } + sb.append(joiner).append(");"); + return sb.toString(); + } + + /** + * Drop columns from the table. + */ + public void dropColumns(List columnNames) throws Exception { + runSQL(generateDropColumnsSQL(columnNames)); + } + + private String generateDropColumnsSQL(List columnNames) { + StringJoiner joiner = new StringJoiner(", "); + for (String columnName : columnNames) { + joiner.add(OdpsCommonUtils.quoteRef(columnName)); + } + return String.format("ALTER TABLE %s DROP COLUMNS %s;", getCoordinate(), joiner); + } + + /** + * Change the type of an existing column in the table. + */ + public void alterColumnType(String columnName, TypeInfo columnType) throws Exception { + runSQL(String.format("ALTER TABLE %s CHANGE COLUMN %s %s %s;", getCoordinate(), columnName, columnName, columnType.getTypeName())); + } + + /** + * Change the name of an existing column in the table. + */ + public void changeColumnName(String oldColumnName, String newColumnName) throws Exception { + runSQL(String.format("ALTER TABLE %s CHANGE COLUMN %s RENAME TO %s;", getCoordinate(), oldColumnName, newColumnName)); + } } diff --git a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Tables.java b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Tables.java index 1080d561..0cee0948 100644 --- a/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Tables.java +++ b/odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/Tables.java @@ -1087,7 +1087,6 @@ private TableCreator(Odps odps, String projectName, String tableName, TableSchem ExceptionUtils.checkArgumentNotNull("odps", odps); ExceptionUtils.checkStringArgumentNotNull("projectName", projectName); ExceptionUtils.checkStringArgumentNotNull("tableName", tableName); - ExceptionUtils.checkArgumentNotNull("tableSchema", tableSchema); this.odps = odps; this.projectName = projectName; this.tableName = tableName; @@ -1250,7 +1249,7 @@ private String handleSelectStatementCause() { if (lifeCycle != null) { sql.append(" LIFECYCLE ").append(lifeCycle); } - sql.append(" AS ").append(selectStatement).append(";"); + sql.append(" AS (").append(selectStatement).append(");"); return sql.toString(); case VIEW: sql.append("CREATE VIEW "); @@ -1258,21 +1257,23 @@ private String handleSelectStatementCause() { sql.append("IF NOT EXISTS "); } sql.append(NameSpaceSchemaUtils.getFullName(projectName, schemaName, tableName)); - List columns = tableSchema.getColumns(); - sql.append(" ("); - for (int i = 0; i < columns.size(); i++) { - Column column = columns.get(i); - sql.append(OdpsCommonUtils.quoteRef(column.getName())).append(" ") - .append(column.getTypeInfo().getTypeName()); - if (column.getComment() != null) { - sql.append(" COMMENT ").append(OdpsCommonUtils.quoteStr(column.getComment())); - } - if (i + 1 < columns.size()) { - sql.append(','); + if (tableSchema != null && !tableSchema.getColumns().isEmpty()) { + List columns = tableSchema.getColumns(); + sql.append(" ("); + for (int i = 0; i < columns.size(); i++) { + Column column = columns.get(i); + sql.append(OdpsCommonUtils.quoteRef(column.getName())).append(" ") + .append(column.getTypeInfo().getTypeName()); + if (column.getComment() != null) { + sql.append(" COMMENT ").append(OdpsCommonUtils.quoteStr(column.getComment())); + } + if (i + 1 < columns.size()) { + sql.append(','); + } } + sql.append(')'); } - sql.append(')'); - sql.append(" AS ").append(selectStatement).append(";"); + sql.append(" AS (").append(selectStatement).append(");"); return sql.toString(); default: throw new IllegalArgumentException("Only 'Append Table' and 'Virtual View' support 'create table as' statement."); diff --git a/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/TableUpdateTest.java b/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/TableUpdateTest.java new file mode 100644 index 00000000..2084f54d --- /dev/null +++ b/odps-sdk/odps-sdk-core/src/test/java/com/aliyun/odps/TableUpdateTest.java @@ -0,0 +1,128 @@ +package com.aliyun.odps; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.aliyun.odps.commons.transport.OdpsTestUtils; +import com.aliyun.odps.type.TypeInfoFactory; +import com.google.common.collect.ImmutableList; + +/** + * @author dingxin (zhangdingxin.zdx@alibaba-inc.com) + */ +public class TableUpdateTest { + + private Table testTable; + + + @Test + public void testSetLifeCycle() throws Exception { + testTable.setLifeCycle(10); + long life = testTable.getLife(); + Assert.assertEquals(10L, life); + } + + @Test + public void testChangeComment() throws Exception { + testTable.changeComment("test comment"); + Assert.assertEquals("test comment", testTable.getComment()); + } + + @Test + public void testTouch() throws Exception { + testTable.touch(); + } + + @Test + public void testChangeClusterInfo() throws Exception { + Table.ClusterInfo + expectClusterInfo = + new Table.ClusterInfo(Table.ClusterInfo.ClusterType.HASH, ImmutableList.of("c1", "c2"), + ImmutableList.of( + new Table.SortColumn("c1", Table.SortColumn.Order.DESC), + new Table.SortColumn("c2", Table.SortColumn.Order.ASC)), 10); + testTable.changeClusterInfo(expectClusterInfo); + Table.ClusterInfo actualClusterInfo = testTable.getClusterInfo(); + + Assert.assertEquals(expectClusterInfo.toString(), actualClusterInfo.toString()); + } + + @Test + public void testRename() throws Exception { + testTable.rename("new_sdk_TableUpdateTest"); + Assert.assertEquals("new_sdk_TableUpdateTest", testTable.getName()); + testTable.reload(); + + testTable.rename("sdk_TableUpdateTest"); + } + + + @Test + public void testAddColumns() throws Exception { + Column newColumn = Column.newBuilder("c3", TypeInfoFactory.STRING).build(); + Column newColumn2 = Column.newBuilder("c4", TypeInfoFactory.STRING).build(); + testTable.addColumns(ImmutableList.of(newColumn, newColumn2), true); + + TableSchema schema = testTable.getSchema(); + Assert.assertEquals(4, schema.getColumns().size()); + } + + + @Test + public void testDropColumns() throws Exception { + testTable.dropColumns(ImmutableList.of("c2")); + TableSchema schema = testTable.getSchema(); + Assert.assertEquals(1, schema.getColumns().size()); + } + + @Test + public void testAlterColumnType() throws Exception { + testTable.alterColumnType("c2", TypeInfoFactory.STRING); + TableSchema schema = testTable.getSchema(); + Assert.assertEquals(TypeInfoFactory.STRING, schema.getColumn("c2").getTypeInfo()); + + try { + testTable.alterColumnType("c1", TypeInfoFactory.BIGINT); + } catch (OdpsException e) { + System.out.println(e.getMessage()); + // ODPS-0130071:[1,89] Semantic analysis exception - changing column data type from STRING to BIGINT is not supported. + if (e.getMessage().contains("ODPS-0130071")) { + System.out.println("ok"); + return; + } else { + throw e; + } + } + schema = testTable.getSchema(); + Assert.assertEquals(TypeInfoFactory.BIGINT, schema.getColumn("c1").getTypeInfo()); + } + + @Test + public void changeColumnName() throws Exception { + testTable.changeColumnName("c1", "c3"); + TableSchema schema = testTable.getSchema(); + Assert.assertEquals("c3", schema.getColumn("c3").getName()); + } + + @Before + public void initTestTable() throws OdpsException { + Odps odps = OdpsTestUtils.newDefaultOdps(); + TableSchema schema = TableSchema.builder() + .withStringColumn("c1") + .withBigintColumn("c2") + .withPartitionColumn(new Column("p1", TypeInfoFactory.STRING)) + .build(); + + odps.tables().newTableCreator("sdk_TableUpdateTest", schema) + .create(); + testTable = odps.tables().get("sdk_TableUpdateTest"); + } + + @After + public void dropTestTable() throws OdpsException { + Odps odps = OdpsTestUtils.newDefaultOdps(); + odps.tables().delete("sdk_TableUpdateTest", true); + } +}