From 8ec9bdffdf4c0925e5670f611a08983e36fe7061 Mon Sep 17 00:00:00 2001 From: Sundar Annamalai Date: Wed, 10 Mar 2021 10:39:10 -0500 Subject: [PATCH] Fix for star tree cube incorrect results 1. Block INSERT INTO CUBE if source table has been updated. 2. Block duplicate data insert into the cube 3. Insert entire data from source table into cube 4. Updated hetu docs --- .../io/hetu/core/spi/cube/CubeMetadata.java | 70 ++++++-- .../core/spi/cube/CubeMetadataBuilder.java | 6 +- .../cube/aggregator/AggregationSignature.java | 24 +-- .../hetu/core/spi/cube/io/CubeMetaStore.java | 4 +- hetu-docs/en/index.md | 5 + hetu-docs/en/startree/create-cube.md | 57 +++++++ hetu-docs/en/startree/drop-cube.md | 33 ++++ hetu-docs/en/startree/insert-cube.md | 29 ++++ .../en/startree/insert-overwrite-cube.md | 28 ++++ hetu-docs/en/startree/overview.md | 63 ++++++++ hetu-docs/en/startree/show-cubes.md | 35 ++++ .../THIRD_PARTY_OPEN_SOURCE_SOFTWARE_NOTICE | 7 + .../cube/startree/io/StarTreeMetaStore.java | 55 +++---- .../cube/startree/tree/StarTreeMetadata.java | 89 +++++----- .../tree/StarTreeMetadataBuilder.java | 45 +++--- .../startree/io/TestStarTreeMetaStore.java | 27 +++- .../startree/tree/TestStarTreeMetadata.java | 18 ++- pom.xml | 1 + .../prestosql/execution/CreateCubeTask.java | 28 ++-- .../io/prestosql/execution/DropTableTask.java | 2 +- .../operator/CubeFinishOperator.java | 54 +++---- .../io/prestosql/sql/analyzer/Analysis.java | 9 +- .../sql/analyzer/FeaturesConfig.java | 2 +- .../sql/analyzer/SemanticErrorCode.java | 3 + .../sql/analyzer/StatementAnalyzer.java | 146 +++++++++++++---- .../sql/planner/LocalExecutionPlanner.java | 4 +- .../prestosql/sql/planner/LogicalPlanner.java | 23 ++- .../AggregationRewriteWithCube.java | 10 +- .../PruneUnreferencedOutputs.java | 4 +- .../StarTreeAggregationRule.java | 16 +- .../planner/optimizations/SymbolMapper.java | 4 +- .../sql/planner/plan/CubeFinishNode.java | 40 ++--- .../planner/planprinter/IoPlanPrinter.java | 2 +- .../sql/rewrite/InsertCubeRewrite.java | 28 ++-- .../sql/rewrite/ShowQueriesRewrite.java | 16 +- .../TestStarTreeAggregationRule.java | 4 +- .../antlr4/io/prestosql/sql/parser/SqlBase.g4 | 4 +- .../java/io/prestosql/sql/SqlFormatter.java | 44 +++-- .../io/prestosql/sql/parser/AstBuilder.java | 26 +-- .../io/prestosql/sql/tree/CreateCube.java | 24 +-- .../io/prestosql/sql/tree/InsertCube.java | 25 ++- .../prestosql/sql/parser/TestSqlParser.java | 122 ++++++++++---- .../spi/cube/CubeUpdateMetadata.java | 77 +++++++++ .../tests/AbstractTestStarTreeQueries.java | 153 ++++++++++++++++-- .../license/license-header-alternate-2020.txt | 12 ++ src/main/resource/license/license-header.txt | 2 +- 46 files changed, 1090 insertions(+), 390 deletions(-) create mode 100644 hetu-docs/en/startree/create-cube.md create mode 100644 hetu-docs/en/startree/drop-cube.md create mode 100644 hetu-docs/en/startree/insert-cube.md create mode 100644 hetu-docs/en/startree/insert-overwrite-cube.md create mode 100644 hetu-docs/en/startree/overview.md create mode 100644 hetu-docs/en/startree/show-cubes.md create mode 100644 presto-spi/src/main/java/io/prestosql/spi/cube/CubeUpdateMetadata.java create mode 100644 src/main/resource/license/license-header-alternate-2020.txt diff --git a/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadata.java b/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadata.java index fcae9df4c..43e6a8581 100644 --- a/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadata.java +++ b/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadata.java @@ -26,22 +26,55 @@ import java.util.stream.Collectors; public interface CubeMetadata extends Serializable { - String getCubeTableName(); + /** + * Returns name of the cube + */ + String getCubeName(); - String getOriginalTableName(); + /** + * Returns the name of the source table + */ + String getSourceTableName(); - long getLastUpdated(); + /** + * Returns the last updated time of the cube + */ + long getLastUpdatedTime(); + /** + * Returns the last updated time of the source table + */ + long getSourceTableLastUpdatedTime(); + + /** + * Return the names of the dimension columns + */ List getDimensions(); + /** + * Return the names of the aggregation columns + */ List getAggregations(); - List getAggregationsAsString(); - + /** + * Return the group by columns + */ Set getGroup(); + /** + * Checks if metadata matches the CubeStatement + * @param statement cube statement + * @return true - if metadata matches CubeStatement + * false - otherwise + */ boolean matches(CubeStatement statement); + /** + * Filters all metadata that matches the cube statement + * @param metadataList metadata list + * @param statement cube statement + * @return all metadata that is matching the cube statement + */ static List filter(List metadataList, CubeStatement statement) { return metadataList.stream() @@ -49,19 +82,36 @@ public interface CubeMetadata .collect(Collectors.toList()); } + /** + * Retrieves the cube column matching the given aggregation signature + * @return name of the aggregation column if found + */ Optional getColumn(AggregationSignature aggSignature); - Optional getAggregationFunction(String starTableColumn); - - Optional getAggregationColumn(String aggFunction, String originalColumn, boolean distinct); + /** + * Return the aggregation function associated with given cube column + * @return name of the aggregation function + */ + Optional getAggregationFunction(String column); + /** + * Get the aggregation information of the given cube column + * @param column name of the cube column + */ Optional getAggregationSignature(String column); + /** + * Return all aggregation column information + */ List getAggregationSignatures(); + /** + * Return cube predicate string + */ String getPredicateString(); - String getGroupString(); - + /** + * Return the status of the cube + */ CubeStatus getCubeStatus(); } diff --git a/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadataBuilder.java b/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadataBuilder.java index e4c0d1578..20cb2a3f7 100644 --- a/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadataBuilder.java +++ b/hetu-cube/src/main/java/io/hetu/core/spi/cube/CubeMetadataBuilder.java @@ -29,7 +29,9 @@ public interface CubeMetadataBuilder void setCubeStatus(CubeStatus cubeStatus); - CubeMetadata build(); + void setTableLastUpdatedTime(long tableLastUpdatedTime); - CubeMetadata build(long createdTime); + void setCubeLastUpdatedTime(long cubeLastUpdatedTime); + + CubeMetadata build(); } diff --git a/hetu-cube/src/main/java/io/hetu/core/spi/cube/aggregator/AggregationSignature.java b/hetu-cube/src/main/java/io/hetu/core/spi/cube/aggregator/AggregationSignature.java index 4acd9fac7..538165145 100644 --- a/hetu-cube/src/main/java/io/hetu/core/spi/cube/aggregator/AggregationSignature.java +++ b/hetu-cube/src/main/java/io/hetu/core/spi/cube/aggregator/AggregationSignature.java @@ -21,16 +21,16 @@ import com.fasterxml.jackson.annotation.JsonProperty; import java.io.Serializable; import java.util.Objects; +import static io.hetu.core.spi.cube.CubeAggregateFunction.AVG; +import static io.hetu.core.spi.cube.CubeAggregateFunction.COUNT; +import static io.hetu.core.spi.cube.CubeAggregateFunction.MAX; +import static io.hetu.core.spi.cube.CubeAggregateFunction.MIN; +import static io.hetu.core.spi.cube.CubeAggregateFunction.SUM; + public class AggregationSignature implements Serializable, Comparable { - public static final String AVG_FUNCTION_NAME = "avg"; - public static final String COUNT_FUNCTION_NAME = "count"; - public static final String SUM_FUNCTION_NAME = "sum"; - public static final String MIN_FUNCTION_NAME = "min"; - public static final String MAX_FUNCTION_NAME = "max"; - - private static final AggregationSignature COUNT_SIGNATURE = new AggregationSignature(COUNT_FUNCTION_NAME, "*", false); + private static final AggregationSignature COUNT_SIGNATURE = new AggregationSignature(COUNT.getName(), "*", false); private String function; private String dimension; @@ -54,27 +54,27 @@ public class AggregationSignature public static AggregationSignature count(String dimension, boolean distinct) { - return new AggregationSignature(COUNT_FUNCTION_NAME, dimension, distinct); + return new AggregationSignature(COUNT.getName(), dimension, distinct); } public static AggregationSignature sum(String dimension, boolean distinct) { - return new AggregationSignature(SUM_FUNCTION_NAME, dimension, distinct); + return new AggregationSignature(SUM.toString(), dimension, distinct); } public static AggregationSignature avg(String dimension, boolean distinct) { - return new AggregationSignature(AVG_FUNCTION_NAME, dimension, distinct); + return new AggregationSignature(AVG.toString(), dimension, distinct); } public static AggregationSignature min(String dimension, boolean distinct) { - return new AggregationSignature(MIN_FUNCTION_NAME, dimension, distinct); + return new AggregationSignature(MIN.getName(), dimension, distinct); } public static AggregationSignature max(String dimension, boolean distinct) { - return new AggregationSignature(MAX_FUNCTION_NAME, dimension, distinct); + return new AggregationSignature(MAX.getName(), dimension, distinct); } @JsonProperty diff --git a/hetu-cube/src/main/java/io/hetu/core/spi/cube/io/CubeMetaStore.java b/hetu-cube/src/main/java/io/hetu/core/spi/cube/io/CubeMetaStore.java index f5c61e887..92b51cbc3 100644 --- a/hetu-cube/src/main/java/io/hetu/core/spi/cube/io/CubeMetaStore.java +++ b/hetu-cube/src/main/java/io/hetu/core/spi/cube/io/CubeMetaStore.java @@ -36,10 +36,10 @@ public interface CubeMetaStore /** * Create a new Metadata builder * @param cubeName Name of the cube - * @param originalTableName Name of the original table + * @param sourceTableName Name of the table from which cube was created * @return a metadata builder */ - CubeMetadataBuilder getBuilder(String cubeName, String originalTableName); + CubeMetadataBuilder getBuilder(String cubeName, String sourceTableName); /** * Create new metadata builder from the existing metadata diff --git a/hetu-docs/en/index.md b/hetu-docs/en/index.md index 570446e21..3da494ae0 100644 --- a/hetu-docs/en/index.md +++ b/hetu-docs/en/index.md @@ -128,6 +128,7 @@ headless: true - [CALL]({{< relref "./docs/sql/call.md" >}}) - [COMMENT]({{< relref "./docs/sql/comment.md" >}}) - [COMMIT]({{< relref "./docs/sql/commit.md" >}}) + - [CREATE CUBE]({{< relref "./docs/sql/create-cube.md" >}}) - [CREATE ROLE]({{< relref "./docs/sql/create-role.md" >}}) - [CREATE SCHEMA]({{< relref "./docs/sql/create-schema.md" >}}) - [CREATE TABLE]({{< relref "./docs/sql/create-table.md" >}}) @@ -139,6 +140,7 @@ headless: true - [DESCRIBE INPUT]({{< relref "./docs/sql/describe-input.md" >}}) - [DESCRIBE OUTPUT]({{< relref "./docs/sql/describe-output.md" >}}) - [DROP CACHE]({{< relref "./docs/sql/drop-cache.md" >}}) + - [DROP CUBE]({{< relref "./docs/sql/drop-cube.md" >}}) - [DROP ROLE]({{< relref "./docs/sql/drop-role.md" >}}) - [DROP SCHEMA]({{< relref "./docs/sql/drop-schema.md" >}}) - [DROP TABLE]({{< relref "./docs/sql/drop-table.md" >}}) @@ -150,6 +152,8 @@ headless: true - [GRANT ROLES]({{< relref "./docs/sql/grant-roles.md" >}}) - [INSERT]({{< relref "./docs/sql/insert.md" >}}) - [INSERT OVERWRITE]({{< relref "./docs/sql/insert-overwrite.md" >}}) + - [INSERT CUBE]({{< relref "./docs/sql/insert-cube.md" >}}) + - [INSERT OVERWRITE CUBE]({{< relref "./docs/sql/insert-overwrite-cube.md" >}}) - [JMX]({{< relref "./docs/sql/jmx.md" >}}) - [PREPARE]({{< relref "./docs/sql/prepare.md" >}}) - [RESET SESSION]({{< relref "./docs/sql/reset-session.md" >}}) @@ -164,6 +168,7 @@ headless: true - [SHOW COLUMNS]({{< relref "./docs/sql/show-columns.md" >}}) - [SHOW CREATE TABLE]({{< relref "./docs/sql/show-create-table.md" >}}) - [SHOW CREATE VIEW]({{< relref "./docs/sql/show-create-view.md" >}}) + - [SHOW CUBES]({{< relref "./docs/sql/show-cubes.md" >}}) - [SHOW FUNCTIONS]({{< relref "./docs/sql/show-functions.md" >}}) - [SHOW EXTERNAL FUNCTION]({{< relref "./docs/sql/show-external-function.md" >}}) - [SHOW GRANTS]({{< relref "./docs/sql/show-grants.md" >}}) diff --git a/hetu-docs/en/startree/create-cube.md b/hetu-docs/en/startree/create-cube.md new file mode 100644 index 000000000..868d8ab5e --- /dev/null +++ b/hetu-docs/en/startree/create-cube.md @@ -0,0 +1,57 @@ +CREATE CUBE +============ + +Synopsis +-------- + +``` sql +CREATE CUBE [ IF NOT EXISTS ] +cube_name ON table_name WITH ( + AGGREGATIONS = ( expression [, ...] ), GROUP = ( column_name [, ...] ) + [, ( property_name = expression [, ...] ) ] +) +``` + +Description +----------- + +Create a new, empty star-tree cube with the specified group and aggregations. Use `insert-into-cube` to insert data. + +The optional `IF NOT EXISTS` clause causes the error to be suppressed if the table already exists. + +The optional `property_name` section can be used to set properties on the newly created cube. To list all available table properties, run the following query: + + SELECT * FROM system.metadata.table_properties + +Examples +-------- + +Create a new cube `orders_cube` on `orders`: + + CREATE CUBE orders_cube ON orders WITH ( + AGGREGATIONS = ( SUM(totalprice), AVG(totalprice) ), + GROUP = ( orderstatus, orderdate ), + format = 'ORC' + ) + +Create a new partitioned cube `orders_cube`: + + CREATE CUBE orders_cube ON orders WITH ( + AGGREGATIONS = ( SUM(totalprice), AVG(totalprice) ), + GROUP = ( orderstatus, orderdate ), + format = 'ORC', + partitioned_by = ARRAY['orderdate'] + ) + +Limitations +----------- + +- Supported aggregate functions: + COUNT, COUNT DISTINCT, MIN, MAX, SUM, AVG +- Only one group is supported per Cube. +- Different connector might support different data type, and different table/column properties. +- Can currently only create cubes in Hive connector, but the cubes can be created on a table from another connector. + +See Also +-------- +[INSERT INTO CUBE](./insert-cube.md), [SHOW CUBES](./show-cubes.md), [DROP CUBE](./drop-cube.md) \ No newline at end of file diff --git a/hetu-docs/en/startree/drop-cube.md b/hetu-docs/en/startree/drop-cube.md new file mode 100644 index 000000000..5974705a7 --- /dev/null +++ b/hetu-docs/en/startree/drop-cube.md @@ -0,0 +1,33 @@ + +DROP CUBE +========== + +Synopsis +-------- + +``` sql +DROP CUBE [ IF EXISTS ] cube_name +``` + +Description +----------- + +Drop an existing cube. + +The optional `IF EXISTS` clause causes the error to be suppressed if the cube does not exist. + +Examples +-------- + +Drop the cube `orders_cube`: + + DROP CUBE orders_cube + +Drop the cube `orders_cube` if it exists: + + DROP CUBE IF EXISTS orders_cube + +See Also +-------- + +[CREATE CUBE](./create-cube.md), [SHOW CUBES](./show-cubes.md), [INSERT INTO CUBE](./insert-cube.md) diff --git a/hetu-docs/en/startree/insert-cube.md b/hetu-docs/en/startree/insert-cube.md new file mode 100644 index 000000000..428926090 --- /dev/null +++ b/hetu-docs/en/startree/insert-cube.md @@ -0,0 +1,29 @@ +INSERT INTO CUBE +====== + +Synopsis +-------- + +``` sql +INSERT INTO CUBE cube_name [WHERE condition] +``` + +Description +----------- + +Insert data into a star-tree cube. Predicate information is optional. If predicate provided, only data matching +the given predicate are processed from the source table and inserted into the cube. Otherwise, entire +data from the source table is processed and inserted into Cube. + +Examples +-------- + +Insert data based on condition into the `orders_cube` cube: + + INSERT INTO CUBE orders_cube WHERE orderdate > date '1999-01-01'; + INSERT INTO CUBE order_all_cube; + +See Also +-------- + +[INSERT OVERWRITE CUBE](./insert-overwrite-cube.md), [CREATE CUBE](./create-cube.md), [SHOW CUBES](./show-cubes.md), [DROP CUBE](./drop-cube.md) diff --git a/hetu-docs/en/startree/insert-overwrite-cube.md b/hetu-docs/en/startree/insert-overwrite-cube.md new file mode 100644 index 000000000..8c55f3817 --- /dev/null +++ b/hetu-docs/en/startree/insert-overwrite-cube.md @@ -0,0 +1,28 @@ +INSERT INTO CUBE +====== + +Synopsis +-------- + +``` sql +INSERT OVERWRITE CUBE cube_name [WHERE condition] +``` + +Description +----------- + +Similar to INSERT INTO CUBE statement but with this statement the existing data is overwritten. Predicates +are optional. + +Examples +-------- + +Insert data based on condition into the `orders_cube` cube: + + INSERT OVERWRITE CUBE orders_cube WHERE orderdate > date '1999-01-01'; + INSERT OVERWRITE CUBE orders_cube; + +See Also +-------- + +[INSERT INTO CUBE](./insert-cube.md), [CREATE CUBE](./create-cube.md), [SHOW CUBES](./show-cubes.md), [DROP CUBE](./drop-cube.md) diff --git a/hetu-docs/en/startree/overview.md b/hetu-docs/en/startree/overview.md new file mode 100644 index 000000000..8c838c2e6 --- /dev/null +++ b/hetu-docs/en/startree/overview.md @@ -0,0 +1,63 @@ +# Star-Tree + +Star tree cubing is a pre-aggregation technique to achieve low latency runtime for iceberg queries. Star tree cubing aimed to reduce +latency for iceberg queries. An iceberg query computes an aggregate function over an attribute ( or set of attributes) in order to +find aggregate values above a specified threshold. + +## Supported functions + COUNT, COUNT DISTINCT, MIN, MAX, SUM, AVG + +## Enabling and Disabling Star-tree +To enable: +```sql +SET SESSION enable_star_tree_index=true; +``` +To disable: +```sql +SET SESSION enable_star_tree_index=false; +``` + +## Configuration Properties +| Property Name | Default Value | Required| Description| +|---------------------------------------------------|---------------------|---------|--------------| +| optimizer.enable-star-tree-index | false | No | Enables star-tree index| +| cube.metadata-cache-size | 5 | No | The maximum number of metadata for star-trees that could be loaded into cache before eviction happens| +| cube.metadata-cache-ttl | 1h | No | The maximum time to live of star-trees that are be loaded into cache before eviction happens | + +## Examples + +Creating a star-tree cube: +```sql +CREATE CUBE nation_cube +ON nation +WITH (AGGREGATIONS=(count(*), count(distinct regionkey), avg(nationkey), max(regionkey)), +GROUP=(nationkey), +format='orc', partitioned_by=ARRAY['nationkey']); +``` +Next, to add data to the cube: +```sql +INSERT INTO CUBE nation_cube WHERE nationkey > 5; +``` +To use the new cube, just query the original table using aggregations that were included in the cube: +```sql +SELECT count(*) FROM nation WHERE nationkey > 5 GROUP BY nationkey; +SELECT nationkey, avg(nationkey), max(regionkey) WHERE nationkey > 5 GROUP BY nationkey; +``` + +## Optimizer Changes + +The star tree aggregation rule is an Iterative optimizer that optimizes the logical plan by replacing the original aggregation sub-tree +and original table scan with pre-aggregation table scan. + + +## Dependencies + +Star Tree index relies on Hetu metastore to store the cube related metadata. +Please check [Hetu Metastore](../admin/meta-store.md) for more information. + +## Limitation + +1. Star tree cube is only effective when the group by cardinality is considerably lower than the number of rows in + source table. +2. A significant amount of user effort required in maintaining Cubes for large datasets. +3. Only incremental insert into cube is supported. Cannot delete specific rows from Cube. \ No newline at end of file diff --git a/hetu-docs/en/startree/show-cubes.md b/hetu-docs/en/startree/show-cubes.md new file mode 100644 index 000000000..3d9b89c9f --- /dev/null +++ b/hetu-docs/en/startree/show-cubes.md @@ -0,0 +1,35 @@ + +SHOW CUBES +========== + +Synopsis +-------- + +``` sql +SHOW CUBES [ FOR table_name ]; +``` + +Description +----------- + +`SHOW CUBES` lists all cubes. Adding the optional `table_name` lists only the cubes for that table. + +Examples +-------- + +Show all cubes: + +```sql + SHOW CUBES; +``` + +Show cubes for `orders` table: + +```sql + SHOW CUBES FOR orders; +``` + +See Also +-------- + +[CREATE CUBE](./create-cube.md), [DROP CUBE](./drop-cube.md), [INSERT INTO CUBE](./insert-cube.md) diff --git a/hetu-server/THIRD_PARTY_OPEN_SOURCE_SOFTWARE_NOTICE b/hetu-server/THIRD_PARTY_OPEN_SOURCE_SOFTWARE_NOTICE index 5a1a27dd4..82da0aba3 100644 --- a/hetu-server/THIRD_PARTY_OPEN_SOURCE_SOFTWARE_NOTICE +++ b/hetu-server/THIRD_PARTY_OPEN_SOURCE_SOFTWARE_NOTICE @@ -666,3 +666,10 @@ Copyright 2008-2015 the Gson author. License: Apache License V2.0 Please see above. + +Software: Caffeine +Copyright notice: +Copyright 2015 Ben Manes. + +License: Apache License 2.0 +Please see above. \ No newline at end of file diff --git a/hetu-startree/src/main/java/io/hetu/core/cube/startree/io/StarTreeMetaStore.java b/hetu-startree/src/main/java/io/hetu/core/cube/startree/io/StarTreeMetaStore.java index ce6880428..ea0416fe7 100644 --- a/hetu-startree/src/main/java/io/hetu/core/cube/startree/io/StarTreeMetaStore.java +++ b/hetu-startree/src/main/java/io/hetu/core/cube/startree/io/StarTreeMetaStore.java @@ -17,6 +17,7 @@ package io.hetu.core.cube.startree.io; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; +import com.google.common.collect.Sets; import io.hetu.core.cube.startree.tree.AggregateColumn; import io.hetu.core.cube.startree.tree.DimensionColumn; import io.hetu.core.cube.startree.tree.StarTreeColumn; @@ -34,18 +35,14 @@ import io.prestosql.spi.metastore.model.TableEntity; import io.prestosql.spi.metastore.model.TableEntityType; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.concurrent.TimeUnit; import static io.hetu.core.cube.startree.tree.StarTreeMetadata.COLUMN_DELIMITER; -import static io.hetu.core.cube.startree.tree.StarTreeMetadata.GROUP_DELIMITER; import static io.hetu.core.cube.startree.util.Constants.CUBE_CATALOG; import static io.hetu.core.cube.startree.util.Constants.CUBE_DATABASE; import static java.util.Objects.requireNonNull; @@ -53,12 +50,15 @@ import static java.util.Objects.requireNonNull; public class StarTreeMetaStore implements CubeMetaStore { - public static final String ORIGINAL_TABLE_NAME = "originalTableName"; + public static final String SOURCE_TABLE_NAME = "sourceTableName"; public static final String ORIGINAL_COLUMN = "originalColumn"; public static final String STAR_TABLE_NAME = "starTableName"; public static final String GROUPING_STRING = "groupingString"; public static final String PREDICATE_STRING = "predicateString"; public static final String CUBE_STATUS = "cubeStatus"; + public static final String SOURCE_TABLE_LAST_UPDATED_TIME = "sourceLastUpdatedTime"; + public static final String CUBE_LAST_UPDATED_TIME = "cubeLastUpdatedTime"; + private final HetuMetastore metastore; private final LoadingCache> cubeCache; @@ -77,7 +77,7 @@ public class StarTreeMetaStore tableEntities.forEach(table -> { List cols = table.getColumns(); StarTreeMetadataBuilder builder = new StarTreeMetadataBuilder(table.getParameters().get(STAR_TABLE_NAME), - table.getParameters().get(ORIGINAL_TABLE_NAME)); + table.getParameters().get(SOURCE_TABLE_NAME)); cols.forEach(col -> { if (col.getType().equals("aggregate")) { builder.addAggregationColumn(col.getName(), col.getParameters().get("aggregateFunction"), col.getParameters().get(ORIGINAL_COLUMN), Boolean.parseBoolean(col.getParameters().get("distinct"))); @@ -87,21 +87,12 @@ public class StarTreeMetaStore } }); String groupingString = table.getParameters().get(GROUPING_STRING); - if (groupingString != null) { - for (String columns : groupingString.split(GROUP_DELIMITER)) { - Set group; - if (columns.equals("")) { - group = new HashSet<>(); - } - else { - group = new HashSet<>(Arrays.asList(columns.split(COLUMN_DELIMITER))); - } - builder.addGroup(group); - } - } + builder.addGroup(Sets.newHashSet(groupingString.split(COLUMN_DELIMITER))); builder.withPredicate(table.getParameters().get(PREDICATE_STRING)); builder.setCubeStatus(CubeStatus.forValue(Integer.parseInt(table.getParameters().get(CUBE_STATUS)))); - cubeMetadataList.add(builder.build(table.getCreateTime())); + builder.setTableLastUpdatedTime(Long.parseLong(table.getParameters().get(SOURCE_TABLE_LAST_UPDATED_TIME))); + builder.setCubeLastUpdatedTime(Long.parseLong(table.getParameters().get(CUBE_LAST_UPDATED_TIME))); + cubeMetadataList.add(builder.build()); }); return cubeMetadataList; } @@ -119,7 +110,7 @@ public class StarTreeMetaStore List tables = metastore.getAllTables(CUBE_CATALOG, CUBE_DATABASE); List matchingTables = new ArrayList<>(); tables.forEach(table -> { - if (table.getParameters().get(ORIGINAL_TABLE_NAME).equals(tableName)) { + if (table.getParameters().get(SOURCE_TABLE_NAME).equals(tableName)) { matchingTables.add(table); } }); @@ -136,8 +127,8 @@ public class StarTreeMetaStore @Override public void removeCube(CubeMetadata cubeMetadata) { - metastore.dropTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata.getCubeTableName().replace(".", "_")); - cubeCache.invalidate(cubeMetadata.getOriginalTableName()); + metastore.dropTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata.getCubeName().replace(".", "_")); + cubeCache.invalidate(cubeMetadata.getSourceTableName()); } @Override @@ -153,7 +144,7 @@ public class StarTreeMetaStore metastore.createDatabase(databaseEntity); } - String cubeNameDelimited = cubeMetadata.getCubeTableName().replace(".", "_"); + String cubeNameDelimited = cubeMetadata.getCubeName().replace(".", "_"); TableEntity table = getTableEntity((StarTreeMetadata) cubeMetadata); if (metastore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeNameDelimited).isPresent()) { //update flow @@ -163,7 +154,7 @@ public class StarTreeMetaStore //create flow metastore.createTable(table); } - cubeCache.invalidate(cubeMetadata.getOriginalTableName()); + cubeCache.invalidate(cubeMetadata.getSourceTableName()); } private CatalogEntity catalogEntity() @@ -189,7 +180,7 @@ public class StarTreeMetaStore private TableEntity getTableEntity(StarTreeMetadata starTreeMetadata) { - String cubeNameDelimited = starTreeMetadata.getCubeTableName().replace(".", "_"); + String cubeNameDelimited = starTreeMetadata.getCubeName().replace(".", "_"); List columns = new ArrayList<>(); starTreeMetadata.getColumns().forEach(col -> { ColumnEntity newCol = new ColumnEntity(); @@ -214,11 +205,14 @@ public class StarTreeMetaStore columns.add(newCol); }); Map parameters = new HashMap<>(); - parameters.put(ORIGINAL_TABLE_NAME, starTreeMetadata.getOriginalTableName()); - parameters.put(STAR_TABLE_NAME, starTreeMetadata.getCubeTableName()); - parameters.put(GROUPING_STRING, starTreeMetadata.getGroupString()); + parameters.put(SOURCE_TABLE_NAME, starTreeMetadata.getSourceTableName()); + parameters.put(STAR_TABLE_NAME, starTreeMetadata.getCubeName()); + parameters.put(GROUPING_STRING, String.join(COLUMN_DELIMITER, starTreeMetadata.getGroup())); parameters.put(PREDICATE_STRING, starTreeMetadata.getPredicateString()); parameters.put(CUBE_STATUS, String.valueOf(starTreeMetadata.getCubeStatus().getValue())); + parameters.put(CUBE_LAST_UPDATED_TIME, String.valueOf(starTreeMetadata.getLastUpdatedTime())); + parameters.put(SOURCE_TABLE_LAST_UPDATED_TIME, String.valueOf(starTreeMetadata.getSourceTableLastUpdatedTime())); + return TableEntity.builder() .setCatalogName(CUBE_CATALOG) .setDatabaseName(CUBE_DATABASE) @@ -226,7 +220,6 @@ public class StarTreeMetaStore .setTableName(cubeNameDelimited) .setColumns(columns) .setParameters(parameters) - .setCreateTime(starTreeMetadata.getLastUpdated()) .build(); } @@ -237,9 +230,9 @@ public class StarTreeMetaStore } @Override - public CubeMetadataBuilder getBuilder(String cubeName, String originalTableName) + public CubeMetadataBuilder getBuilder(String cubeName, String sourceTableName) { - return new StarTreeMetadataBuilder(cubeName, originalTableName); + return new StarTreeMetadataBuilder(cubeName, sourceTableName); } @Override diff --git a/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadata.java b/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadata.java index 688e8421c..98d68ef3a 100644 --- a/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadata.java +++ b/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadata.java @@ -38,6 +38,7 @@ import java.util.stream.Collectors; import static io.hetu.core.cube.startree.tree.StarTreeColumn.ColumnType.AGGREGATE; import static io.hetu.core.cube.startree.tree.StarTreeColumn.ColumnType.DIMENSION; +import static io.hetu.core.spi.cube.CubeAggregateFunction.AVG; import static java.util.Objects.requireNonNull; public class StarTreeMetadata @@ -45,7 +46,7 @@ public class StarTreeMetadata { private final String starTreeName; - private final String originalTableName; + private final String sourceTableName; private final List columns; @@ -53,47 +54,50 @@ public class StarTreeMetadata private final String predicateString; - private final long lastUpdated; + private final long sourceTableLastUpdatedTime; - private CubeStatus cubeStatus; + private final long lastUpdatedTime; + + private final CubeStatus cubeStatus; public static final String COLUMN_DELIMITER = ","; - public static final String GROUP_DELIMITER = "\t"; @JsonCreator public StarTreeMetadata( @JsonProperty("starTreeName") String starTreeName, - @JsonProperty("originalTableName") String originalTableName, + @JsonProperty("sourceTableName") String sourceTableName, + @JsonProperty("sourceTableLastUpdatedTime") long sourceTableLastUpdatedTime, @JsonProperty("columns") List columns, @JsonProperty("groups") List> groups, @JsonProperty("predicateString") String predicateString, - @JsonProperty("lastUpdated") long lastUpdated, + @JsonProperty("lastUpdatedTime") long lastUpdatedTime, @JsonProperty("cubeStatus") CubeStatus cubeStatus) { this.starTreeName = requireNonNull(starTreeName, "starTreeName is null").toLowerCase(Locale.ENGLISH); - this.originalTableName = requireNonNull(originalTableName, "tableName is null").toLowerCase(Locale.ENGLISH); + this.sourceTableName = requireNonNull(sourceTableName, "tableName is null").toLowerCase(Locale.ENGLISH); this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null")); this.groups = new ArrayList<>(); requireNonNull(groups, "groups is null").forEach(group -> { this.groups.add(new TreeSet<>(group)); }); this.predicateString = predicateString; - this.lastUpdated = lastUpdated; + this.sourceTableLastUpdatedTime = sourceTableLastUpdatedTime; + this.lastUpdatedTime = lastUpdatedTime; this.cubeStatus = cubeStatus; } @JsonProperty @Override - public String getCubeTableName() + public String getCubeName() { return starTreeName; } @JsonProperty @Override - public String getOriginalTableName() + public String getSourceTableName() { - return originalTableName; + return sourceTableName; } @JsonProperty @@ -117,9 +121,9 @@ public class StarTreeMetadata @JsonProperty @Override - public long getLastUpdated() + public long getLastUpdatedTime() { - return lastUpdated; + return lastUpdatedTime; } @JsonIgnore @@ -140,18 +144,10 @@ public class StarTreeMetadata .map(StarTreeColumn::getName).collect(Collectors.toList())); } - @Override - public List getAggregationsAsString() - { - return Collections.unmodifiableList(this.columns.stream() - .filter(column -> AGGREGATE == column.getType()) - .map(StarTreeColumn::getUserFriendlyName).collect(Collectors.toList())); - } - @Override public boolean matches(CubeStatement statement) { - return this.originalTableName.equals(statement.getFrom()) && + return this.sourceTableName.equals(statement.getFrom()) && hasDimensions(statement.getSelection()) && hasGroup(statement.getGroupBy()) && supportAggregations(statement.getAggregations()) && @@ -173,7 +169,7 @@ public class StarTreeMetadata { Collection decomposedAggregations = new ArrayList<>(); aggregations.forEach(aggregationSignature -> { - if (AggregationSignature.AVG_FUNCTION_NAME.equals(aggregationSignature.getFunction())) { + if (AVG.getName().equals(aggregationSignature.getFunction())) { decomposedAggregations.add(AggregationSignature.sum(aggregationSignature.getDimension(), false)); decomposedAggregations.add(AggregationSignature.count(aggregationSignature.getDimension(), false)); } @@ -201,9 +197,11 @@ public class StarTreeMetadata return cubeStatus; } - public void setCubeStatus(CubeStatus cubeStatus) + @JsonProperty + @Override + public long getSourceTableLastUpdatedTime() { - this.cubeStatus = cubeStatus; + return sourceTableLastUpdatedTime; } @JsonIgnore @@ -214,8 +212,7 @@ public class StarTreeMetadata } @JsonIgnore - @Override - public Optional getAggregationColumn(String aggFunction, String originalColumn, boolean distinct) + private Optional getAggregationColumn(String aggFunction, String originalColumn, boolean distinct) { return this.columns.stream() .filter(column -> AGGREGATE == column.getType()) @@ -260,23 +257,6 @@ public class StarTreeMetadata .map(AggregateColumn::getAggregateFunction); } - @Override - public String getGroupString() - { - StringBuilder stringBuilder = new StringBuilder(); - groups.forEach(group -> { - if (!group.isEmpty()) { - group.forEach(column -> { - stringBuilder.append(column).append(COLUMN_DELIMITER); - }); - stringBuilder.deleteCharAt(stringBuilder.length() - 1); - } - stringBuilder.append(GROUP_DELIMITER); - }); - stringBuilder.deleteCharAt(stringBuilder.length() - 1); - return stringBuilder.toString(); - } - @Override public boolean equals(Object o) { @@ -287,16 +267,20 @@ public class StarTreeMetadata return false; } StarTreeMetadata that = (StarTreeMetadata) o; - return lastUpdated == that.lastUpdated && - starTreeName.equals(that.starTreeName) && - originalTableName.equals(that.originalTableName) && - columns.equals(that.columns); + return sourceTableLastUpdatedTime == that.sourceTableLastUpdatedTime + && lastUpdatedTime == that.lastUpdatedTime + && starTreeName.equals(that.starTreeName) + && sourceTableName.equals(that.sourceTableName) + && columns.equals(that.columns) + && groups.equals(that.groups) + && predicateString.equals(that.predicateString) + && cubeStatus == that.cubeStatus; } @Override public int hashCode() { - return Objects.hash(starTreeName, originalTableName, columns, lastUpdated); + return Objects.hash(starTreeName, sourceTableName, columns, groups, predicateString, sourceTableLastUpdatedTime, lastUpdatedTime, cubeStatus); } @Override @@ -304,10 +288,13 @@ public class StarTreeMetadata { return "StarTreeMetadata{" + "starTreeName='" + starTreeName + '\'' + - ", originalTableName='" + originalTableName + '\'' + + ", sourceTableName='" + sourceTableName + '\'' + ", columns=" + columns + + ", groups=" + groups + ", predicateString='" + predicateString + '\'' + - ", lastUpdated=" + lastUpdated + + ", sourceTableLastUpdatedTime=" + sourceTableLastUpdatedTime + + ", lastUpdatedTime=" + lastUpdatedTime + + ", cubeStatus=" + cubeStatus + '}'; } } diff --git a/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadataBuilder.java b/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadataBuilder.java index f7acc5ed9..aeb450b4c 100644 --- a/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadataBuilder.java +++ b/hetu-startree/src/main/java/io/hetu/core/cube/startree/tree/StarTreeMetadataBuilder.java @@ -28,25 +28,30 @@ public class StarTreeMetadataBuilder implements CubeMetadataBuilder { private final String starTableName; - private final String tableName; + private final String sourceTableName; private final List columns = new ArrayList<>(); private final List> groups = new ArrayList<>(); private String predicateString; private CubeStatus cubeStatus; + private long tableLastUpdatedTime; + private long cubeLastUpdatedTime; - public StarTreeMetadataBuilder(String starTableName, String tableName) + public StarTreeMetadataBuilder(String starTableName, String sourceTableName) { this.starTableName = starTableName; - this.tableName = tableName; + this.sourceTableName = sourceTableName; } public StarTreeMetadataBuilder(StarTreeMetadata starTreeMetadata) { - this.starTableName = starTreeMetadata.getCubeTableName(); - this.tableName = starTreeMetadata.getOriginalTableName(); + this.starTableName = starTreeMetadata.getCubeName(); + this.sourceTableName = starTreeMetadata.getSourceTableName(); this.columns.addAll(starTreeMetadata.getColumns()); this.groups.add(starTreeMetadata.getGroup()); this.predicateString = starTreeMetadata.getPredicateString(); + this.tableLastUpdatedTime = starTreeMetadata.getSourceTableLastUpdatedTime(); + this.cubeLastUpdatedTime = starTreeMetadata.getLastUpdatedTime(); + this.cubeStatus = starTreeMetadata.getCubeStatus(); } @Override @@ -55,6 +60,18 @@ public class StarTreeMetadataBuilder this.cubeStatus = cubeStatus; } + @Override + public void setTableLastUpdatedTime(long tableLastUpdatedTime) + { + this.tableLastUpdatedTime = tableLastUpdatedTime; + } + + @Override + public void setCubeLastUpdatedTime(long cubeLastUpdatedTime) + { + this.cubeLastUpdatedTime = cubeLastUpdatedTime; + } + @Override public void addDimensionColumn(String name, String originalColumn) { @@ -84,24 +101,12 @@ public class StarTreeMetadataBuilder { return new StarTreeMetadata( starTableName, - tableName, + sourceTableName, + tableLastUpdatedTime, columns, groups, predicateString, - System.currentTimeMillis(), - cubeStatus); - } - - @Override - public CubeMetadata build(long updatedTime) - { - return new StarTreeMetadata( - starTableName, - tableName, - columns, - groups, - predicateString, - updatedTime, + cubeLastUpdatedTime, cubeStatus); } } diff --git a/hetu-startree/src/test/java/io/hetu/core/cube/startree/io/TestStarTreeMetaStore.java b/hetu-startree/src/test/java/io/hetu/core/cube/startree/io/TestStarTreeMetaStore.java index 11c2c6ad5..4e25176bd 100644 --- a/hetu-startree/src/test/java/io/hetu/core/cube/startree/io/TestStarTreeMetaStore.java +++ b/hetu-startree/src/test/java/io/hetu/core/cube/startree/io/TestStarTreeMetaStore.java @@ -68,6 +68,7 @@ public class TestStarTreeMetaStore cubeMetadataService = new StarTreeProvider().getCubeMetaStore(metaStore, properties); cubeMetadata1 = new StarTreeMetadata("star1", "a", + 1000, ImmutableList.of( new AggregateColumn("sum_cost", "SUM", "cost", false), new DimensionColumn("value", "value")), @@ -77,6 +78,7 @@ public class TestStarTreeMetaStore CubeStatus.READY); cubeMetadata2 = new StarTreeMetadata("star2", "a", + 1000, ImmutableList.of( new AggregateColumn("sum_cost", "SUM", "cost", false), new DimensionColumn("value", "value")), @@ -93,7 +95,7 @@ public class TestStarTreeMetaStore assertFalse(metaStore.getCatalogs().isEmpty()); assertFalse(metaStore.getAllDatabases(CUBE_CATALOG).isEmpty()); - assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeTableName()).isPresent()); + assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeName()).isPresent()); } @Test @@ -103,8 +105,8 @@ public class TestStarTreeMetaStore cubeMetadataService.persist(cubeMetadata2); assertFalse(metaStore.getCatalogs().isEmpty()); assertFalse(metaStore.getAllDatabases(CUBE_CATALOG).isEmpty()); - assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeTableName()).isPresent()); - assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata2.getCubeTableName()).isPresent()); + assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeName()).isPresent()); + assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata2.getCubeName()).isPresent()); } @Test @@ -117,8 +119,8 @@ public class TestStarTreeMetaStore assertFalse(metaStore.getCatalogs().isEmpty()); assertFalse(metaStore.getAllDatabases(CUBE_CATALOG).isEmpty()); - assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeTableName()).isPresent()); - assertFalse(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata2.getCubeTableName()).isPresent()); + assertTrue(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata1.getCubeName()).isPresent()); + assertFalse(metaStore.getTable(CUBE_CATALOG, CUBE_DATABASE, cubeMetadata2.getCubeName()).isPresent()); } @Test @@ -135,7 +137,7 @@ public class TestStarTreeMetaStore assertFalse(cubeMetadataService.getMetadataList("a").isEmpty()); cubeMetadataService.getMetadataList("a").forEach(cube -> { - assertEquals(cube.getOriginalTableName(), "a"); + assertEquals(cube.getSourceTableName(), "a"); assertEquals(cube.getDimensions(), Collections.singleton("value")); }); assertEquals(cubeMetadataService.getMetadataList("a").size(), 2); @@ -152,13 +154,24 @@ public class TestStarTreeMetaStore assertNotEquals(found, cubeMetadata2); } + @Test + public void testGetAllCubes() + { + cubeMetadataService.persist(cubeMetadata1); + cubeMetadataService.persist(cubeMetadata2); + List result = cubeMetadataService.getAllCubes(); + assertEquals(result.size(), 2); + assertTrue(result.containsAll(ImmutableList.of(cubeMetadata1, cubeMetadata2))); + } + @Test public void testUpdateMetadata() { cubeMetadataService.persist(cubeMetadata1); StarTreeMetadata starTreeMetadata = (StarTreeMetadata) cubeMetadata1; StarTreeMetadataBuilder builder = new StarTreeMetadataBuilder(starTreeMetadata); - CubeMetadata updated = builder.build(System.currentTimeMillis()); + builder.setCubeLastUpdatedTime(System.currentTimeMillis()); + CubeMetadata updated = builder.build(); assertNotEquals(cubeMetadata1, updated); } diff --git a/hetu-startree/src/test/java/io/hetu/core/cube/startree/tree/TestStarTreeMetadata.java b/hetu-startree/src/test/java/io/hetu/core/cube/startree/tree/TestStarTreeMetadata.java index 4d7004ab8..41394aa0b 100644 --- a/hetu-startree/src/test/java/io/hetu/core/cube/startree/tree/TestStarTreeMetadata.java +++ b/hetu-startree/src/test/java/io/hetu/core/cube/startree/tree/TestStarTreeMetadata.java @@ -32,6 +32,7 @@ public class TestStarTreeMetadata private final CubeMetadata metadata = new StarTreeMetadata( "memory.default.cube1", "tpch.tiny.lineitem", + 100, ImmutableList.of( new DimensionColumn("suppkey", "suppkey"), new DimensionColumn("returnflag", "returnflag"), @@ -46,18 +47,23 @@ public class TestStarTreeMetadata new AggregateColumn("count_discount", "count", "discount", false)), ImmutableList.of(ImmutableSet.of("returnflag", "linestatus")), null, - 1000, CubeStatus.READY); + 1000, + CubeStatus.READY); @Test public void testCubeMetadata() { - assertEquals(metadata.getCubeTableName(), "memory.default.cube1", "incorrect name"); - assertEquals(metadata.getOriginalTableName(), "tpch.tiny.lineitem", "incorrect table name"); - assertEquals(metadata.getLastUpdated(), 1000, "incorrect updating time"); + assertEquals(metadata.getCubeName(), "memory.default.cube1", "incorrect name"); + assertEquals(metadata.getSourceTableName(), "tpch.tiny.lineitem", "incorrect table name"); + assertEquals(metadata.getLastUpdatedTime(), 1000, "incorrect updating time"); + assertEquals(metadata.getAggregations(), ImmutableList.of("sum_quantity", + "sum_extendedprice", "count_quantity", "count_extendprice", "count_discount")); + assertEquals(metadata.getGroup(), ImmutableSet.of("returnflag", "linestatus")); + assertEquals(metadata.getDimensions(), ImmutableList.of("suppkey", "returnflag", "linestatus", "shipdate", "discount", "quantity")); } @Test - public void testMatchingValidStatement() + public void testMetadataMatchesCubeStatement() { CubeStatement statement = CubeStatement.newBuilder() .select("returnflag", "linestatus") @@ -69,7 +75,7 @@ public class TestStarTreeMetadata } @Test - public void testNotMatchingValidStatement1() + public void testMetadataNotMatchesCubeStatement() { CubeStatement statement = CubeStatement.newBuilder() .select("returnflag", "linestatus") diff --git a/pom.xml b/pom.xml index e846a9dd4..be8217c4a 100644 --- a/pom.xml +++ b/pom.xml @@ -1622,6 +1622,7 @@ ${air.main.basedir}/src/main/resource/license/license-header.txt ${air.main.basedir}/src/main/resource/license/license-header-alternate-2010.txt ${air.main.basedir}/src/main/resource/license/license-header-alternate-2012.txt + ${air.main.basedir}/src/main/resource/license/license-header-alternate-2020.txt diff --git a/presto-main/src/main/java/io/prestosql/execution/CreateCubeTask.java b/presto-main/src/main/java/io/prestosql/execution/CreateCubeTask.java index 6fd1da023..94b2f2b65 100644 --- a/presto-main/src/main/java/io/prestosql/execution/CreateCubeTask.java +++ b/presto-main/src/main/java/io/prestosql/execution/CreateCubeTask.java @@ -17,6 +17,7 @@ package io.prestosql.execution; import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.ListenableFuture; +import io.hetu.core.spi.cube.CubeAggregateFunction; import io.hetu.core.spi.cube.CubeMetadataBuilder; import io.hetu.core.spi.cube.CubeStatus; import io.hetu.core.spi.cube.aggregator.AggregationSignature; @@ -108,7 +109,7 @@ public class CreateCubeTask throw new RuntimeException("HetuMetaStore is not initialized"); } QualifiedObjectName cubeName = createQualifiedObjectName(session, statement, statement.getCubeName()); - QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getTableName()); + QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getSourceTableName()); Optional cubeHandle = metadata.getTableHandle(session, cubeName); Optional tableHandle = metadata.getTableHandle(session, tableName); @@ -134,12 +135,12 @@ public class CreateCubeTask } if (!tableHandle.isPresent()) { - throw new SemanticException(MISSING_TABLE, statement, "Table %s does not exist", cubeName); + throw new SemanticException(MISSING_TABLE, statement, "Table '%s' does not exist", tableName); } TableMetadata tableMetadata = metadata.getTableMetadata(session, tableHandle.get()); List groupingSet = statement.getGroupingSet().stream().map(Identifier::getValue).collect(Collectors.toList()); - Map originalTableColumns = tableMetadata.getColumns().stream().collect(Collectors.toMap(ColumnMetadata::getName, col -> col)); + Map sourceTableColumns = tableMetadata.getColumns().stream().collect(Collectors.toMap(ColumnMetadata::getName, col -> col)); List cubeColumns = new ArrayList<>(); Map aggregations = new HashMap<>(); Analysis analysis = analyzeStatement(statement, session, metadata, accessControl, parameters, stateMachine.getWarningCollector()); @@ -150,21 +151,22 @@ public class CreateCubeTask String argument = aggFunction.getArguments().isEmpty() || aggFunction.getArguments().get(0) instanceof LongLiteral ? null : ((Identifier) aggFunction.getArguments().get(0)).getValue(); boolean distinct = aggFunction.isDistinct(); String cubeColumnName = aggFunctionName + "_" + (argument == null ? "all" : argument) + (aggFunction.isDistinct() ? "_distinct" : ""); - switch (aggFunctionName) { - case AggregationSignature.SUM_FUNCTION_NAME: + CubeAggregateFunction cubeAggregateFunction = CubeAggregateFunction.valueOf(aggFunctionName.toUpperCase(ENGLISH)); + switch (cubeAggregateFunction) { + case SUM: aggregations.put(cubeColumnName, AggregationSignature.sum(argument, distinct)); break; - case AggregationSignature.COUNT_FUNCTION_NAME: + case COUNT: AggregationSignature aggregationSignature = argument == null ? AggregationSignature.count() : AggregationSignature.count(argument, distinct); aggregations.put(cubeColumnName, aggregationSignature); break; - case AggregationSignature.AVG_FUNCTION_NAME: + case AVG: aggregations.put(cubeColumnName, AggregationSignature.avg(argument, distinct)); break; - case AggregationSignature.MAX_FUNCTION_NAME: + case MAX: aggregations.put(cubeColumnName, AggregationSignature.max(argument, distinct)); break; - case AggregationSignature.MIN_FUNCTION_NAME: + case MIN: aggregations.put(cubeColumnName, AggregationSignature.min(argument, distinct)); break; default: @@ -195,15 +197,16 @@ public class CreateCubeTask if (properties.containsKey("partitioned_by")) { List partitionCols = new ArrayList<>(((List) properties.get("partitioned_by"))); + // put all partition columns at the end of the list groupingSet.removeAll(partitionCols); groupingSet.addAll(partitionCols); } for (String dimension : groupingSet) { - if (!originalTableColumns.containsKey(dimension)) { + if (!sourceTableColumns.containsKey(dimension)) { throw new SemanticException(MISSING_COLUMN, statement, "Column %s does not exist", dimension); } - ColumnMetadata tableCol = originalTableColumns.get(dimension); + ColumnMetadata tableCol = sourceTableColumns.get(dimension); ColumnMetadata cubeCol = new ColumnMetadata( dimension, tableCol.getType(), @@ -230,7 +233,10 @@ public class CreateCubeTask groupingSet.forEach(dimension -> builder.addDimensionColumn(dimension, dimension)); aggregations.forEach((column, aggregationSignature) -> builder.addAggregationColumn(column, aggregationSignature.getFunction(), aggregationSignature.getDimension(), aggregationSignature.isDistinct())); builder.addGroup(new HashSet<>(groupingSet)); + //Status and Table modified time will be updated on the first insert into the cube builder.setCubeStatus(CubeStatus.INACTIVE); + builder.setTableLastUpdatedTime(-1L); + builder.setCubeLastUpdatedTime(System.currentTimeMillis()); optionalCubeMetaStore.get().persist(builder.build()); return immediateFuture(null); diff --git a/presto-main/src/main/java/io/prestosql/execution/DropTableTask.java b/presto-main/src/main/java/io/prestosql/execution/DropTableTask.java index 63a2f8ee8..c80f3e992 100644 --- a/presto-main/src/main/java/io/prestosql/execution/DropTableTask.java +++ b/presto-main/src/main/java/io/prestosql/execution/DropTableTask.java @@ -93,7 +93,7 @@ public class DropTableTask if (optionalCubeMetaStore.isPresent()) { List cubes = optionalCubeMetaStore.get().getMetadataList(tableName.toString()); for (CubeMetadata cube : cubes) { - String[] parts = cube.getCubeTableName().split("\\."); + String[] parts = cube.getCubeName().split("\\."); Optional cubeHandle = metadata.getTableHandle(session, createQualifiedObjectName(session, null, QualifiedName.of(parts[0], parts[1], parts[2]))); cubeHandle.ifPresent(cubeTable -> metadata.dropTable(session, cubeTable)); optionalCubeMetaStore.get().removeCube(cube); diff --git a/presto-main/src/main/java/io/prestosql/operator/CubeFinishOperator.java b/presto-main/src/main/java/io/prestosql/operator/CubeFinishOperator.java index 8079e8195..97f871935 100644 --- a/presto-main/src/main/java/io/prestosql/operator/CubeFinishOperator.java +++ b/presto-main/src/main/java/io/prestosql/operator/CubeFinishOperator.java @@ -19,6 +19,7 @@ import io.hetu.core.spi.cube.io.CubeMetaStore; import io.prestosql.Session; import io.prestosql.cube.CubeManager; import io.prestosql.spi.Page; +import io.prestosql.spi.cube.CubeUpdateMetadata; import io.prestosql.spi.plan.PlanNodeId; import io.prestosql.spi.snapshot.RestorableConfig; import io.prestosql.sql.ExpressionFormatter; @@ -46,9 +47,7 @@ public class CubeFinishOperator private final PlanNodeId planNodeId; private final Session session; private final CubeManager cubeManager; - private final String cubeName; - private final Expression newDataPredicate; - private final boolean overwrite; + private final CubeUpdateMetadata metadata; private boolean closed; public CubeFinishOperatorFactory( @@ -56,17 +55,13 @@ public class CubeFinishOperator PlanNodeId planNodeId, Session session, CubeManager cubeManager, - String cubeName, - Expression newDataPredicate, - boolean overwrite) + CubeUpdateMetadata metadata) { this.operatorId = operatorId; this.planNodeId = requireNonNull(planNodeId, "planNodeId is null"); this.session = requireNonNull(session, "session is null"); this.cubeManager = requireNonNull(cubeManager, "cubeManager is null"); - this.cubeName = requireNonNull(cubeName, "starTableName is null"); - this.newDataPredicate = newDataPredicate; - this.overwrite = overwrite; + this.metadata = requireNonNull(metadata, "metadata is null"); } @Override @@ -74,7 +69,7 @@ public class CubeFinishOperator { checkState(!closed, "Factory is already closed"); OperatorContext context = driverContext.addOperatorContext(operatorId, planNodeId, CubeFinishOperator.class.getSimpleName()); - return new CubeFinishOperator(context, cubeManager, cubeName, newDataPredicate, overwrite); + return new CubeFinishOperator(context, cubeManager, metadata); } @Override @@ -86,7 +81,7 @@ public class CubeFinishOperator @Override public OperatorFactory duplicate() { - return new CubeFinishOperatorFactory(operatorId, planNodeId, session, cubeManager, cubeName, newDataPredicate, overwrite); + return new CubeFinishOperatorFactory(operatorId, planNodeId, session, cubeManager, metadata); } } @@ -99,24 +94,18 @@ public class CubeFinishOperator private final OperatorContext operatorContext; private final CubeMetaStore cubeMetastore; - private final String cubeName; - private final Expression newDataPredicate; - private final boolean overwrite; + private final CubeUpdateMetadata updateMetadata; private State state = State.NEEDS_INPUT; private Page page; public CubeFinishOperator( OperatorContext operatorContext, CubeManager cubeManager, - String cubeName, - Expression newDataPredicate, - boolean overwrite) + CubeUpdateMetadata updateMetadata) { this.operatorContext = requireNonNull(operatorContext, "operatorContext is null"); this.cubeMetastore = cubeManager.getMetaStore(STAR_TREE).get(); - this.cubeName = cubeName; - this.newDataPredicate = newDataPredicate; - this.overwrite = overwrite; + this.updateMetadata = updateMetadata; } @Override @@ -146,21 +135,26 @@ public class CubeFinishOperator if (state != State.HAS_OUTPUT) { return null; } - CubeMetadata cubeMetadata = cubeMetastore.getMetadataFromCubeName(cubeName).get(); + CubeMetadata cubeMetadata = cubeMetastore.getMetadataFromCubeName(updateMetadata.getCubeName()).get(); CubeMetadataBuilder builder = cubeMetastore.getBuilder(cubeMetadata); - Expression updatable; - if (overwrite || cubeMetadata.getPredicateString() == null) { - updatable = newDataPredicate; + if (updateMetadata.getDataPredicateString() == null) { + //Ensure that existing predicate metadata is reset. + builder.withPredicate(null); } else { - Expression existing = new SqlParser().createExpression(cubeMetadata.getPredicateString(), new ParsingOptions()); - updatable = ExpressionUtils.or(existing, newDataPredicate); + Expression updatable = new SqlParser().createExpression(updateMetadata.getDataPredicateString(), new ParsingOptions()); + //Merge new data predicate with existing predicate string + if (!updateMetadata.isOverwrite() && cubeMetadata.getPredicateString() != null) { + Expression existing = new SqlParser().createExpression(cubeMetadata.getPredicateString(), new ParsingOptions()); + updatable = ExpressionUtils.or(existing, updatable); + } + //TODO: Add Logic to simplify expression. Check if Two between predicates can be merged into one + builder.withPredicate(ExpressionFormatter.formatExpression(updatable, Optional.empty())); } - //TODO: Add Logic to simplify expression. Check if Two between predicates can be merged into one - builder.withPredicate(ExpressionFormatter.formatExpression(updatable, Optional.empty())); + builder.setTableLastUpdatedTime(updateMetadata.getTableLastUpdatedTime()); + builder.setCubeLastUpdatedTime(System.currentTimeMillis()); builder.setCubeStatus(READY); - CubeMetadata update = builder.build(System.currentTimeMillis()); - cubeMetastore.persist(update); + cubeMetastore.persist(builder.build()); state = State.FINISHED; return page; } diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/Analysis.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/Analysis.java index 1d36ed351..2a259d5dc 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/Analysis.java +++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/Analysis.java @@ -875,11 +875,13 @@ public class Analysis public static final class CubeInsert { private final TableHandle target; + private final TableHandle sourceTable; private final List columns; - public CubeInsert(TableHandle target, List columns) + public CubeInsert(TableHandle target, TableHandle sourceTable, List columns) { this.target = requireNonNull(target, "target is null"); + this.sourceTable = requireNonNull(sourceTable, "sourceTable is null"); this.columns = requireNonNull(columns, "columns is null"); checkArgument(columns.size() > 0, "No columns given to insert"); } @@ -893,6 +895,11 @@ public class Analysis { return target; } + + public TableHandle getSourceTable() + { + return sourceTable; + } } @Immutable diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java index a631c03a2..e1ad68dd3 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java +++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java @@ -1341,7 +1341,7 @@ public class FeaturesConfig } @Config("cube.metadata-cache-ttl") - @ConfigDescription("The maximum time to live that are be loaded into cache before eviction happens") + @ConfigDescription("The maximum time to live for cube metadata that were loaded into cache before eviction happens") public FeaturesConfig setCubeMetadataCacheTtl(Duration cubeMetadataCacheTtl) { this.cubeMetadataCacheTtl = cubeMetadataCacheTtl; diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/SemanticErrorCode.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/SemanticErrorCode.java index 9a36fee88..2b53ae3f3 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/SemanticErrorCode.java +++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/SemanticErrorCode.java @@ -118,4 +118,7 @@ public enum SemanticErrorCode INVALID_FETCH_FIRST_ROW_COUNT, INVALID_LIMIT_ROW_COUNT, MISSING_ORDER_BY, + + TABLE_STATE_INCORRECT, + PREDICATE_OVERLAP } diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java index 0f041fb17..227914ad3 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java +++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java @@ -21,6 +21,8 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import com.google.common.collect.Multimap; import io.hetu.core.spi.cube.CubeAggregateFunction; +import io.hetu.core.spi.cube.CubeMetadata; +import io.hetu.core.spi.cube.CubeStatus; import io.hetu.core.spi.cube.io.CubeMetaStore; import io.prestosql.Session; import io.prestosql.SystemSessionProperties; @@ -50,6 +52,7 @@ import io.prestosql.spi.function.OperatorType; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.Pair; import io.prestosql.spi.metadata.TableHandle; +import io.prestosql.spi.plan.Symbol; import io.prestosql.spi.security.AccessDeniedException; import io.prestosql.spi.security.Identity; import io.prestosql.spi.security.ViewExpression; @@ -62,9 +65,12 @@ import io.prestosql.spi.type.Type; import io.prestosql.spi.type.TypeNotFoundException; import io.prestosql.spi.type.TypeSignature; import io.prestosql.spi.type.VarcharType; +import io.prestosql.sql.ExpressionFormatter; +import io.prestosql.sql.ExpressionUtils; import io.prestosql.sql.SqlPath; import io.prestosql.sql.parser.ParsingException; import io.prestosql.sql.parser.SqlParser; +import io.prestosql.sql.planner.ExpressionDomainTranslator; import io.prestosql.sql.planner.ExpressionInterpreter; import io.prestosql.sql.planner.SymbolsExtractor; import io.prestosql.sql.planner.TypeProvider; @@ -73,6 +79,7 @@ import io.prestosql.sql.tree.AliasedRelation; import io.prestosql.sql.tree.AllColumns; import io.prestosql.sql.tree.Analyze; import io.prestosql.sql.tree.AssignmentItem; +import io.prestosql.sql.tree.BooleanLiteral; import io.prestosql.sql.tree.Call; import io.prestosql.sql.tree.Comment; import io.prestosql.sql.tree.Commit; @@ -182,6 +189,7 @@ import java.util.Optional; import java.util.OptionalLong; import java.util.Properties; import java.util.Set; +import java.util.function.LongSupplier; import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkArgument; @@ -245,6 +253,7 @@ import static io.prestosql.sql.analyzer.SemanticErrorCode.MISMATCHED_SET_COLUMN_ import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_ATTRIBUTE; import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_CATALOG; import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_COLUMN; +import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_CUBE; import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_ORDER_BY; import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_SCHEMA; import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_TABLE; @@ -254,7 +263,9 @@ import static io.prestosql.sql.analyzer.SemanticErrorCode.NONDETERMINISTIC_ORDER import static io.prestosql.sql.analyzer.SemanticErrorCode.NON_NUMERIC_SAMPLE_PERCENTAGE; import static io.prestosql.sql.analyzer.SemanticErrorCode.NOT_SUPPORTED; import static io.prestosql.sql.analyzer.SemanticErrorCode.ORDER_BY_MUST_BE_IN_SELECT; +import static io.prestosql.sql.analyzer.SemanticErrorCode.PREDICATE_OVERLAP; import static io.prestosql.sql.analyzer.SemanticErrorCode.TABLE_ALREADY_EXISTS; +import static io.prestosql.sql.analyzer.SemanticErrorCode.TABLE_STATE_INCORRECT; import static io.prestosql.sql.analyzer.SemanticErrorCode.TOO_MANY_ARGUMENTS; import static io.prestosql.sql.analyzer.SemanticErrorCode.TOO_MANY_GROUPING_SETS; import static io.prestosql.sql.analyzer.SemanticErrorCode.TYPE_MISMATCH; @@ -446,38 +457,6 @@ class StatementAnalyzer return createAndAssignScope(insert, scope, Field.newUnqualified("rows", BIGINT)); } - @Override - protected Scope visitInsertCube(InsertCube insertCube, Optional scope) - { - QualifiedObjectName targetCube = createQualifiedObjectName(session, insertCube, insertCube.getCubeName()); // check if target is view - if (metadata.getView(session, targetCube).isPresent()) { - throw new SemanticException(NOT_SUPPORTED, insertCube, "Inserting into view is not supported"); - } // check if target cube is present - Optional optionalCubeMetaStore = cubeManager.getMetaStore(STAR_TREE); - if (!optionalCubeMetaStore.isPresent() || !optionalCubeMetaStore.get().getMetadataFromCubeName(targetCube.toString()).isPresent()) { - throw new SemanticException(INSERT_INTO_CUBE, insertCube, "%s is not a star-tree cube, INSERT INTO CUBE is not applicable.", targetCube); - } // check if target cube is present as a table - Optional targetCubeHandle = metadata.getTableHandle(session, targetCube); - if (!targetCubeHandle.isPresent()) { - throw new SemanticException(MISSING_TABLE, insertCube, "Table '%s' does not exist", targetCube); - } // analyze the query that creates the data - Scope queryScope = process(insertCube.getQuery(), scope); - accessControl.checkCanInsertIntoTable(session.getRequiredTransactionId(), session.getIdentity(), targetCube); - if (insertCube.isOverwrite()) { - // set the insert as insert overwrite - analysis.setUpdateType("INSERT OVERWRITE CUBE", targetCube); - analysis.setCubeOverwrite(true); - } - else { - analysis.setUpdateType("INSERT CUBE", targetCube); - } - Map columnHandles = metadata.getColumnHandles(session, targetCubeHandle.get()); - analysis.setCubeInsert(new Analysis.CubeInsert( - targetCubeHandle.get(), - insertCube.getColumns().stream().map(Identifier::getValue).map(columnHandles::get).collect(Collectors.toList()))); - return createAndAssignScope(insertCube, scope, Field.newUnqualified("rows", BIGINT)); - } - private boolean typesMatchForInsert(Iterable tableTypes, Iterable queryTypes) { if (Iterables.size(tableTypes) != Iterables.size(queryTypes)) { @@ -530,6 +509,107 @@ class StatementAnalyzer return type instanceof CharType || (type instanceof VarcharType && !((VarcharType) type).isUnbounded()) || hasNestedBoundedCharacterType(type); } + @Override + protected Scope visitInsertCube(InsertCube insertCube, Optional scope) + { + QualifiedObjectName targetCube = createQualifiedObjectName(session, insertCube, insertCube.getCubeName()); + CubeMetaStore cubeMetaStore = cubeManager.getMetaStore(STAR_TREE).orElseThrow(() -> new RuntimeException("Hetu metastore must be initialized")); + CubeMetadata cubeMetadata = cubeMetaStore.getMetadataFromCubeName(targetCube.toString()) + .orElseThrow(() -> new SemanticException(INSERT_INTO_CUBE, insertCube, "Cube '%s' is not found, INSERT INTO CUBE is not applicable.", targetCube)); + Optional targetCubeHandle = metadata.getTableHandle(session, targetCube); + if (!targetCubeHandle.isPresent()) { + throw new SemanticException(MISSING_CUBE, insertCube, "Cube '%s' table handle does not exist", targetCube); + } + + QualifiedObjectName tableName = QualifiedObjectName.valueOf(cubeMetadata.getSourceTableName()); + TableHandle sourceTableHandle = metadata.getTableHandle(session, tableName) + .orElseThrow(() -> new SemanticException(MISSING_TABLE, insertCube, "Source table '%s' on which cube was built is missing", tableName.toString())); + + //Cube status is determined based on the last modified timestamp of the source table + //Without that Cube might return incorrect results if the table was updated but cube was not. + LongSupplier tableLastModifiedTime = metadata.getTableLastModifiedTimeSupplier(session, sourceTableHandle); + if (tableLastModifiedTime == null) { + throw new SemanticException(TABLE_STATE_INCORRECT, insertCube, "Cannot allow insert into cube. Cube might return incorrect results. Unable to identify last modified of the time source table."); + } + // If Original table was updated since Cube was built then We cannot allow any more updates on the Cube. + // User must create new cube from the source table and try insert overwrite cube + if (!insertCube.isOverwrite() && cubeMetadata.getCubeStatus() == CubeStatus.READY && tableLastModifiedTime.getAsLong() > cubeMetadata.getSourceTableLastUpdatedTime()) { + throw new SemanticException(TABLE_STATE_INCORRECT, insertCube, "Cannot insert into cube. Source table has been updated since Cube was last updated. Try INSERT OVERWRITE CUBE or Create new a cube"); + } + + Scope queryScope = process(insertCube.getQuery(), scope); + accessControl.checkCanInsertIntoTable(session.getRequiredTransactionId(), session.getIdentity(), targetCube); + if (insertCube.isOverwrite()) { + // set the insert as insert overwrite + analysis.setUpdateType("INSERT OVERWRITE CUBE", targetCube); + analysis.setCubeOverwrite(true); + } + else { + analysis.setUpdateType("INSERT CUBE", targetCube); + } + if (!insertCube.isOverwrite() && !insertCube.getWhere().isPresent() && cubeMetadata.getCubeStatus() != CubeStatus.INACTIVE) { + //Means data some data was inserted before, but trying to insert entire dataset + throw new SemanticException(PREDICATE_OVERLAP, insertCube, "Cannot allow insert. Inserting entire dataset but cube already has partial data"); + } + else if (!insertCube.isOverwrite() && insertCube.getWhere().isPresent() && arePredicatesOverlapping(insertCube.getWhere().get(), cubeMetadata)) { + throw new SemanticException(PREDICATE_OVERLAP, insertCube, "Cannot allow insert. Cube already contains data for the given predicate '%s'", ExpressionFormatter.formatExpression(insertCube.getWhere().get(), Optional.empty())); + } + Map columnHandles = metadata.getColumnHandles(session, targetCubeHandle.get()); + analysis.setCubeInsert(new Analysis.CubeInsert( + targetCubeHandle.get(), + sourceTableHandle, + insertCube.getColumns().stream().map(Identifier::getValue).map(columnHandles::get).collect(Collectors.toList()))); + return createAndAssignScope(insertCube, scope, Field.newUnqualified("rows", BIGINT)); + } + + private boolean arePredicatesOverlapping(Expression newDataPredicate, CubeMetadata cubeMetadata) + { + ImmutableMap.Builder typesBuilder = ImmutableMap.builder(); + new SymbolTypeBuilderVisitor(analysis.getTypes()).process(newDataPredicate, typesBuilder); + TypeProvider types = TypeProvider.viewOf(typesBuilder.build()); + + newDataPredicate = ExpressionUtils.rewriteIdentifiersToSymbolReferences(newDataPredicate); + ExpressionDomainTranslator.ExtractionResult decomposedNewDataPredicate = ExpressionDomainTranslator.fromPredicate(metadata, session, newDataPredicate, types); + if (!BooleanLiteral.TRUE_LITERAL.equals(decomposedNewDataPredicate.getRemainingExpression())) { + throw new RuntimeException(String.format("Cannot support predicate '%s'", ExpressionFormatter.formatExpression(newDataPredicate, Optional.empty()))); + } + if (cubeMetadata.getCubeStatus() == CubeStatus.INACTIVE) { + //Inactive cubes are empty. So inserts should be allowed. + return false; + } + + if (cubeMetadata.getPredicateString() == null) { + //Means Cube was created for entire dataset. + return true; + } + SqlParser sqlParser = new SqlParser(); + Expression cubePredicateAsExpr = sqlParser.createExpression(cubeMetadata.getPredicateString(), createParsingOptions(session)); + cubePredicateAsExpr = ExpressionUtils.rewriteIdentifiersToSymbolReferences(cubePredicateAsExpr); + ExpressionDomainTranslator.ExtractionResult decomposedCubePredicate = ExpressionDomainTranslator.fromPredicate(metadata, session, cubePredicateAsExpr, types); + return decomposedCubePredicate.getTupleDomain().overlaps(decomposedNewDataPredicate.getTupleDomain()); + } + + private class SymbolTypeBuilderVisitor + extends DefaultTraversalVisitor> + { + private final Map, Type> types; + + private SymbolTypeBuilderVisitor(Map, Type> types) + { + this.types = requireNonNull(types, "types is null"); + } + + @Override + protected Void visitIdentifier(Identifier identifier, ImmutableMap.Builder builder) + { + NodeRef expressionRef = NodeRef.of(identifier); + if (types.containsKey(expressionRef)) { + builder.put(new Symbol(identifier.getValue()), types.get(expressionRef)); + } + return null; + } + } + @Override protected Scope visitDelete(Delete node, Optional scope) { @@ -717,7 +797,7 @@ class StatementAnalyzer Set cubeSupportedFunctions = CubeAggregateFunction.SUPPORTED_FUNCTIONS; Set aggFunctions = node.getAggregations(); - Scope queryScope = process(new Table(node.getTableName()), scope); + Scope queryScope = process(new Table(node.getSourceTableName()), scope); ImmutableList.Builder outputFields = ImmutableList.builder(); for (FunctionCall aggFunction : aggFunctions) { String argument = aggFunction.getArguments().isEmpty() || aggFunction.getArguments().get(0) instanceof LongLiteral ? null : ((Identifier) aggFunction.getArguments().get(0)).getValue(); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java index 1231aff99..42dbfb937 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java @@ -2907,9 +2907,7 @@ public class LocalExecutionPlanner node.getId(), session, cubeManager, - node.getCubeName(), - node.getDataPredicate(), - node.isOverwrite()); + node.getMetadata()); Map layout = ImmutableMap.of(node.getOutputSymbols().get(0), 0); return new PhysicalOperation(operatorFactory, layout, context, source); } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LogicalPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/LogicalPlanner.java index 71dacca13..e896d4791 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LogicalPlanner.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/LogicalPlanner.java @@ -34,6 +34,7 @@ import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.connector.ColumnMetadata; import io.prestosql.spi.connector.ConnectorTableMetadata; import io.prestosql.spi.connector.QualifiedObjectName; +import io.prestosql.spi.cube.CubeUpdateMetadata; import io.prestosql.spi.function.Signature; import io.prestosql.spi.metadata.TableHandle; import io.prestosql.spi.operator.ReuseExchangeOperator; @@ -51,6 +52,7 @@ import io.prestosql.spi.statistics.TableStatisticsMetadata; import io.prestosql.spi.type.CharType; import io.prestosql.spi.type.Type; import io.prestosql.spi.type.VarcharType; +import io.prestosql.sql.ExpressionFormatter; import io.prestosql.sql.analyzer.Analysis; import io.prestosql.sql.analyzer.Field; import io.prestosql.sql.analyzer.RelationId; @@ -107,6 +109,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Optional; import java.util.UUID; +import java.util.function.LongSupplier; import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkState; @@ -474,15 +477,25 @@ public class LogicalPlanner newTableLayout, statisticsMetadata); Expression cubeWhere = analysis.getWhere((QuerySpecification) (insertCubeStatement.getQuery().getQueryBody())); - Expression rewritten = new QueryPlanner(analysis, planSymbolAllocator, idAllocator, buildLambdaDeclarationToSymbolMap(analysis, planSymbolAllocator), metadata, session, namedSubPlan, uniqueIdAllocator) - .rewriteExpression(tableWriterPlan, cubeWhere, analysis, buildLambdaDeclarationToSymbolMap(analysis, planSymbolAllocator)); + Expression rewritten = null; + if (cubeWhere != null) { + rewritten = new QueryPlanner(analysis, planSymbolAllocator, idAllocator, buildLambdaDeclarationToSymbolMap(analysis, planSymbolAllocator), metadata, session, namedSubPlan, uniqueIdAllocator) + .rewriteExpression(tableWriterPlan, cubeWhere, analysis, buildLambdaDeclarationToSymbolMap(analysis, planSymbolAllocator)); + } + TableHandle sourceTableHandle = insert.getSourceTable(); + //At this point it has been verified that source table has not been updated + //so insert into cube should be allowed + LongSupplier tableLastModifiedTimeSupplier = metadata.getTableLastModifiedTimeSupplier(session, sourceTableHandle); + checkState(tableLastModifiedTimeSupplier != null, "Table last modified time is null"); CubeFinishNode cubeFinishNode = new CubeFinishNode( idAllocator.getNextId(), tableWriterPlan.getRoot(), planSymbolAllocator.newSymbol("rows", BIGINT), - tableMetadata.getQualifiedName().toString(), - rewritten, - insertCubeStatement.isOverwrite()); + new CubeUpdateMetadata( + tableMetadata.getQualifiedName().toString(), + tableLastModifiedTimeSupplier.getAsLong(), + cubeWhere != null ? ExpressionFormatter.formatExpression(rewritten, Optional.empty()) : null, + insertCubeStatement.isOverwrite())); return new RelationPlan(cubeFinishNode, analysis.getScope(insertCubeStatement), cubeFinishNode.getOutputSymbols()); } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/AggregationRewriteWithCube.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/AggregationRewriteWithCube.java index fd58145e5..fde10e232 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/AggregationRewriteWithCube.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/AggregationRewriteWithCube.java @@ -68,6 +68,8 @@ import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; +import static io.hetu.core.spi.cube.CubeAggregateFunction.COUNT; +import static io.hetu.core.spi.cube.CubeAggregateFunction.SUM; import static io.prestosql.spi.StandardErrorCode.CUBE_ERROR; import static io.prestosql.spi.plan.AggregationNode.singleGroupingSet; import static io.prestosql.sql.planner.SymbolUtils.toSymbolReference; @@ -98,7 +100,7 @@ public class AggregationRewriteWithCube public PlanNode rewrite(AggregationNode originalAggregationNode, PlanNode filterNode) { - QualifiedObjectName starTreeTableName = QualifiedObjectName.valueOf(cubeMetadata.getCubeTableName()); + QualifiedObjectName starTreeTableName = QualifiedObjectName.valueOf(cubeMetadata.getCubeName()); TableHandle cubeTableHandle = metadata.getTableHandle(session, starTreeTableName) .orElseThrow(() -> new CubeNotFoundException(starTreeTableName.toString())); Map cubeColumnsMap = metadata.getColumnHandles(session, cubeTableHandle); @@ -130,7 +132,7 @@ public class AggregationRewriteWithCube ColumnMetadata cubeColumnMetadata = metadata.getColumnMetadata(session, cubeTableHandle, cubeColHandle); AggregationSignature aggregationSignature = cubeMetadata.getAggregationSignature(cubeColumnMetadata.getName()) .orElseThrow(() -> new ColumnNotFoundException(new SchemaTableName(starTreeTableName.getSchemaName(), starTreeTableName.getObjectName()), cubeColHandle.getColumnName())); - String aggFunction = AggregationSignature.COUNT_FUNCTION_NAME.equals(aggregationSignature.getFunction()) ? "sum" : aggregationSignature.getFunction(); + String aggFunction = COUNT.getName().equals(aggregationSignature.getFunction()) ? "sum" : aggregationSignature.getFunction(); SymbolReference argument = toSymbolReference(aggregatorSource.getScanSymbol()); FunctionHandle functionHandle = metadata.getFunctionAndTypeManager().lookupFunction(aggFunction, TypeSignatureProvider.fromTypeSignatures(typeSignature)); aggregationsBuilder.put(aggregatorSource.getOriginalAggSymbol(), new AggregationNode.Aggregation( @@ -320,7 +322,7 @@ public class AggregationRewriteWithCube } break; case "avg": - AggregationSignature sumSignature = new AggregationSignature(AggregationSignature.SUM_FUNCTION_NAME, originalColumnName, distinct); + AggregationSignature sumSignature = new AggregationSignature(SUM.getName(), originalColumnName, distinct); String sumColumnName = cubeMetadata.getColumn(sumSignature) .orElseThrow(() -> new PrestoException(CUBE_ERROR, "Cannot find column associated with aggregation " + sumSignature)); ColumnHandle sumColumnHandle = cubeColumnsMap.get(sumColumnName); @@ -341,7 +343,7 @@ public class AggregationRewriteWithCube } } } - AggregationSignature countSignature = new AggregationSignature(AggregationSignature.COUNT_FUNCTION_NAME, originalColumnName, distinct); + AggregationSignature countSignature = new AggregationSignature(COUNT.getName(), originalColumnName, distinct); String countColumnName = cubeMetadata.getColumn(countSignature) .orElseThrow(() -> new PrestoException(CUBE_ERROR, "Cannot find column associated with aggregation " + countSignature)); ColumnHandle countColumnHandle = cubeColumnsMap.get(countColumnName); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PruneUnreferencedOutputs.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PruneUnreferencedOutputs.java index 084cf77c8..fbe4b360d 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PruneUnreferencedOutputs.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PruneUnreferencedOutputs.java @@ -764,9 +764,7 @@ public class PruneUnreferencedOutputs node.getId(), source, node.getRowCountSymbol(), - node.getCubeName(), - node.getDataPredicate(), - node.isOverwrite()); + node.getMetadata()); } @Override diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/StarTreeAggregationRule.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/StarTreeAggregationRule.java index 5989593cb..1b4fd8d6f 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/StarTreeAggregationRule.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/StarTreeAggregationRule.java @@ -275,13 +275,17 @@ public class StarTreeAggregationRule } LongSupplier lastModifiedTimeSupplier = metadata.getTableLastModifiedTimeSupplier(session, tableHandle); - if (lastModifiedTimeSupplier != null) { - long lastModifiedTime = lastModifiedTimeSupplier.getAsLong(); - matchedCubeMetadataList = matchedCubeMetadataList.stream() - .filter(cubeMetadata -> cubeMetadata.getLastUpdated() > lastModifiedTime) - .collect(Collectors.toList()); + if (lastModifiedTimeSupplier == null) { + warningCollector.add(new PrestoWarning(EXPIRED_CUBE, "Unable to identify last modified time of " + tableName + ". Ignoring star tree cubes.")); + return Result.empty(); } + //Filter out cubes that were created before the source table was updated + long lastModifiedTime = lastModifiedTimeSupplier.getAsLong(); + matchedCubeMetadataList = matchedCubeMetadataList.stream() + .filter(cubeMetadata -> cubeMetadata.getSourceTableLastUpdatedTime() >= lastModifiedTime) + .collect(Collectors.toList()); + if (matchedCubeMetadataList.isEmpty()) { warningCollector.add(new PrestoWarning(EXPIRED_CUBE, tableName + " has been modified after creating cubes. Ignoring expired cubes.")); return Result.empty(); @@ -289,7 +293,7 @@ public class StarTreeAggregationRule //If multiple cubes are matching then lets select the recent built cube //so sort the cube based on the last updated time stamp - matchedCubeMetadataList.sort(Comparator.comparingLong(CubeMetadata::getLastUpdated).reversed()); + matchedCubeMetadataList.sort(Comparator.comparingLong(CubeMetadata::getLastUpdatedTime).reversed()); AggregationRewriteWithCube aggregationRewriteWithCube = new AggregationRewriteWithCube(metadata, session, symbolAllocator, idAllocator, symbolMapping, matchedCubeMetadataList.get(0)); return Result.ofPlanNode(aggregationRewriteWithCube.rewrite(aggregationNode, filterNode.orElse(null))); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/SymbolMapper.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/SymbolMapper.java index 932cb6524..b99fd67c1 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/SymbolMapper.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/SymbolMapper.java @@ -253,9 +253,7 @@ public class SymbolMapper node.getId(), source, map(node.getRowCountSymbol()), - node.getCubeName(), - node.getDataPredicate(), - node.isOverwrite()); + node.getMetadata()); } private PartitioningScheme canonicalize(PartitioningScheme scheme, PlanNode source) diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/plan/CubeFinishNode.java b/presto-main/src/main/java/io/prestosql/sql/planner/plan/CubeFinishNode.java index 9e91907a8..df08f63d4 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/plan/CubeFinishNode.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/plan/CubeFinishNode.java @@ -17,42 +17,34 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; +import io.prestosql.spi.cube.CubeUpdateMetadata; import io.prestosql.spi.plan.PlanNode; import io.prestosql.spi.plan.PlanNodeId; import io.prestosql.spi.plan.Symbol; -import io.prestosql.sql.tree.Expression; import javax.annotation.concurrent.Immutable; import java.util.List; -import static java.util.Objects.requireNonNull; - @Immutable public class CubeFinishNode extends InternalPlanNode { private final PlanNode source; - private final String cubeName; private final Symbol rowCountSymbol; - private final Expression dataPredicate; - private final boolean overwrite; + private final CubeUpdateMetadata metadata; @JsonCreator public CubeFinishNode( @JsonProperty("id") PlanNodeId id, @JsonProperty("source") PlanNode source, @JsonProperty("rowCountSymbol") Symbol rowCountSymbol, - @JsonProperty("cubeName") String cubeName, - @JsonProperty("dataPredicate") Expression dataPredicate, - @JsonProperty("overwrite") boolean overwrite) + @JsonProperty("metadata") CubeUpdateMetadata metadata) { super(id); - this.source = requireNonNull(source, "source is null"); - this.cubeName = requireNonNull(cubeName, "Cube name is null"); - this.rowCountSymbol = requireNonNull(rowCountSymbol, "rowCountSymbol is null"); - this.dataPredicate = requireNonNull(dataPredicate, "Predicate is null"); - this.overwrite = overwrite; + this.source = source; + this.rowCountSymbol = rowCountSymbol; + this.metadata = metadata; } @JsonProperty @@ -68,21 +60,9 @@ public class CubeFinishNode } @JsonProperty - public String getCubeName() + public CubeUpdateMetadata getMetadata() { - return cubeName; - } - - @JsonProperty - public Expression getDataPredicate() - { - return dataPredicate; - } - - @JsonProperty - public boolean isOverwrite() - { - return overwrite; + return metadata; } @Override @@ -110,8 +90,6 @@ public class CubeFinishNode getId(), Iterables.getOnlyElement(newChildren), rowCountSymbol, - cubeName, - dataPredicate, - overwrite); + metadata); } } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/planprinter/IoPlanPrinter.java b/presto-main/src/main/java/io/prestosql/sql/planner/planprinter/IoPlanPrinter.java index 77094de43..f29cb3068 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/planprinter/IoPlanPrinter.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/planprinter/IoPlanPrinter.java @@ -536,7 +536,7 @@ public class IoPlanPrinter @Override public Void visitCubeFinish(CubeFinishNode node, IoPlanBuilder context) { - QualifiedObjectName qualifiedObjectName = QualifiedObjectName.valueOf(node.getCubeName()); + QualifiedObjectName qualifiedObjectName = QualifiedObjectName.valueOf(node.getMetadata().getCubeName()); context.setOutputTable(new CatalogSchemaTableName( qualifiedObjectName.getCatalogName(), qualifiedObjectName.getSchemaName(), diff --git a/presto-main/src/main/java/io/prestosql/sql/rewrite/InsertCubeRewrite.java b/presto-main/src/main/java/io/prestosql/sql/rewrite/InsertCubeRewrite.java index 03a30ee66..efce7b2b3 100644 --- a/presto-main/src/main/java/io/prestosql/sql/rewrite/InsertCubeRewrite.java +++ b/presto-main/src/main/java/io/prestosql/sql/rewrite/InsertCubeRewrite.java @@ -29,6 +29,7 @@ import io.prestosql.security.AccessControl; import io.prestosql.spi.PrestoException; import io.prestosql.spi.StandardErrorCode; import io.prestosql.spi.connector.QualifiedObjectName; +import io.prestosql.sql.ExpressionFormatter; import io.prestosql.sql.analyzer.QueryExplainer; import io.prestosql.sql.parser.SqlParser; import io.prestosql.sql.tree.AstVisitor; @@ -101,23 +102,28 @@ public class InsertCubeRewrite CubeMetadata cubeMetadata = cubeMetaStore.getMetadataFromCubeName(targetCube.toString()).orElseThrow(() -> new PrestoException(StandardErrorCode.CUBE_ERROR, String.format("Cube not found '%s'", targetCube.toString()))); Set group = cubeMetadata.getGroup(); ImmutableList.Builder builder = ImmutableList.builder(); - new IdentifierBuilderVisitor().process(node.getWhere(), builder); - Set whereColumns = builder.build() - .stream() - .map(Identifier::getValue) - .collect(Collectors.toCollection(() -> new TreeSet<>(String.CASE_INSENSITIVE_ORDER))); - if (!group.containsAll(whereColumns)) { - throw new IllegalArgumentException("All columns in where clause must be part Cube group."); + if (node.getWhere().isPresent()) { + new IdentifierBuilderVisitor().process(node.getWhere().get(), builder); + Set whereColumns = builder.build() + .stream() + .map(Identifier::getValue) + .collect(Collectors.toCollection(() -> new TreeSet<>(String.CASE_INSENSITIVE_ORDER))); + if (whereColumns.isEmpty()) { + throw new IllegalArgumentException("Invalid predicate. " + ExpressionFormatter.formatExpression(node.getWhere().get(), Optional.empty())); + } + if (!group.containsAll(whereColumns)) { + throw new IllegalArgumentException("All columns in where clause must be part Cube group."); + } } return buildCubeInsert(cubeMetadata, node, group); } private InsertCube buildCubeInsert(CubeMetadata cubeMetadata, InsertCube node, Set cubeGroup) { - Expression newDataPredicate = node.getWhere(); - QualifiedObjectName originalTableName = QualifiedObjectName.valueOf(cubeMetadata.getOriginalTableName()); + Optional newDataPredicate = node.getWhere(); + QualifiedObjectName sourceTableName = QualifiedObjectName.valueOf(cubeMetadata.getSourceTableName()); List insertColumns = new ArrayList<>(); - QualifiedName sourceTable = QualifiedName.of(originalTableName.getCatalogName(), originalTableName.getSchemaName(), originalTableName.getObjectName()); + QualifiedName sourceTable = QualifiedName.of(sourceTableName.getCatalogName(), sourceTableName.getSchemaName(), sourceTableName.getObjectName()); List selectItems = new ArrayList<>(); cubeMetadata.getAggregations().forEach(aggColumn -> { AggregationSignature aggregationSignature = cubeMetadata.getAggregationSignature(aggColumn).get(); @@ -152,7 +158,7 @@ public class InsertCubeRewrite QuerySpecification selectQuery = new QuerySpecification( new Select(false, selectItems), Optional.of(new Table(sourceTable)), - Optional.of(newDataPredicate), + newDataPredicate, Optional.of(groupBy), Optional.empty(), Optional.empty(), diff --git a/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java b/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java index bc5b05b9b..a5d66d1f9 100644 --- a/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java +++ b/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java @@ -271,11 +271,15 @@ final class ShowQueriesRewrite } else { QualifiedObjectName qualifiedTableName = createQualifiedObjectName(session, node, node.getTableName().get()); + Optional tableHandle = metadata.getTableHandle(session, qualifiedTableName); + if (!tableHandle.isPresent()) { + throw new SemanticException(MISSING_TABLE, node, "Table %s does not exist", qualifiedTableName.toString()); + } cubeMetadataList = cubeMetaStore.getMetadataList(qualifiedTableName.toString()); } Map cubeStatusMap = new HashMap<>(); cubeMetadataList.forEach(cubeMetadata -> { - QualifiedObjectName qualifiedTableName = QualifiedObjectName.valueOf(cubeMetadata.getOriginalTableName()); + QualifiedObjectName qualifiedTableName = QualifiedObjectName.valueOf(cubeMetadata.getSourceTableName()); Map tableLastModifiedTimeMap = new HashMap<>(); long tableLastModifiedTime = tableLastModifiedTimeMap.computeIfAbsent(qualifiedTableName, ignored -> { TableHandle tableHandle = metadata.getTableHandle(session, qualifiedTableName).get(); @@ -284,10 +288,10 @@ final class ShowQueriesRewrite }); CubeStatus status = cubeMetadata.getCubeStatus(); if (status == CubeStatus.INACTIVE) { - cubeStatusMap.put(cubeMetadata.getCubeTableName(), "InActive"); + cubeStatusMap.put(cubeMetadata.getCubeName(), "Inactive"); } else { - cubeStatusMap.put(cubeMetadata.getCubeTableName(), tableLastModifiedTime > cubeMetadata.getLastUpdated() ? "Expired" : "Active"); + cubeStatusMap.put(cubeMetadata.getCubeName(), tableLastModifiedTime > cubeMetadata.getSourceTableLastUpdatedTime() ? "Expired" : "Active"); } }); rows.add(row( @@ -300,9 +304,9 @@ final class ShowQueriesRewrite FALSE_LITERAL)); cubeMetadataList.forEach(cubeMetadata -> { rows.add(row( - new StringLiteral(cubeMetadata.getCubeTableName()), - new StringLiteral(cubeMetadata.getOriginalTableName()), - new StringLiteral(cubeStatusMap.get(cubeMetadata.getCubeTableName())), + new StringLiteral(cubeMetadata.getCubeName()), + new StringLiteral(cubeMetadata.getSourceTableName()), + new StringLiteral(cubeStatusMap.get(cubeMetadata.getCubeName())), new StringLiteral(String.join(",", cubeMetadata.getDimensions())), new StringLiteral(cubeMetadata.getAggregationSignatures().stream().map(AggregationSignature::toString).collect(Collectors.joining(","))), new StringLiteral(String.join(",", cubeMetadata.getPredicateString())), diff --git a/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestStarTreeAggregationRule.java b/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestStarTreeAggregationRule.java index ac94250ee..df7b91f6a 100644 --- a/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestStarTreeAggregationRule.java +++ b/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestStarTreeAggregationRule.java @@ -747,7 +747,7 @@ public class TestStarTreeAggregationRule } @Test - public void testDoNotUseCubeIfOriginalTableUpdatedAfterCubeCreated() + public void testDoNotUseCubeIfSourceTableUpdatedAfterCubeCreated() { Mockito.when(cubeManager.getCubeProvider(anyString())).then(new Returns(Optional.of(provider))); Mockito.when(cubeManager.getMetaStore(anyString())).then(new Returns(Optional.of(cubeMetaStore))); @@ -764,7 +764,7 @@ public class TestStarTreeAggregationRule List metadataList = ImmutableList.of(cubeMetadata); Mockito.when(cubeMetaStore.getMetadataList(eq("local.sf1.0.orders"))).then(new Returns(metadataList)); Mockito.when(cubeMetadata.matches(any(CubeStatement.class))).thenReturn(true); - Mockito.when(cubeMetadata.getLastUpdated()).thenReturn(DateTimeUtils.parseTimestampWithoutTimeZone("2020-01-01 12:00:00")); + Mockito.when(cubeMetadata.getLastUpdatedTime()).thenReturn(DateTimeUtils.parseTimestampWithoutTimeZone("2020-01-01 12:00:00")); StarTreeAggregationRule starTreeAggregationRule = new StarTreeAggregationRule(cubeManager, metadata); tester().assertThat(starTreeAggregationRule) diff --git a/presto-parser/src/main/antlr4/io/prestosql/sql/parser/SqlBase.g4 b/presto-parser/src/main/antlr4/io/prestosql/sql/parser/SqlBase.g4 index 73a9780d5..805e4c032 100644 --- a/presto-parser/src/main/antlr4/io/prestosql/sql/parser/SqlBase.g4 +++ b/presto-parser/src/main/antlr4/io/prestosql/sql/parser/SqlBase.g4 @@ -61,8 +61,8 @@ statement | CREATE CUBE (IF NOT EXISTS)? cubeName=qualifiedName ON tableName=qualifiedName WITH '(' AGGREGATIONS EQ '(' aggregations ')' ',' GROUP EQ '(' cubeGroup ')' (',' cubeProperties)? ')' #createCube - | INSERT INTO CUBE cubeName=qualifiedName WHERE expression #insertCube - | INSERT OVERWRITE CUBE cubeName=qualifiedName WHERE expression #insertOverwriteCube + | INSERT INTO CUBE cubeName=qualifiedName (WHERE expression)? #insertCube + | INSERT OVERWRITE CUBE cubeName=qualifiedName (WHERE expression)? #insertOverwriteCube | DROP CUBE (IF EXISTS)? cubeName=qualifiedName #dropCube | SHOW CUBES (FOR tableName=qualifiedName)? #showCubes | CREATE INDEX (IF NOT EXISTS)? indexName=qualifiedName diff --git a/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java b/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java index 0b54cb7d4..be8f71059 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java +++ b/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java @@ -63,6 +63,7 @@ import io.prestosql.sql.tree.GrantRoles; import io.prestosql.sql.tree.GrantorSpecification; import io.prestosql.sql.tree.Identifier; import io.prestosql.sql.tree.Insert; +import io.prestosql.sql.tree.InsertCube; import io.prestosql.sql.tree.Intersect; import io.prestosql.sql.tree.Isolation; import io.prestosql.sql.tree.Join; @@ -979,19 +980,22 @@ public final class SqlFormatter } builder.append(formatName(node.getCubeName())); builder.append(" ON "); - builder.append(formatName(node.getTableName())); - builder.append(" WITH "); - List aggregations = node.getAggregations().stream().map(Expression::toString).collect(Collectors.toList()); - String propertyList = node.getProperties().stream() - .map(element -> formatExpression(element.getName(), parameters) + " = " + - formatExpression(element.getValue(), parameters)) + builder.append(formatName(node.getSourceTableName())); + builder.append(" WITH ("); + String aggregations = node.getAggregations().stream() + .map(Expression::toString) .collect(joining(", ")); - String groupsList = node.getGroupingSet().stream() - .map(Identifier::toString) + String group = node.getGroupingSet().stream() + .map(Identifier::getValue) .collect(joining(", ")); - builder.append(" ( AGGREGATIONS = (").append(String.join(", ", aggregations)).append(")"); - builder.append(", GROUP=(").append(groupsList).append(")"); - builder.append(", PROPERTIES = (").append(propertyList).append(")"); + builder.append("AGGREGATIONS = (").append(aggregations).append("), "); + builder.append("GROUP=(").append(group).append(")"); + if (!node.getProperties().isEmpty()) { + String properties = node.getProperties().stream() + .map(element -> formatExpression(element.getName(), parameters) + " = " + formatExpression(element.getValue(), parameters)) + .collect(joining(", ")); + builder.append(", ").append(properties); + } builder.append(" )"); return null; } @@ -1298,6 +1302,24 @@ public final class SqlFormatter return null; } + @Override + public Void visitInsertCube(InsertCube node, Integer indent) + { + if (node.isOverwrite()) { + builder.append("INSERT OVERWRITE CUBE ") + .append(node.getCubeName()); + } + else { + builder.append("INSERT INTO CUBE ") + .append(node.getCubeName()); + } + if (node.getWhere().isPresent()) { + builder.append(" WHERE ") + .append(formatExpression(node.getWhere().get(), Optional.empty())); + } + return null; + } + @Override public Void visitUpdate(Update node, Integer indent) { diff --git a/presto-parser/src/main/java/io/prestosql/sql/parser/AstBuilder.java b/presto-parser/src/main/java/io/prestosql/sql/parser/AstBuilder.java index 9af263f30..20f4394b2 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/parser/AstBuilder.java +++ b/presto-parser/src/main/java/io/prestosql/sql/parser/AstBuilder.java @@ -54,7 +54,6 @@ import io.prestosql.sql.tree.CurrentTime; import io.prestosql.sql.tree.CurrentUser; import io.prestosql.sql.tree.Deallocate; import io.prestosql.sql.tree.DecimalLiteral; -import io.prestosql.sql.tree.DefaultExpressionTraversalVisitor; import io.prestosql.sql.tree.Delete; import io.prestosql.sql.tree.DereferenceExpression; import io.prestosql.sql.tree.DescribeInput; @@ -332,14 +331,14 @@ class AstBuilder }); QualifiedName cubeName = getQualifiedName(context.cubeName); - QualifiedName originalTableName = getQualifiedName(context.tableName); + QualifiedName sourceTableName = getQualifiedName(context.tableName); List properties = ImmutableList.of(); if (context.cubeProperties() != null) { properties = visit(context.cubeProperties().property(), Property.class); } - return new CreateCube(getLocation(context), cubeName, originalTableName, groupingSet, decomposedAggregations, context.EXISTS() != null, properties); + return new CreateCube(getLocation(context), cubeName, sourceTableName, groupingSet, decomposedAggregations, context.EXISTS() != null, properties); } @Override @@ -347,10 +346,7 @@ class AstBuilder { QualifiedName cubeName = getQualifiedName(context.qualifiedName()); Optional optionalExpression = visitIfPresent(context.expression(), Expression.class); - if (!optionalExpression.isPresent()) { - throw new IllegalArgumentException("WHERE expression is mandatory!"); - } - return new InsertCube(getLocation(context), cubeName, optionalExpression.get(), null, false); + return new InsertCube(getLocation(context), cubeName, optionalExpression, false); } @Override @@ -358,21 +354,7 @@ class AstBuilder { QualifiedName cubeName = getQualifiedName(context.qualifiedName()); Optional optionalExpression = visitIfPresent(context.expression(), Expression.class); - if (!optionalExpression.isPresent()) { - throw new IllegalArgumentException("WHERE expression is mandatory!"); - } - return new InsertCube(getLocation(context), cubeName, optionalExpression.get(), null, true); - } - - private static class IdentifierBuilderVisitor - extends DefaultExpressionTraversalVisitor> - { - @Override - protected Void visitIdentifier(Identifier node, ImmutableList.Builder builder) - { - builder.add(node); - return null; - } + return new InsertCube(getLocation(context), cubeName, optionalExpression, true); } @Override diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/CreateCube.java b/presto-parser/src/main/java/io/prestosql/sql/tree/CreateCube.java index dcbcac429..345b65ff1 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/tree/CreateCube.java +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/CreateCube.java @@ -30,29 +30,29 @@ public class CreateCube { private final QualifiedName cubeName; private final boolean notExists; - private final QualifiedName tableName; + private final QualifiedName sourceTableName; private final List groupingSet; private final Set aggregations; private final List properties; - public CreateCube(QualifiedName cubeName, QualifiedName tableName, List groupingSet, + public CreateCube(QualifiedName cubeName, QualifiedName sourceTableName, List groupingSet, Set aggregations, boolean notExists, List properties) { - this(Optional.empty(), cubeName, tableName, groupingSet, aggregations, notExists, properties); + this(Optional.empty(), cubeName, sourceTableName, groupingSet, aggregations, notExists, properties); } - public CreateCube(NodeLocation location, QualifiedName cubeName, QualifiedName tableName, List groupingSet, + public CreateCube(NodeLocation location, QualifiedName cubeName, QualifiedName sourceTableName, List groupingSet, Set aggregations, boolean notExists, List properties) { - this(Optional.of(location), cubeName, tableName, groupingSet, aggregations, notExists, properties); + this(Optional.of(location), cubeName, sourceTableName, groupingSet, aggregations, notExists, properties); } - private CreateCube(Optional location, QualifiedName cubeName, QualifiedName tableName, List groupingSet, + private CreateCube(Optional location, QualifiedName cubeName, QualifiedName sourceTableName, List groupingSet, Set aggregations, boolean notExists, List properties) { super(location); this.cubeName = requireNonNull(cubeName, "cube name is null"); - this.tableName = requireNonNull(tableName, "table name is null"); + this.sourceTableName = requireNonNull(sourceTableName, "table name is null"); this.groupingSet = groupingSet; this.aggregations = aggregations; this.notExists = notExists; @@ -64,9 +64,9 @@ public class CreateCube return cubeName; } - public QualifiedName getTableName() + public QualifiedName getSourceTableName() { - return tableName; + return sourceTableName; } public List getProperties() @@ -109,7 +109,7 @@ public class CreateCube { return Objects.hash( cubeName, - tableName, + sourceTableName, groupingSet, aggregations, notExists, @@ -121,7 +121,7 @@ public class CreateCube { return toStringHelper(this) .add("cubeName", cubeName) - .add("tableName", tableName) + .add("tableName", sourceTableName) .add("groupingSet", groupingSet) .add("aggregations", aggregations) .add("notExists", notExists) @@ -140,7 +140,7 @@ public class CreateCube } CreateCube that = (CreateCube) o; return Objects.equals(cubeName, that.cubeName) && - Objects.equals(tableName, that.tableName) && + Objects.equals(sourceTableName, that.sourceTableName) && Objects.equals(groupingSet, that.groupingSet) && Objects.equals(aggregations, that.aggregations) && Objects.equals(notExists, that.notExists) && diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/InsertCube.java b/presto-parser/src/main/java/io/prestosql/sql/tree/InsertCube.java index f092b7e6e..49ffa51c0 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/tree/InsertCube.java +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/InsertCube.java @@ -25,22 +25,32 @@ public class InsertCube extends Statement { private final QualifiedName cubeName; - private final Expression where; + private final Optional where; private final List columns; private final Query query; private final boolean overwrite; - public InsertCube(QualifiedName cubeName, Expression where, List columns, boolean overwrite) + public InsertCube(QualifiedName cubeName, Optional where, boolean overwrite) + { + this(cubeName, where, null, overwrite); + } + + public InsertCube(NodeLocation location, QualifiedName cubeName, Optional where, boolean overwrite) + { + this(location, cubeName, where, null, overwrite); + } + + public InsertCube(QualifiedName cubeName, Optional where, List columns, boolean overwrite) { this(cubeName, where, columns, overwrite, null); } - public InsertCube(NodeLocation location, QualifiedName cubeName, Expression where, List columns, boolean overwrite) + public InsertCube(NodeLocation location, QualifiedName cubeName, Optional where, List columns, boolean overwrite) { this(location, cubeName, where, columns, overwrite, null); } - public InsertCube(QualifiedName cubeName, Expression where, List columns, boolean overwrite, Query query) + public InsertCube(QualifiedName cubeName, Optional where, List columns, boolean overwrite, Query query) { super(Optional.empty()); this.cubeName = cubeName; @@ -50,7 +60,7 @@ public class InsertCube this.query = query; } - public InsertCube(NodeLocation location, QualifiedName cubeName, Expression where, List columns, boolean overwrite, Query query) + public InsertCube(NodeLocation location, QualifiedName cubeName, Optional where, List columns, boolean overwrite, Query query) { super(Optional.of(location)); this.cubeName = cubeName; @@ -65,7 +75,7 @@ public class InsertCube return cubeName; } - public Expression getWhere() + public Optional getWhere() { return where; } @@ -126,6 +136,7 @@ public class InsertCube } InsertCube o = (InsertCube) obj; return Objects.equals(cubeName, o.cubeName) && - Objects.equals(where, o.where); + Objects.equals(where, o.where) && + Objects.equals(overwrite, o.overwrite); } } diff --git a/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParser.java b/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParser.java index 5e62491f7..98f507cef 100644 --- a/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParser.java +++ b/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParser.java @@ -36,6 +36,7 @@ import io.prestosql.sql.tree.ColumnDefinition; import io.prestosql.sql.tree.Comment; import io.prestosql.sql.tree.Commit; import io.prestosql.sql.tree.ComparisonExpression; +import io.prestosql.sql.tree.CreateCube; import io.prestosql.sql.tree.CreateRole; import io.prestosql.sql.tree.CreateSchema; import io.prestosql.sql.tree.CreateTable; @@ -76,6 +77,7 @@ import io.prestosql.sql.tree.GroupingSets; import io.prestosql.sql.tree.Identifier; import io.prestosql.sql.tree.IfExpression; import io.prestosql.sql.tree.Insert; +import io.prestosql.sql.tree.InsertCube; import io.prestosql.sql.tree.Intersect; import io.prestosql.sql.tree.IntervalLiteral; import io.prestosql.sql.tree.IntervalLiteral.IntervalField; @@ -1384,38 +1386,6 @@ public class TestSqlParser new Identifier("b"))))); } - @Test - public void testCreateCube() - { -// assertStatement("CREATE CUBE foo ON bar WITH (DIMENSIONS=key,val, AGGREGATIONS=count(\"*\"))", -// new CreateCube(QualifiedName.of("foo"), -// QualifiedName.of("bar"), -// ImmutableList.of( -// new Identifier("key"), -// new Identifier("val")), -// false, -// ImmutableList.of(new FunctionCall(QualifiedName.of("COUNT"), ImmutableList.of(new Identifier("*")))), -// Optional.empty())); -// assertStatement("CREATE CUBE foo ON bar WITH (DIMENSIONS=key,val, AGGREGATIONS=sum(cost))", -// new CreateCube(QualifiedName.of("foo"), -// QualifiedName.of("bar"), -// ImmutableList.of( -// new Identifier("key"), -// new Identifier("val")), -// false, -// ImmutableList.of(new FunctionCall(QualifiedName.of("SUM"), ImmutableList.of(new Identifier("cost")))), -// Optional.empty())); -// assertStatement("CREATE CUBE IF NOT EXISTS foo ON bar WITH (DIMENSIONS=key,val, AGGREGATIONS=sum(cost))", -// new CreateCube(QualifiedName.of("foo"), -// QualifiedName.of("bar"), -// ImmutableList.of( -// new Identifier("key"), -// new Identifier("val")), -// true, -// ImmutableList.of(new FunctionCall(QualifiedName.of("SUM"), ImmutableList.of(new Identifier("cost")))), -// Optional.empty())); - } - @Test public void testCreateTable() { @@ -1661,6 +1631,68 @@ public class TestSqlParser assertStatement(queryUnparenthesizedWithHasAlias, new CreateTableAsSelect(table, query, false, ImmutableList.of(), false, Optional.of(ImmutableList.of(new Identifier("a"))), Optional.empty())); } + @Test + public void testCreateCube() + { + assertStatement("CREATE CUBE foo ON bar WITH (AGGREGATIONS=(count(c)), GROUP = (a, b))", + new CreateCube(QualifiedName.of("foo"), + QualifiedName.of("bar"), + ImmutableList.of( + new Identifier("a"), + new Identifier("b")), + ImmutableSet.of( + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("c")))), + false, + ImmutableList.of())); + + assertStatement("CREATE CUBE foo ON bar WITH (AGGREGATIONS=(count(c), sum(d), avg(e)), GROUP = (a, b))", + new CreateCube(QualifiedName.of("foo"), + QualifiedName.of("bar"), + ImmutableList.of( + new Identifier("a"), + new Identifier("b")), + ImmutableSet.of( + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("c"))), + new FunctionCall(QualifiedName.of("sum"), ImmutableList.of(new Identifier("d"))), + new FunctionCall(QualifiedName.of("sum"), ImmutableList.of(new Identifier("e"))), + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("e")))), + false, + ImmutableList.of())); + + assertStatement("CREATE CUBE c1.s1.foo ON c2.s2.bar WITH (AGGREGATIONS=(count(c)), GROUP = (a, b))", + new CreateCube(QualifiedName.of("c1", "s1", "foo"), + QualifiedName.of("c2", "s2", "bar"), + ImmutableList.of( + new Identifier("a"), + new Identifier("b")), + ImmutableSet.of( + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("c")))), + false, + ImmutableList.of())); + + assertStatement("CREATE CUBE IF NOT EXISTS foo ON bar WITH (AGGREGATIONS=(count(c)), GROUP = (a, b))", + new CreateCube(QualifiedName.of("foo"), + QualifiedName.of("bar"), + ImmutableList.of( + new Identifier("a"), + new Identifier("b")), + ImmutableSet.of( + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("c")))), + true, + ImmutableList.of())); + + assertStatement("CREATE CUBE IF NOT EXISTS foo ON bar WITH (AGGREGATIONS=(count(c)), GROUP = (a, b), format = 'ORC', partitioned_by = ARRAY[ 'd' ])", + new CreateCube(QualifiedName.of("foo"), + QualifiedName.of("bar"), + ImmutableList.of( + new Identifier("a"), + new Identifier("b")), + ImmutableSet.of( + new FunctionCall(QualifiedName.of("count"), ImmutableList.of(new Identifier("c")))), + true, + ImmutableList.of(new Property(new Identifier("format"), new StringLiteral("ORC")), new Property(new Identifier("partitioned_by"), new ArrayConstructor(ImmutableList.of(new StringLiteral("d"))))))); + } + @Test public void testDropCache() { @@ -1735,6 +1767,32 @@ public class TestSqlParser new Insert(table, Optional.of(ImmutableList.of(identifier("c1"), identifier("c2"))), query, true)); } + @Test + public void testInsertIntoCube() + { + assertStatement("INSERT INTO CUBE foo WHERE d1 > 10", + new InsertCube(QualifiedName.of("foo"), + Optional.of(new ComparisonExpression(GREATER_THAN, new Identifier("d1"), new LongLiteral("10"))), + false)); + assertStatement("INSERT INTO CUBE c1.s1.foo WHERE d1 > 10", + new InsertCube(QualifiedName.of("c1", "s1", "foo"), + Optional.of(new ComparisonExpression(GREATER_THAN, new Identifier("d1"), new LongLiteral("10"))), + false)); + } + + @Test + public void testInsertOverwriteCube() + { + assertStatement("INSERT OVERWRITE CUBE foo WHERE d1 BETWEEN 1012020 AND 31012020", + new InsertCube(QualifiedName.of("foo"), + Optional.of(new BetweenPredicate(new Identifier("d1"), new LongLiteral("1012020"), new LongLiteral("31012020"))), + true)); + assertStatement("INSERT OVERWRITE CUBE c1.s1.foo WHERE d1 BETWEEN 1012020 AND 31012020", + new InsertCube(QualifiedName.of("c1", "s1", "foo"), + Optional.of(new BetweenPredicate(new Identifier("d1"), new LongLiteral("1012020"), new LongLiteral("31012020"))), + true)); + } + @Test public void testDelete() { diff --git a/presto-spi/src/main/java/io/prestosql/spi/cube/CubeUpdateMetadata.java b/presto-spi/src/main/java/io/prestosql/spi/cube/CubeUpdateMetadata.java new file mode 100644 index 000000000..15e85996c --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/cube/CubeUpdateMetadata.java @@ -0,0 +1,77 @@ +/* + * Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.prestosql.spi.cube; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import static java.util.Objects.requireNonNull; + +public class CubeUpdateMetadata +{ + private final String cubeName; + private final long tableLastUpdatedTime; + private final String dataPredicateString; + private final boolean overwrite; + + @JsonCreator + public CubeUpdateMetadata( + @JsonProperty("cubeName") String cubeName, + @JsonProperty("tableLastUpdatedTime") long tableLastUpdatedTime, + @JsonProperty("dataPredicate") String dataPredicateString, + @JsonProperty("overwrite") boolean overwrite) + { + this.cubeName = requireNonNull(cubeName, "cubeName is null"); + this.tableLastUpdatedTime = tableLastUpdatedTime; + this.dataPredicateString = dataPredicateString; + this.overwrite = overwrite; + } + + @JsonProperty + public String getCubeName() + { + return cubeName; + } + + @JsonProperty + public long getTableLastUpdatedTime() + { + return tableLastUpdatedTime; + } + + @JsonProperty + public String getDataPredicateString() + { + return dataPredicateString; + } + + @JsonProperty + public boolean isOverwrite() + { + return overwrite; + } + + @Override + public String toString() + { + return "CubeUpdateMetadata{" + + "cubeName='" + cubeName + '\'' + + ", tableLastUpdatedTime=" + tableLastUpdatedTime + + ", dataPredicateString='" + dataPredicateString + '\'' + + ", overwrite=" + overwrite + + '}'; + } +} diff --git a/presto-tests/src/main/java/io/prestosql/tests/AbstractTestStarTreeQueries.java b/presto-tests/src/main/java/io/prestosql/tests/AbstractTestStarTreeQueries.java index 20f004442..057bb7ca1 100644 --- a/presto-tests/src/main/java/io/prestosql/tests/AbstractTestStarTreeQueries.java +++ b/presto-tests/src/main/java/io/prestosql/tests/AbstractTestStarTreeQueries.java @@ -15,24 +15,45 @@ package io.prestosql.tests; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import io.prestosql.Session; import io.prestosql.SystemSessionProperties; import io.prestosql.testing.MaterializedResult; +import io.prestosql.testing.MaterializedRow; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; import static com.google.common.collect.Iterables.getOnlyElement; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; public abstract class AbstractTestStarTreeQueries extends AbstractTestQueryFramework { + Session sessionStarTree; + Session sessionNoStarTree; + protected AbstractTestStarTreeQueries(QueryRunnerSupplier supplier) { super(supplier); } + @BeforeClass + public void setUp() + { + sessionStarTree = Session.builder(getSession()) + .setSystemProperty(SystemSessionProperties.ENABLE_STAR_TREE_INDEX, "true") + .build(); + sessionNoStarTree = Session.builder(getSession()) + .setSystemProperty(SystemSessionProperties.ENABLE_STAR_TREE_INDEX, "false") + .build(); + //Create Empty to force create Metadata catalog and schema. To avoid concurrency issue. + assertUpdate(sessionNoStarTree, "CREATE CUBE nation_count_all ON nation WITH (AGGREGATIONS=(count(*)), group=())"); + assertUpdate("DROP CUBE nation_count_all"); + } + @Test public void testStarTreeSessionProperty() { @@ -45,22 +66,134 @@ public abstract class AbstractTestStarTreeQueries } @Test - public void testStarTree() + public void testAggregations() { - Session sessionStarTree = Session.builder(getSession()) - .setSystemProperty(SystemSessionProperties.ENABLE_STAR_TREE_INDEX, "true") - .build(); - Session sessionNoStarTree = Session.builder(getSession()) - .setSystemProperty(SystemSessionProperties.ENABLE_STAR_TREE_INDEX, "false") - .build(); - assertUpdate(sessionNoStarTree, "CREATE CUBE nation_cube ON nation " + + assertUpdate(sessionNoStarTree, "CREATE CUBE nation_aggregations_cube_1 ON nation " + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); - assertUpdate(sessionNoStarTree, "INSERT INTO CUBE nation_cube where nationkey > -1", 25); + assertUpdate(sessionNoStarTree, "INSERT INTO CUBE nation_aggregations_cube_1 where nationkey > -1", 25); + assertQueryFails(sessionNoStarTree, "INSERT INTO CUBE nation_aggregations_cube_1 where 1 > 0", "Invalid predicate\\. \\(1 > 0\\)"); assertQuery(sessionStarTree, "SELECT min(regionkey), max(regionkey), sum(regionkey) from nation group by nationkey"); + assertQuery(sessionStarTree, "SELECT COUNT(distinct nationkey), count(distinct regionkey) from nation"); assertQuery(sessionStarTree, "SELECT COUNT(distinct nationkey), count(distinct regionkey) from nation group by nationkey"); assertQuery(sessionStarTree, "SELECT avg(nationkey) from nation group by nationkey"); - assertUpdate("DROP CUBE nation_cube"); + assertUpdate("DROP CUBE nation_aggregations_cube_1"); + } + + @Test + public void testShowCubes() + { + assertUpdate(sessionNoStarTree, "CREATE CUBE nation_show_cube_1 ON nation " + + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); + assertUpdate(sessionNoStarTree, "CREATE CUBE nation_show_cube_2 ON nation " + + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + + " group=())"); + MaterializedResult result = computeActual("SHOW CUBES"); + MaterializedRow matchingRow1 = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_show_cube_1")).findFirst().orElse(null); + assertNotNull(matchingRow1); + assertTrue(matchingRow1.getFields().containsAll(ImmutableList.of("hive.tpch.nation_show_cube_1", "hive.tpch.nation", "Inactive", "nationkey"))); + + MaterializedRow matchingRow2 = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_show_cube_2")).findFirst().orElse(null); + assertNotNull(matchingRow2); + assertTrue(matchingRow2.getFields().containsAll(ImmutableList.of("hive.tpch.nation_show_cube_2", "hive.tpch.nation", "Inactive", ""))); + + result = computeActual("SHOW CUBES FOR nation"); + assertEquals(result.getRowCount(), 2); + + matchingRow1 = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_show_cube_1")).findFirst().orElse(null); + assertNotNull(matchingRow1); + assertTrue(result.getMaterializedRows().get(0).getFields().containsAll(ImmutableList.of("hive.tpch.nation_show_cube_1", "hive.tpch.nation", "Inactive", "nationkey"))); + + matchingRow2 = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_show_cube_2")).findFirst().orElse(null); + assertNotNull(matchingRow2); + assertTrue(result.getMaterializedRows().get(1).getFields().containsAll(ImmutableList.of("hive.tpch.nation_show_cube_2", "hive.tpch.nation", "Inactive", ""))); + assertUpdate("DROP CUBE nation_show_cube_1"); + assertUpdate("DROP CUBE nation_show_cube_2"); + } + + @Test + public void testInsertIntoCube() + { + computeActual("CREATE TABLE nation_table_cube_insert_test_1 AS SELECT * FROM nation"); + assertUpdate("CREATE CUBE nation_insert_cube_1 ON nation_table_cube_insert_test_1 " + + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); + assertUpdate("INSERT INTO CUBE nation_insert_cube_1 where nationkey > 5", 19); + assertQueryFails("INSERT INTO CUBE nation where 1 > 0", "Cube not found 'hive.tpch.nation'"); + assertQueryFails("INSERT INTO CUBE nation_insert_cube_1 where regionkey > 5", "All columns in where clause must be part Cube group\\."); + assertUpdate("DROP CUBE nation_insert_cube_1"); + assertUpdate("DROP TABLE nation_table_cube_insert_test_1"); + } + + @Test + public void testInsertOverwriteCube() + { + computeActual("CREATE TABLE nation_table_cube_insert_overwrite_test_1 AS SELECT * FROM nation"); + assertUpdate("CREATE CUBE nation_insert_overwrite_cube_1 ON nation_table_cube_insert_overwrite_test_1 " + + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); + assertUpdate("INSERT INTO CUBE nation_insert_overwrite_cube_1 where nationkey > 5", 19); + assertEquals(computeScalar("SELECT COUNT(*) FROM nation_insert_overwrite_cube_1"), 19L); + assertUpdate("INSERT OVERWRITE CUBE nation_insert_overwrite_cube_1 where nationkey > 5", 19); + assertEquals(computeScalar("SELECT COUNT(*) FROM nation_insert_overwrite_cube_1"), 19L); + assertUpdate("DROP CUBE nation_insert_overwrite_cube_1"); + assertUpdate("DROP TABLE nation_table_cube_insert_overwrite_test_1"); + } + + @Test + public void testCreateCube() + { + computeActual("CREATE TABLE nation_table_create_cube_test_1 AS SELECT * FROM nation"); + assertQueryFails("CREATE CUBE nation ON nation " + + "WITH (AGGREGATIONS=(count(*))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])", "line 1:1: Table 'hive.tpch.nation' already exists"); + assertQueryFails("CREATE CUBE nation_create_cube_1 ON abcd " + + "WITH (AGGREGATIONS=(count(*), count(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])", "line 1:1: Table 'hive.tpch.abcd' does not exist"); + assertQueryFails("CREATE CUBE nation_create_cube_1 ON nation " + + "WITH (AGGREGATIONS=(sum(distinct nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])", "line 1:1: Distinct is currently only supported for count"); + assertUpdate("CREATE CUBE nation_create_cube_1 ON nation_table_create_cube_test_1 " + + "WITH (AGGREGATIONS=(count(*))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); + assertQueryFails("CREATE CUBE nation_create_cube_1 ON nation_table_create_cube_test_1 " + + "WITH (AGGREGATIONS=(count(*), count(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])", "line 1:1: Cube 'hive.tpch.nation_create_cube_1' already exists"); + assertUpdate("DROP CUBE nation_create_cube_1"); + assertUpdate("DROP TABLE nation_table_create_cube_test_1"); + } + + @Test + public void testCubeStatusChange() + { + computeActual("CREATE TABLE nation_table_status_test AS SELECT * FROM nation"); + assertUpdate("CREATE CUBE nation_status_cube_1 ON nation_table_status_test " + + "WITH (AGGREGATIONS=(count(*), COUNT(distinct nationkey), count(distinct regionkey), avg(nationkey), count(regionkey), sum(regionkey)," + + " min(regionkey), max(regionkey), max(nationkey), min(nationkey))," + + " group=(nationkey), format= 'orc', partitioned_by = ARRAY['nationkey'])"); + MaterializedResult result = computeActual("SHOW CUBES FOR nation_table_status_test"); + MaterializedRow matchingRow = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_status_cube_1")).findFirst().orElse(null); + assertNotNull(matchingRow); + assertEquals(matchingRow.getField(2), "Inactive"); + + assertUpdate("INSERT INTO CUBE nation_status_cube_1 where nationkey > 5", 19); + result = computeActual("SHOW CUBES FOR nation_table_status_test"); + matchingRow = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_status_cube_1")).findFirst().orElse(null); + assertNotNull(matchingRow); + assertEquals(matchingRow.getField(2), "Active"); + + assertUpdate("INSERT INTO nation_table_status_test VALUES (12345, 'name', 54321, 'comment')", 1); + result = computeActual("SHOW CUBES FOR nation_table_status_test"); + matchingRow = result.getMaterializedRows().stream().filter(row -> row.getField(0).toString().contains("nation_status_cube_1")).findFirst().orElse(null); + assertNotNull(matchingRow); + assertEquals(matchingRow.getField(2), "Expired"); + assertUpdate("DROP CUBE nation_status_cube_1"); + assertUpdate("DROP TABLE nation_table_status_test"); } } diff --git a/src/main/resource/license/license-header-alternate-2020.txt b/src/main/resource/license/license-header-alternate-2020.txt new file mode 100644 index 000000000..0de8ccbbc --- /dev/null +++ b/src/main/resource/license/license-header-alternate-2020.txt @@ -0,0 +1,12 @@ +Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. \ No newline at end of file diff --git a/src/main/resource/license/license-header.txt b/src/main/resource/license/license-header.txt index 0de8ccbbc..6a498fa42 100644 --- a/src/main/resource/license/license-header.txt +++ b/src/main/resource/license/license-header.txt @@ -1,4 +1,4 @@ -Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. +Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at